1301 lines
40 KiB
C++
1301 lines
40 KiB
C++
/*
|
|
* SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
|
*/
|
|
|
|
#include <clickhouse/client.h>
|
|
#include <memory>
|
|
#include <string>
|
|
#include <iostream>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <ctime>
|
|
#include <cmath>
|
|
#include <unordered_map>
|
|
#include "fmtlog.h"
|
|
|
|
namespace ch = clickhouse;
|
|
|
|
extern "C" { // For connect_to(), socket_send(), parse_database(), etc.
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <signal.h>
|
|
#include <string.h>
|
|
#include <unistd.h>
|
|
#include "database.h"
|
|
#include "utils.h"
|
|
#include "../T_defs.h"
|
|
#include "configuration.h"
|
|
}
|
|
|
|
const int MAX_QUERY_SIZE = 65536;
|
|
const char *DEFAULT_CLICKHOUSE_HOST = "localhost";
|
|
const int DEFAULT_CLICKHOUSE_PORT = 9000;
|
|
const char *DEFAULT_CLICKHOUSE_DB = "oai";
|
|
|
|
// Buffer multiple rows per event type and insert when:
|
|
// 1. Buffer reaches BATCH_SIZE rows, OR
|
|
// 2. INSERT_INTERVAL_MS milliseconds have elapsed since last insert
|
|
// This reduces insertion time from ~30ms/event to <1ms/event amortized
|
|
const size_t BATCH_SIZE = 500; // Number of rows to buffer before inserting
|
|
const uint64_t INSERT_INTERVAL_MS = 500; // Insert every 500ms
|
|
const size_t BUFFER_SIZE = 10 * 1024 * 1024; // 10MB
|
|
const size_t BUFFER_MARGIN = 183'456; // UE_PHY_PDSCH_IQ size
|
|
|
|
// 5G NR timing constants
|
|
const uint64_t NS_PER_SEC = 1000000000ULL;
|
|
const uint64_t FRAME_PERIOD_NS = 10000000ULL; // 10ms per frame
|
|
const int SFN_MAX = 1024;
|
|
// Duration of one full SFN cycle (1024 frames = 10.24 seconds)
|
|
const uint64_t SFN_CYCLE_NS = FRAME_PERIOD_NS * SFN_MAX;
|
|
|
|
// TAI to UTC offset in seconds (37 leap seconds as of 2017)
|
|
// This may need to be updated when new leap seconds are added
|
|
const int TAI_UTC_OFFSET_SEC = 37;
|
|
|
|
// GPS epoch is January 6, 1980 00:00:00 UTC
|
|
// Unix epoch is January 1, 1970 00:00:00 UTC
|
|
// Offset = 315964800 seconds
|
|
const uint64_t UNIX_TO_GPS_OFFSET_SEC = 315964800ULL;
|
|
const uint64_t UNIX_TO_GPS_OFFSET_NS = UNIX_TO_GPS_OFFSET_SEC * NS_PER_SEC;
|
|
const uint8_t UTC_TO_GPS_OFFSET_SEC = 18; // As of 2017, GPS time is 18 seconds ahead of UTC time.
|
|
const uint64_t UTC_TO_GPS_OFFSET_NS = UTC_TO_GPS_OFFSET_SEC * NS_PER_SEC;
|
|
|
|
/**
|
|
* Calculate duration between two timestamps in nanoseconds
|
|
* Works with both std::chrono::high_resolution_clock::time_point and uint64_t nanosecond timestamps
|
|
*/
|
|
#define DURATION_NS(start, end) (std::chrono::duration_cast<std::chrono::nanoseconds>((end) - (start)).count())
|
|
#define DURATION_US(start, end) (std::chrono::duration_cast<std::chrono::microseconds>((end) - (start)).count())
|
|
#define DURATION_MS(start, end) (std::chrono::duration_cast<std::chrono::milliseconds>((end) - (start)).count())
|
|
|
|
/**
|
|
* Calculate duration between two uint64_t nanosecond timestamps
|
|
*/
|
|
#define DURATION_NS_U64(end_ns, start_ns) ((int64_t)(end_ns) - (int64_t)(start_ns))
|
|
/**
|
|
* Get slot duration in nanoseconds for a given numerology
|
|
* mu=0 (15kHz): 1ms, mu=1 (30kHz): 0.5ms, mu=2 (60kHz): 0.25ms, etc.
|
|
*/
|
|
inline uint64_t get_slot_duration_ns(int mu)
|
|
{
|
|
static const uint64_t slot_durations_ns[] = {
|
|
1000000, // mu=0: 1ms
|
|
500000, // mu=1: 0.5ms
|
|
250000, // mu=2: 0.25ms
|
|
125000, // mu=3: 0.125ms
|
|
62500 // mu=4: 0.0625ms
|
|
};
|
|
if (mu < 0 || mu > 4) {
|
|
loge("Invalid numerology mu={}, using mu=0", mu);
|
|
return slot_durations_ns[0];
|
|
}
|
|
return slot_durations_ns[mu];
|
|
}
|
|
|
|
/**
|
|
* Get number of slots per frame for a given numerology
|
|
*/
|
|
inline int get_slots_per_frame(int mu)
|
|
{
|
|
static const int slots_per_frame[] = {10, 20, 40, 80, 160};
|
|
if (mu < 0 || mu > 4) {
|
|
loge("Invalid numerology mu={}, using mu=0", mu);
|
|
return slots_per_frame[0];
|
|
}
|
|
return slots_per_frame[mu];
|
|
}
|
|
|
|
/**
|
|
* Convert Unix/Linux time to GPS time
|
|
*
|
|
* @param unix_time_ns Unix time in nanoseconds since Unix epoch (Jan 1, 1970)
|
|
* @return GPS time in nanoseconds since GPS epoch (Jan 6, 1980)
|
|
*/
|
|
inline uint64_t unix_to_gps_ns(uint64_t unix_time_ns)
|
|
{
|
|
return unix_time_ns - UNIX_TO_GPS_OFFSET_NS + UTC_TO_GPS_OFFSET_NS;
|
|
}
|
|
|
|
/**
|
|
* Convert GPS time to Unix/Linux time
|
|
*
|
|
* @param gps_time_ns GPS time in nanoseconds since GPS epoch
|
|
* @return GPS time in nanoseconds since Unix epoch
|
|
*/
|
|
inline uint64_t gps_to_posix_ns(uint64_t gps_time_ns)
|
|
{
|
|
return gps_time_ns + UNIX_TO_GPS_OFFSET_NS;
|
|
}
|
|
|
|
/**
|
|
* Convert GPS time to Unix/Linux time
|
|
*
|
|
* @param gps_time_ns GPS time in nanoseconds since GPS epoch
|
|
* @return UTC time in nanoseconds since Unix epoch
|
|
*/
|
|
inline uint64_t gps_to_utc_ns(uint64_t gps_time_ns)
|
|
{
|
|
return gps_to_posix_ns(gps_time_ns) - UTC_TO_GPS_OFFSET_NS;
|
|
}
|
|
|
|
/**
|
|
* Calculate the absolute GPS timestamp for a given SFN and slot, given a close time reference in Unix time.
|
|
*
|
|
* @param sfn System Frame Number (0-1023)
|
|
* @param slot Slot number within the frame (depends on numerology)
|
|
* @param mu Numerology (0-4): 0=15kHz, 1=30kHz, 2=60kHz, 3=120kHz, 4=240kHz
|
|
* @param reference_utc_ns Reference UTC time in nanoseconds since Unix epoch
|
|
* @return UTC timestamp in nanoseconds when the SFN/slot was transmitted
|
|
*/
|
|
uint64_t calculate_sfn_slot_gps_time(int sfn, int slot, int mu, uint64_t reference_unix_ns)
|
|
{
|
|
// Validate inputs
|
|
if (sfn < 0 || sfn >= SFN_MAX) {
|
|
loge("Invalid SFN={} (must be 0-1023)", sfn);
|
|
sfn = sfn % SFN_MAX;
|
|
return 0;
|
|
}
|
|
|
|
int slots_per_frame = get_slots_per_frame(mu);
|
|
if (slot < 0 || slot >= slots_per_frame) {
|
|
loge("Invalid slot={} for mu={} (must be 0-{})", slot, mu, slots_per_frame - 1);
|
|
slot = slot % slots_per_frame;
|
|
}
|
|
// Nanoseconds since GPS epoch start, should be slightly larger than actual transmit time.
|
|
uint64_t reference_gps_ns = unix_to_gps_ns(reference_unix_ns);
|
|
|
|
// Calculate the offset within one SFN cycle for the given SFN/slot
|
|
uint64_t slot_duration_ns = get_slot_duration_ns(mu);
|
|
uint64_t offset_in_cycle_ns = (sfn * FRAME_PERIOD_NS) + (slot * slot_duration_ns);
|
|
|
|
// Find the number of complete SFN cycles since GPS epoch
|
|
// We'll use the reference time to determine which SFN cycle we're in
|
|
uint64_t cycles_since_epoch = reference_gps_ns / SFN_CYCLE_NS;
|
|
|
|
// Calculate the base time for the current SFN cycle
|
|
uint64_t current_cycle_base_ns = cycles_since_epoch * SFN_CYCLE_NS;
|
|
|
|
// Calculate the target time within this cycle
|
|
uint64_t gps_time_ns = current_cycle_base_ns + offset_in_cycle_ns;
|
|
|
|
// Check if the GPS time is too far in the future (more than 1 second)
|
|
// If so, it probably occurred in the previous cycle
|
|
if (gps_time_ns > reference_gps_ns + NS_PER_SEC) {
|
|
gps_time_ns -= SFN_CYCLE_NS;
|
|
}
|
|
// Check if the target is too far in the past (more than 9.24 seconds)
|
|
// If so, it probably occurs in the next cycle
|
|
if (gps_time_ns < reference_gps_ns - (SFN_CYCLE_NS - NS_PER_SEC)) {
|
|
gps_time_ns += SFN_CYCLE_NS;
|
|
}
|
|
return gps_time_ns;
|
|
}
|
|
|
|
void calculate_sfn_slot_from_gps(uint64_t gps_time_ns, int mu, int *sfn, int *slot)
|
|
{
|
|
uint64_t phase_ns = gps_time_ns % SFN_CYCLE_NS;
|
|
|
|
*sfn = phase_ns / FRAME_PERIOD_NS; // Truncate to integer
|
|
uint64_t position_in_frame_ns = phase_ns % FRAME_PERIOD_NS; // Remainder
|
|
|
|
uint64_t slot_duration_ns = get_slot_duration_ns(mu);
|
|
*slot = position_in_frame_ns / slot_duration_ns;
|
|
|
|
if (*sfn >= SFN_MAX)
|
|
*sfn = SFN_MAX - 1;
|
|
int max_slot = get_slots_per_frame(mu) - 1;
|
|
if (*slot > max_slot)
|
|
*slot = max_slot;
|
|
}
|
|
|
|
void calculate_sfn_slot_from_unix(uint64_t unix_time_ns, int mu, int *sfn, int *slot)
|
|
{
|
|
uint64_t gps_time_ns = unix_to_gps_ns(unix_time_ns);
|
|
calculate_sfn_slot_from_gps(gps_time_ns, mu, sfn, slot);
|
|
}
|
|
|
|
void usage(void)
|
|
{
|
|
printf(
|
|
"options:\n"
|
|
" -d <database file> this option is mandatory\n"
|
|
" -on <GROUP or ID> turn log ON for given GROUP or ID\n"
|
|
" -off <GROUP or ID> turn log OFF for given GROUP or ID\n"
|
|
" -ON turn all logs ON\n"
|
|
" -OFF turn all logs OFF\n"
|
|
" note: you may pass several -on/-off/-ON/-OFF,\n"
|
|
" they will be processed in order\n"
|
|
" by default, all is off\n"
|
|
" -ip <host> connect to given IP address (default %s)\n"
|
|
" -p <port> connect to given port (default %d)\n"
|
|
" -ch-host <host> ClickHouse host (default %s)\n"
|
|
" -ch-port <port> ClickHouse native port (default %d)\n"
|
|
" -ch-db <database> ClickHouse database name (default %s)\n"
|
|
" -ch-user <user> ClickHouse user (optional)\n"
|
|
" -ch-pass <password> ClickHouse password (optional)\n"
|
|
" -ch-droptables drop the ClickHouse database before run\n"
|
|
" -debug enable debug logging\n",
|
|
DEFAULT_REMOTE_IP,
|
|
DEFAULT_REMOTE_PORT,
|
|
DEFAULT_CLICKHOUSE_HOST,
|
|
DEFAULT_CLICKHOUSE_PORT,
|
|
DEFAULT_CLICKHOUSE_DB);
|
|
exit(1);
|
|
}
|
|
|
|
volatile int run = 1;
|
|
|
|
static int socket_fd = -1;
|
|
|
|
void force_stop(int x)
|
|
{
|
|
logi("gently quit %d", x);
|
|
close(socket_fd);
|
|
socket_fd = -1;
|
|
run = 0;
|
|
}
|
|
|
|
struct clickhouse_config_t {
|
|
std::string host;
|
|
int port;
|
|
std::string database;
|
|
std::string user;
|
|
std::string password;
|
|
std::unique_ptr<ch::Client> client;
|
|
int drop_tables;
|
|
};
|
|
|
|
// Buffer management structures
|
|
struct DataBuffer {
|
|
uint8_t *buffer; // Large preallocated buffer
|
|
size_t buffer_size; // Total size of buffer
|
|
size_t current_offset; // Current write position in buffer
|
|
std::shared_ptr<ch::ColumnUInt8> inner_col; // Reusable inner column for ClickHouse
|
|
std::shared_ptr<ch::ColumnUInt64> offsets_col; // Reusable offsets column for ClickHouse (cumulative end offsets)
|
|
bool insert_now;
|
|
|
|
DataBuffer(size_t size) : buffer_size(size - BUFFER_MARGIN), current_offset(0), insert_now(false)
|
|
{
|
|
buffer = (uint8_t *)malloc(size);
|
|
if (!buffer) {
|
|
throw std::bad_alloc();
|
|
}
|
|
|
|
// Create and reserve ClickHouse columns for reuse
|
|
inner_col = std::make_shared<ch::ColumnUInt8>();
|
|
auto &inner_data = inner_col->GetWritableData();
|
|
inner_data.reserve(size);
|
|
|
|
offsets_col = std::make_shared<ch::ColumnUInt64>();
|
|
offsets_col->Reserve(BATCH_SIZE);
|
|
}
|
|
|
|
~DataBuffer()
|
|
{
|
|
if (buffer) {
|
|
free(buffer);
|
|
buffer = nullptr;
|
|
}
|
|
}
|
|
|
|
// Disable copy constructor and assignment operator
|
|
DataBuffer(const DataBuffer &) = delete;
|
|
DataBuffer &operator=(const DataBuffer &) = delete;
|
|
|
|
// Copy data into buffer and return offset
|
|
size_t append(const void *data, size_t length)
|
|
{
|
|
if (current_offset + length > buffer_size) {
|
|
insert_now = true; // If this puts the buffer into the margin, insert it immediately
|
|
if (length > BUFFER_MARGIN) {
|
|
loge("DataBuffer full: current_offset={} length={} buffer_size={}", current_offset, length, buffer_size);
|
|
return (size_t)-1;
|
|
} // Else we have margin
|
|
}
|
|
|
|
size_t offset = current_offset;
|
|
memcpy(buffer + current_offset, data, length);
|
|
current_offset += length;
|
|
|
|
offsets_col->Append(current_offset);
|
|
|
|
return offset;
|
|
}
|
|
|
|
// Get pointer to data at a specific offset
|
|
const uint8_t *get_data(size_t offset) const
|
|
{
|
|
if (offset >= current_offset) {
|
|
return nullptr;
|
|
}
|
|
return buffer + offset;
|
|
}
|
|
|
|
// Clear buffer for reuse
|
|
void clear()
|
|
{
|
|
current_offset = 0;
|
|
insert_now = false;
|
|
inner_col->Clear();
|
|
offsets_col->Clear();
|
|
}
|
|
|
|
// Get number of entries
|
|
size_t entry_count() const
|
|
{
|
|
return offsets_col->Size();
|
|
}
|
|
|
|
// Get current usage
|
|
size_t usage() const
|
|
{
|
|
return current_offset;
|
|
}
|
|
};
|
|
|
|
struct EventBuffer {
|
|
std::vector<ch::ColumnRef> columns; // Ordered list of columns
|
|
std::vector<std::string> column_names; // Column names in same order
|
|
std::string event_table_name;
|
|
size_t row_count;
|
|
bool insert_now;
|
|
std::chrono::steady_clock::time_point last_insert_time;
|
|
database_event_format format;
|
|
std::string event_name;
|
|
bool has_frame_slot;
|
|
std::unordered_map<int, std::unique_ptr<DataBuffer>> data_buffers;
|
|
|
|
// Cached typed column pointers to avoid expensive As<> casts on every insert
|
|
std::shared_ptr<ch::ColumnDateTime64> ts_sw_col;
|
|
std::shared_ptr<ch::ColumnDateTime64> ts_tx_col;
|
|
std::vector<std::shared_ptr<ch::ColumnInt32>> int_cols;
|
|
std::vector<std::shared_ptr<ch::ColumnUInt64>> ulong_cols;
|
|
std::vector<std::shared_ptr<ch::ColumnString>> string_cols;
|
|
std::unordered_map<int, int> field_idx_to_typed_col_idx; // Maps format field index to typed column index
|
|
|
|
// Performance metrics
|
|
uint64_t last_column_build_duration_ns;
|
|
uint64_t last_block_append_duration_ns;
|
|
uint64_t last_insertion_duration_ns;
|
|
|
|
// Statistics tracking
|
|
uint64_t total_rows; // Total rows inserted since start
|
|
uint64_t rows_since_print; // Rows since last print
|
|
|
|
EventBuffer()
|
|
: row_count(0),
|
|
insert_now(false),
|
|
last_insert_time(std::chrono::steady_clock::now()),
|
|
has_frame_slot(false),
|
|
last_column_build_duration_ns(0),
|
|
last_block_append_duration_ns(0),
|
|
last_insertion_duration_ns(0),
|
|
total_rows(0),
|
|
rows_since_print(0)
|
|
{
|
|
}
|
|
};
|
|
|
|
// Global buffer map: event_id -> EventBuffer
|
|
std::unordered_map<int, EventBuffer> g_event_buffers;
|
|
|
|
/* Convert T format type to ClickHouse type */
|
|
const char *get_clickhouse_type(const char *t_type)
|
|
{
|
|
if (strcmp(t_type, "int") == 0)
|
|
return "Int32";
|
|
if (strcmp(t_type, "ulong") == 0)
|
|
return "UInt64";
|
|
if (strcmp(t_type, "string") == 0)
|
|
return "String";
|
|
if (strcmp(t_type, "buffer") == 0)
|
|
return "Array(UInt8)"; // Store as byte array
|
|
if (strncmp(t_type, "int", 3) == 0)
|
|
return "Int32";
|
|
return "String"; // default
|
|
}
|
|
|
|
/* Execute ClickHouse query */
|
|
int clickhouse_execute(clickhouse_config_t *ch, const char *query)
|
|
{
|
|
try {
|
|
ch->client->Execute(query);
|
|
return 0;
|
|
} catch (const std::exception &e) {
|
|
loge("ClickHouse query failed: {}", e.what());
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
/* Initialize ClickHouse connection and create database */
|
|
int clickhouse_init(clickhouse_config_t *ch)
|
|
{
|
|
char query[1024];
|
|
|
|
try {
|
|
// Set connection options
|
|
ch::ClientOptions options;
|
|
options.SetHost(ch->host);
|
|
options.SetPort(ch->port);
|
|
|
|
if (!ch->user.empty()) {
|
|
options.SetUser(ch->user);
|
|
}
|
|
if (!ch->password.empty()) {
|
|
options.SetPassword(ch->password);
|
|
}
|
|
|
|
// Create client
|
|
ch->client = std::make_unique<ch::Client>(options);
|
|
/* Handle drop database option */
|
|
if (ch->drop_tables) {
|
|
logi("Dropping database: {}", ch->database.c_str());
|
|
snprintf(query, sizeof(query), "DROP DATABASE IF EXISTS %s", ch->database.c_str());
|
|
|
|
if (clickhouse_execute(ch, query) != 0) {
|
|
loge("Failed to drop database {}", ch->database);
|
|
exit(1);
|
|
}
|
|
}
|
|
|
|
// Create database if not exists
|
|
snprintf(query, sizeof(query), "CREATE DATABASE IF NOT EXISTS %s", ch->database.c_str());
|
|
if (clickhouse_execute(ch, query) != 0) {
|
|
loge("Failed to create database");
|
|
return -1;
|
|
}
|
|
|
|
logi("ClickHouse connection initialized");
|
|
return 0;
|
|
} catch (const std::exception &e) {
|
|
loge("Failed to create ClickHouse client: {}", e.what());
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
/* Create table schema from database format */
|
|
char *create_table_schema(char *schema_buf, const char *event_name, database_event_format format, const char *database)
|
|
{
|
|
char *p = schema_buf;
|
|
bool has_frame = false;
|
|
bool has_slot = false;
|
|
int i;
|
|
|
|
// Start CREATE TABLE statement
|
|
p += snprintf(p,
|
|
MAX_QUERY_SIZE - (p - schema_buf),
|
|
"CREATE TABLE IF NOT EXISTS %s.%s (\n TsSwNs DateTime64(9) DEFAULT now64(9)",
|
|
database,
|
|
event_name);
|
|
|
|
for (i = 0; i < format.count; i++) {
|
|
if (strcmp(format.name[i], "frame") == 0)
|
|
has_frame = true;
|
|
if (strcmp(format.name[i], "slot") == 0)
|
|
has_slot = true;
|
|
if (strcmp(format.name[i], "subframe") == 0)
|
|
has_slot = true;
|
|
}
|
|
if (has_frame && has_slot) {
|
|
p += snprintf(p, MAX_QUERY_SIZE - (p - schema_buf), ",\n TsTxUtcNs DateTime64(9)");
|
|
uint64_t timestamp_ns = std::chrono::system_clock::now().time_since_epoch().count() / NS_PER_SEC;
|
|
if (timestamp_ns > 1861920000) {
|
|
// 1861920000 -> 2029-01-01 00:00:00
|
|
logw("WARNING: Linux time {} is greater than 2029-01-01 00:00:00.", timestamp_ns);
|
|
logw(" Check if a negative leap second has been announced and change UTC_TO_GPS_OFFSET_NS accordingly.",
|
|
timestamp_ns);
|
|
}
|
|
}
|
|
|
|
// Add columns from format
|
|
for (i = 0; i < format.count; i++) {
|
|
const char *ch_type = get_clickhouse_type(format.type[i]);
|
|
p += snprintf(p, MAX_QUERY_SIZE - (p - schema_buf), ",\n %s %s", format.name[i], ch_type);
|
|
}
|
|
|
|
p += snprintf(p, MAX_QUERY_SIZE - (p - schema_buf), "\n) ENGINE = Memory();");
|
|
|
|
#ifdef DB_DEBUG
|
|
logd("schema: {}", schema_buf);
|
|
#endif
|
|
return schema_buf;
|
|
}
|
|
|
|
/* Initialize buffer for an event */
|
|
void init_event_buffer(int event_id, void *database, const char *database_name)
|
|
{
|
|
char *event_name = event_name_from_id(database, event_id);
|
|
database_event_format format = get_format(database, event_id);
|
|
|
|
if (!event_name)
|
|
return;
|
|
|
|
EventBuffer &buf = g_event_buffers[event_id];
|
|
buf.event_name = std::string(event_name);
|
|
buf.event_table_name = std::string(database_name) + "." + event_name;
|
|
buf.format = format;
|
|
buf.row_count = 0;
|
|
buf.last_insert_time = std::chrono::steady_clock::now();
|
|
|
|
// Check if this event has frame/slot fields
|
|
buf.has_frame_slot = false;
|
|
bool has_frame = false;
|
|
bool has_slot = false;
|
|
for (int i = 0; i < format.count; i++) {
|
|
if (strcmp(format.name[i], "frame") == 0)
|
|
has_frame = true;
|
|
if (strcmp(format.name[i], "slot") == 0 || strcmp(format.name[i], "subframe") == 0)
|
|
has_slot = true;
|
|
}
|
|
buf.has_frame_slot = has_frame && has_slot;
|
|
|
|
// Create TsSwNs column
|
|
auto ts_col = std::make_shared<ch::ColumnDateTime64>(9);
|
|
ts_col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(ts_col);
|
|
buf.column_names.push_back("TsSwNs");
|
|
buf.ts_sw_col = ts_col; // Cache typed pointer
|
|
|
|
// Add data columns
|
|
for (int i = 0; i < format.count; i++) {
|
|
if (strcmp(format.type[i], "int") == 0) {
|
|
auto col = std::make_shared<ch::ColumnInt32>();
|
|
col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(col);
|
|
buf.column_names.push_back(format.name[i]);
|
|
buf.field_idx_to_typed_col_idx[i] = buf.int_cols.size();
|
|
buf.int_cols.push_back(col);
|
|
} else if (strcmp(format.type[i], "ulong") == 0) {
|
|
auto col = std::make_shared<ch::ColumnUInt64>();
|
|
col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(col);
|
|
buf.column_names.push_back(format.name[i]);
|
|
buf.field_idx_to_typed_col_idx[i] = buf.ulong_cols.size();
|
|
buf.ulong_cols.push_back(col);
|
|
} else if (strcmp(format.type[i], "buffer") == 0) {
|
|
buf.data_buffers[i] = std::make_unique<DataBuffer>(BUFFER_SIZE);
|
|
|
|
// Placeholder - array column will be created from DataBuffer at flush time
|
|
buf.columns.push_back(nullptr);
|
|
buf.column_names.push_back(format.name[i]);
|
|
|
|
} else if (strcmp(format.type[i], "string") == 0) {
|
|
auto col = std::make_shared<ch::ColumnString>();
|
|
col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(col);
|
|
buf.column_names.push_back(format.name[i]);
|
|
buf.field_idx_to_typed_col_idx[i] = buf.string_cols.size();
|
|
buf.string_cols.push_back(col);
|
|
} else {
|
|
// Default: treat as int
|
|
auto col = std::make_shared<ch::ColumnInt32>();
|
|
col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(col);
|
|
buf.column_names.push_back(format.name[i]);
|
|
buf.field_idx_to_typed_col_idx[i] = buf.int_cols.size();
|
|
buf.int_cols.push_back(col);
|
|
}
|
|
}
|
|
|
|
// Add TsTxUtcNs column if event has frame/slot
|
|
if (buf.has_frame_slot) {
|
|
auto gps_col = std::make_shared<ch::ColumnDateTime64>(9);
|
|
gps_col->Reserve(BATCH_SIZE);
|
|
buf.columns.push_back(gps_col);
|
|
buf.column_names.push_back("TsTxUtcNs");
|
|
buf.ts_tx_col = gps_col; // Cache typed pointer
|
|
}
|
|
|
|
logi("Initialized buffer for event_id:{:<4} name:{:<31} has_frame_slot:{} columns:{}",
|
|
event_id,
|
|
event_name,
|
|
buf.has_frame_slot,
|
|
buf.columns.size());
|
|
|
|
// Print sizes of buffer fields
|
|
for (int i = 0; i < format.count; i++) {
|
|
if (strcmp(format.type[i], "buffer") == 0) {
|
|
logd(" Buffer field '{}' size={} MB", format.name[i], BUFFER_SIZE / (1024 * 1024));
|
|
}
|
|
}
|
|
}
|
|
|
|
/* Create tables for enabled events */
|
|
int create_tables(clickhouse_config_t *ch, void *database, int *is_on, int number_of_events)
|
|
{
|
|
int i;
|
|
size_t longest_table_name = 0;
|
|
|
|
for (i = 0; i < number_of_events; i++) {
|
|
if (!is_on[i])
|
|
continue;
|
|
|
|
char *event_name = event_name_from_id(database, i);
|
|
database_event_format format = get_format(database, i);
|
|
if (strlen(event_name) > longest_table_name)
|
|
longest_table_name = strlen(event_name);
|
|
|
|
if (!event_name)
|
|
continue;
|
|
|
|
logd("Creating table for event: {:<31}", event_name);
|
|
|
|
char schema[MAX_QUERY_SIZE];
|
|
create_table_schema(schema, event_name, format, ch->database.c_str());
|
|
|
|
if (clickhouse_execute(ch, schema) != 0) {
|
|
loge("Failed to create table {}", event_name);
|
|
return -1;
|
|
}
|
|
|
|
// Initialize buffer for this event
|
|
init_event_buffer(i, database, ch->database.c_str());
|
|
}
|
|
|
|
return (int)longest_table_name;
|
|
}
|
|
|
|
int insert_buffer(clickhouse_config_t *ch, int event_id)
|
|
{
|
|
auto it = g_event_buffers.find(event_id);
|
|
if (it == g_event_buffers.end() || it->second.row_count == 0) {
|
|
return 0; // Nothing to flush
|
|
}
|
|
|
|
EventBuffer &buf = it->second;
|
|
buf.insert_now = false;
|
|
|
|
auto t_reconstruct_start = std::chrono::high_resolution_clock::now();
|
|
size_t total_data_size = 0;
|
|
for (auto &dbuf_pair : buf.data_buffers) {
|
|
int field_idx = dbuf_pair.first;
|
|
DataBuffer *dbuf = dbuf_pair.second.get();
|
|
int col_idx = field_idx + 1; // +1 because TsSwNs is at index 0
|
|
|
|
// Build the array column from the DataBuffer
|
|
size_t num_entries = dbuf->offsets_col->Size();
|
|
|
|
if (num_entries != buf.row_count) {
|
|
loge("DataBuffer entry count mismatch for field '{}': {} entries vs {} rows",
|
|
buf.format.name[field_idx],
|
|
num_entries,
|
|
buf.row_count);
|
|
}
|
|
|
|
// Copy all data to the DataBuffer's inner column (space already reserved in constructor)
|
|
if (dbuf->current_offset > 0) {
|
|
auto &inner_data = dbuf->inner_col->GetWritableData();
|
|
inner_data.resize(dbuf->current_offset); // Set size (doesn't reallocate since already reserved)
|
|
memcpy(inner_data.data(), dbuf->buffer, dbuf->current_offset);
|
|
}
|
|
|
|
// Create the array column from DataBuffer's columns and store it
|
|
auto array_col = std::make_shared<ch::ColumnArrayT<ch::ColumnUInt8>>(dbuf->inner_col, dbuf->offsets_col);
|
|
buf.columns[col_idx] = array_col;
|
|
|
|
total_data_size += dbuf->current_offset;
|
|
}
|
|
auto t_reconstruct_end = std::chrono::high_resolution_clock::now();
|
|
|
|
ch::Block block;
|
|
for (size_t i = 0; i < buf.columns.size(); i++) {
|
|
if (buf.columns[i]) {
|
|
block.AppendColumn(buf.column_names[i], buf.columns[i]);
|
|
} else {
|
|
loge("Column '{}' at index {} is null when creating block", buf.column_names[i], i);
|
|
exit(1);
|
|
}
|
|
}
|
|
try {
|
|
auto t_insert_start = std::chrono::high_resolution_clock::now();
|
|
ch->client->Insert(buf.event_table_name.c_str(), block);
|
|
auto t_insert_end = std::chrono::high_resolution_clock::now();
|
|
|
|
buf.last_column_build_duration_ns = DURATION_NS(t_reconstruct_start, t_reconstruct_end);
|
|
buf.last_block_append_duration_ns = DURATION_NS(t_reconstruct_end, t_insert_start);
|
|
buf.last_insertion_duration_ns = DURATION_NS(t_insert_start, t_insert_end);
|
|
|
|
// Clear all columns and re-reserve (skip buffer-type columns which are rebuilt each time)
|
|
for (size_t i = 0; i < buf.columns.size(); i++) {
|
|
if (buf.columns[i]) { // Skip nullptr placeholders for buffer-type columns
|
|
auto col = std::const_pointer_cast<ch::Column>(buf.columns[i]);
|
|
col->Clear();
|
|
col->Reserve(BATCH_SIZE);
|
|
}
|
|
}
|
|
|
|
// Reset buffer-type columns back to nullptr
|
|
for (auto &dbuf_pair : buf.data_buffers) {
|
|
int field_idx = dbuf_pair.first;
|
|
int col_idx = field_idx + 1;
|
|
buf.columns[col_idx] = nullptr;
|
|
}
|
|
|
|
// Clear all DataBuffers
|
|
for (auto &pair : buf.data_buffers) {
|
|
pair.second->clear();
|
|
}
|
|
|
|
int rows = buf.row_count;
|
|
buf.row_count = 0;
|
|
buf.last_insert_time = std::chrono::steady_clock::now();
|
|
|
|
// Update statistics
|
|
buf.total_rows += rows;
|
|
buf.rows_since_print += rows;
|
|
|
|
return rows;
|
|
} catch (const std::exception &e) {
|
|
loge("Failed to insert buffer for event {}: {}", buf.event_name, e.what());
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
/* Flush all buffers to ClickHouse */
|
|
int insert_all_buffers(clickhouse_config_t *ch)
|
|
{
|
|
int errors = 0;
|
|
for (auto &pair : g_event_buffers) {
|
|
if (insert_buffer(ch, pair.first) <= 0) {
|
|
errors++;
|
|
}
|
|
}
|
|
return errors > 0 ? -1 : 0;
|
|
}
|
|
|
|
/* Check if buffer should be inserted based on size or time */
|
|
uint8_t should_insert(EventBuffer &buf)
|
|
{
|
|
if (buf.row_count == 0) {
|
|
return 0;
|
|
}
|
|
|
|
if (buf.row_count >= BATCH_SIZE) {
|
|
return 1;
|
|
}
|
|
|
|
if (static_cast<uint64_t>(DURATION_MS(buf.last_insert_time, std::chrono::steady_clock::now())) >= INSERT_INTERVAL_MS) {
|
|
return 2;
|
|
}
|
|
|
|
if (buf.insert_now) {
|
|
return 3;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
int insert_event(clickhouse_config_t *ch, int event_id, const char *event_data, int data_length)
|
|
{
|
|
auto t_fn_start = std::chrono::high_resolution_clock::now();
|
|
|
|
// Get buffer for this event
|
|
auto it = g_event_buffers.find(event_id);
|
|
if (it == g_event_buffers.end()) {
|
|
loge("Buffer not found for event_id {}", event_id);
|
|
return -1;
|
|
}
|
|
|
|
EventBuffer &buf = it->second;
|
|
database_event_format format = buf.format;
|
|
const char *event_name = buf.event_name.c_str();
|
|
auto t_names_done = std::chrono::high_resolution_clock::now();
|
|
|
|
int16_t frame = -1;
|
|
int16_t slot = -1;
|
|
|
|
try {
|
|
const char *data_ptr = event_data;
|
|
int remaining = data_length;
|
|
int buf_len = 0;
|
|
int i;
|
|
|
|
auto t_append_start = std::chrono::high_resolution_clock::now();
|
|
|
|
// Append to TsSwNs column using cached pointer
|
|
buf.ts_sw_col->Append(std::chrono::system_clock::now().time_since_epoch().count());
|
|
|
|
// Process data columns
|
|
for (i = 0; i < format.count && remaining > 0; i++) {
|
|
if (strcmp(format.type[i], "int") == 0) {
|
|
int val = 0;
|
|
if (remaining >= 4) {
|
|
memcpy(&val, data_ptr, 4);
|
|
if (strcmp(format.name[i], "frame") == 0)
|
|
frame = val;
|
|
else if (strcmp(format.name[i], "slot") == 0)
|
|
slot = val;
|
|
else if (strcmp(format.name[i], "subframe") == 0)
|
|
slot = val;
|
|
data_ptr += 4;
|
|
remaining -= 4;
|
|
}
|
|
|
|
int typed_col_idx = buf.field_idx_to_typed_col_idx[i];
|
|
buf.int_cols[typed_col_idx]->Append(val);
|
|
|
|
} else if (strcmp(format.type[i], "ulong") == 0) {
|
|
uint64_t val = 0;
|
|
if (remaining >= 8) {
|
|
memcpy(&val, data_ptr, 8);
|
|
data_ptr += 8;
|
|
remaining -= 8;
|
|
}
|
|
|
|
int typed_col_idx = buf.field_idx_to_typed_col_idx[i];
|
|
buf.ulong_cols[typed_col_idx]->Append(val);
|
|
|
|
} else if (strcmp(format.type[i], "buffer") == 0) {
|
|
buf_len = 0;
|
|
const char *buf_data = nullptr;
|
|
|
|
if (remaining >= 4) {
|
|
memcpy(&buf_len, data_ptr, 4);
|
|
data_ptr += 4;
|
|
remaining -= 4;
|
|
|
|
if (buf_len > remaining)
|
|
buf_len = remaining;
|
|
if (buf_len > 0)
|
|
buf_data = data_ptr;
|
|
|
|
data_ptr += buf_len;
|
|
remaining -= buf_len;
|
|
}
|
|
|
|
auto it_dbuf = buf.data_buffers.find(i);
|
|
if (it_dbuf != buf.data_buffers.end()) {
|
|
size_t offset = it_dbuf->second->append(buf_data, buf_len);
|
|
|
|
buf.insert_now |= it_dbuf->second->insert_now;
|
|
|
|
if (offset == (size_t)-1) {
|
|
loge("DataBuffer full for field '{}', cannot append {} bytes", format.name[i], buf_len);
|
|
// Handle dropping tail for other columns and insert
|
|
}
|
|
}
|
|
} else if (strcmp(format.type[i], "string") == 0) {
|
|
std::string str_val;
|
|
if (remaining >= 4) {
|
|
int str_len;
|
|
memcpy(&str_len, data_ptr, 4);
|
|
data_ptr += 4;
|
|
remaining -= 4;
|
|
|
|
if (str_len > remaining)
|
|
str_len = remaining;
|
|
if (str_len > 0)
|
|
str_val = std::string(data_ptr, str_len);
|
|
|
|
data_ptr += str_len;
|
|
remaining -= str_len;
|
|
}
|
|
|
|
int typed_col_idx = buf.field_idx_to_typed_col_idx[i];
|
|
buf.string_cols[typed_col_idx]->Append(str_val);
|
|
|
|
} else {
|
|
// Default: treat as int
|
|
int val = 0;
|
|
if (remaining >= 4) {
|
|
memcpy(&val, data_ptr, 4);
|
|
data_ptr += 4;
|
|
remaining -= 4;
|
|
}
|
|
|
|
int typed_col_idx = buf.field_idx_to_typed_col_idx[i];
|
|
buf.int_cols[typed_col_idx]->Append(val);
|
|
}
|
|
}
|
|
auto t_append_data_end = std::chrono::high_resolution_clock::now();
|
|
|
|
// Add TsTxUtcNs column if event has frame/slot
|
|
if (buf.has_frame_slot) {
|
|
uint64_t timestamp_ns = std::chrono::system_clock::now().time_since_epoch().count();
|
|
uint64_t gps_elapsed_ns;
|
|
uint64_t transmit_time_in_utc_ns;
|
|
auto t_time_calc_start = std::chrono::high_resolution_clock::now();
|
|
|
|
if (frame == -1 || slot == -1) { // nr_ue_decode_BCCH_DL_SCH sets frame and slot to -1
|
|
transmit_time_in_utc_ns = timestamp_ns;
|
|
} else {
|
|
gps_elapsed_ns = calculate_sfn_slot_gps_time(frame, slot, 1, timestamp_ns);
|
|
transmit_time_in_utc_ns = gps_to_utc_ns(gps_elapsed_ns);
|
|
|
|
if (0) {
|
|
int check_frame, check_slot;
|
|
calculate_sfn_slot_from_unix(transmit_time_in_utc_ns, 1, &check_frame, &check_slot);
|
|
if (check_frame != frame || check_slot != slot) {
|
|
loge("ERROR1: sfn/slot check failed: {} {} {} {}", event_name, frame, slot, check_frame, check_slot);
|
|
}
|
|
}
|
|
|
|
auto t_time_calc_end = std::chrono::high_resolution_clock::now();
|
|
|
|
int64_t ref_to_t0_offset_ns = (int64_t)timestamp_ns - (int64_t)transmit_time_in_utc_ns;
|
|
// Handle sign separately to properly display negative fractional seconds
|
|
const char *sign = (ref_to_t0_offset_ns < 0) ? "-" : " ";
|
|
int64_t abs_offset = (ref_to_t0_offset_ns < 0) ? -ref_to_t0_offset_ns : ref_to_t0_offset_ns;
|
|
int64_t offset_sec = abs_offset / NS_PER_SEC;
|
|
int64_t offset_ns = abs_offset % NS_PER_SEC;
|
|
|
|
if (abs_offset > 50'000'000) { // 50ms
|
|
logi("{:<31} {:4d}.{:2d} TsSwNs={}.{:09} TsTxUtcNs={}.{:09} delta={}{}.{:09d}, {}",
|
|
event_name,
|
|
frame,
|
|
slot,
|
|
timestamp_ns / NS_PER_SEC,
|
|
timestamp_ns % NS_PER_SEC,
|
|
transmit_time_in_utc_ns / NS_PER_SEC,
|
|
transmit_time_in_utc_ns % NS_PER_SEC,
|
|
sign,
|
|
offset_sec,
|
|
offset_ns,
|
|
DURATION_NS(t_time_calc_start, t_time_calc_end));
|
|
}
|
|
}
|
|
|
|
// Append to TsTxUtcNs column using cached pointer
|
|
buf.ts_tx_col->Append(transmit_time_in_utc_ns);
|
|
}
|
|
auto t_append_ts_end = std::chrono::high_resolution_clock::now();
|
|
|
|
// Increment row count
|
|
buf.row_count++;
|
|
|
|
// Check if we should insert this buffer
|
|
int rows = 0;
|
|
auto t_insert_start = std::chrono::high_resolution_clock::now();
|
|
uint8_t insert = should_insert(buf);
|
|
auto t_insert_should_insert_end = std::chrono::high_resolution_clock::now();
|
|
if (insert) {
|
|
rows = insert_buffer(ch, event_id);
|
|
}
|
|
auto t_insert_end = std::chrono::high_resolution_clock::now();
|
|
|
|
if (insert) {
|
|
if (DURATION_MS(t_insert_start, t_insert_end) > 5) {
|
|
logi("Slow Insert: {:<31} {:4d}.{:2d}: {:3d} {:3d} {:5d} {:4d} {:5d} {:4d} {:8d} (ns) {:3d} (ms) rows:{:3d} reason: {}",
|
|
event_name,
|
|
frame,
|
|
slot,
|
|
DURATION_NS(t_fn_start, t_names_done),
|
|
DURATION_NS(t_names_done, t_append_start),
|
|
DURATION_NS(t_append_start, t_append_data_end),
|
|
DURATION_NS(t_append_data_end, t_append_ts_end),
|
|
buf.last_column_build_duration_ns,
|
|
buf.last_block_append_duration_ns,
|
|
buf.last_insertion_duration_ns,
|
|
DURATION_MS(t_insert_start, t_insert_end),
|
|
rows,
|
|
insert);
|
|
}
|
|
} else if (DURATION_US(t_fn_start, t_insert_end) > 50) {
|
|
logd("Slow saving data: {:<31} {:4d}.{:2d}: {:3d} {:3d} {:5d} {:4d} {:5d} {:4d} {:8d} (ns) buf_len:{:3d} rows:{:3d} ---",
|
|
event_name,
|
|
frame,
|
|
slot,
|
|
DURATION_NS(t_fn_start, t_names_done),
|
|
DURATION_NS(t_names_done, t_append_start),
|
|
DURATION_NS(t_append_start, t_append_data_end),
|
|
DURATION_NS(t_append_data_end, t_append_ts_end),
|
|
DURATION_NS(t_append_ts_end, t_insert_start),
|
|
DURATION_NS(t_insert_start, t_insert_should_insert_end),
|
|
DURATION_NS(t_insert_should_insert_end, t_insert_end),
|
|
buf_len,
|
|
buf.row_count);
|
|
}
|
|
|
|
return 0;
|
|
} catch (const std::exception &e) {
|
|
loge("Failed to insert event {}: {}", event_name, e.what());
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
int main(int n, char **v)
|
|
{
|
|
const char *default_database_filename = "T_messages.txt";
|
|
char *database_filename = NULL;
|
|
void *database;
|
|
const char *ip = DEFAULT_REMOTE_IP;
|
|
int port = DEFAULT_REMOTE_PORT;
|
|
char **on_off_name;
|
|
int *on_off_action;
|
|
int on_off_n = 0;
|
|
int *is_on;
|
|
int number_of_events;
|
|
int i;
|
|
char mt;
|
|
int debug_mode = 0;
|
|
|
|
clickhouse_config_t ch_config;
|
|
ch_config.drop_tables = 0;
|
|
ch_config.host = DEFAULT_CLICKHOUSE_HOST;
|
|
ch_config.port = DEFAULT_CLICKHOUSE_PORT;
|
|
ch_config.database = DEFAULT_CLICKHOUSE_DB;
|
|
|
|
/* write on a socket fails if the other end is closed and we get SIGPIPE */
|
|
if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
|
|
abort();
|
|
|
|
on_off_name = (char **)malloc(n * sizeof(char *));
|
|
if (on_off_name == NULL)
|
|
abort();
|
|
on_off_action = (int *)malloc(n * sizeof(int));
|
|
if (on_off_action == NULL)
|
|
abort();
|
|
|
|
for (i = 1; i < n; i++) {
|
|
if (!strcmp(v[i], "-h") || !strcmp(v[i], "--help"))
|
|
usage();
|
|
if (!strcmp(v[i], "-d")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
database_filename = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ip")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ip = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-p")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
port = atoi(v[++i]);
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-host")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ch_config.host = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-port")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ch_config.port = atoi(v[++i]);
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-db")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ch_config.database = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-user")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ch_config.user = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-pass")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
ch_config.password = v[++i];
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ch-droptables")) {
|
|
ch_config.drop_tables = 1;
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-debug")) {
|
|
debug_mode = 1;
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-on")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
on_off_name[on_off_n] = v[++i];
|
|
on_off_action[on_off_n++] = 1;
|
|
logi("on: item: {}, on_off_name: {}, on_off_action: {}", v[i], on_off_name[on_off_n - 1], on_off_action[on_off_n - 1]);
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-off")) {
|
|
if (i > n - 2)
|
|
usage();
|
|
on_off_name[on_off_n] = v[++i];
|
|
on_off_action[on_off_n++] = 0;
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-ON")) {
|
|
on_off_name[on_off_n] = NULL;
|
|
on_off_action[on_off_n++] = 1;
|
|
continue;
|
|
}
|
|
if (!strcmp(v[i], "-OFF")) {
|
|
on_off_name[on_off_n] = NULL;
|
|
on_off_action[on_off_n++] = 0;
|
|
continue;
|
|
}
|
|
usage();
|
|
}
|
|
|
|
if (database_filename == NULL) {
|
|
logi("No database file provided, using default: {}", default_database_filename);
|
|
database_filename = strdup(default_database_filename);
|
|
}
|
|
|
|
database = parse_database(database_filename);
|
|
|
|
load_config_file(database_filename);
|
|
|
|
number_of_events = number_of_ids(database);
|
|
logi("Number of events: {}", number_of_events);
|
|
is_on = (int *)calloc(number_of_events, sizeof(int));
|
|
if (is_on == NULL) {
|
|
loge("aborting");
|
|
abort();
|
|
}
|
|
|
|
for (i = 0; i < on_off_n; i++) {
|
|
on_off(database, on_off_name[i], is_on, on_off_action[i]);
|
|
}
|
|
|
|
/* Initialize fmtlog for benchmarking */
|
|
fmtlog::setLogLevel(debug_mode ? fmtlog::DBG : fmtlog::INF);
|
|
fmtlog::setThreadName("main");
|
|
|
|
/* Initialize ClickHouse */
|
|
if (clickhouse_init(&ch_config) != 0) {
|
|
loge("Failed to initialize ClickHouse connection");
|
|
exit(1);
|
|
}
|
|
|
|
/* Create tables for enabled events */
|
|
int longest_table_name = create_tables(&ch_config, database, is_on, number_of_events);
|
|
if (longest_table_name < 0) {
|
|
loge("Failed to create tables");
|
|
exit(1);
|
|
}
|
|
|
|
socket_fd = connect_to(const_cast<char *>(ip), port);
|
|
|
|
/* activate selected traces */
|
|
mt = 1;
|
|
if (socket_send(socket_fd, &mt, 1) == -1 || socket_send(socket_fd, &number_of_events, sizeof(int)) == -1
|
|
|| socket_send(socket_fd, is_on, number_of_events * sizeof(int)) == -1)
|
|
abort();
|
|
|
|
/* exit on ctrl+c and ctrl+z */
|
|
if (signal(SIGQUIT, force_stop) == SIG_ERR)
|
|
abort();
|
|
if (signal(SIGINT, force_stop) == SIG_ERR)
|
|
abort();
|
|
if (signal(SIGTSTP, force_stop) == SIG_ERR)
|
|
abort();
|
|
|
|
OBUF ebuf = {.osize = 0, .omaxsize = 0, .obuf = NULL};
|
|
|
|
/* read messages */
|
|
logi("Tables created, waiting for messages...");
|
|
int poll_count = 0;
|
|
auto last_periodic_insert = std::chrono::steady_clock::now();
|
|
auto last_stats_print = std::chrono::steady_clock::now();
|
|
const uint64_t STATS_PRINT_INTERVAL_MS = 5000; // Print stats every 5 seconds
|
|
|
|
while (run) {
|
|
int type;
|
|
int32_t length;
|
|
char *v;
|
|
int vpos = 0;
|
|
|
|
// Poll fmtlog periodically to insert buffered logs
|
|
if (++poll_count % 10 == 0) {
|
|
fmtlog::poll();
|
|
}
|
|
|
|
// Periodic insert check (every 10 messages)
|
|
if (poll_count % 10 == 0) {
|
|
auto now = std::chrono::steady_clock::now();
|
|
auto elapsed_ms =
|
|
static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(now - last_periodic_insert).count());
|
|
|
|
if (elapsed_ms >= INSERT_INTERVAL_MS) {
|
|
for (auto &pair : g_event_buffers) {
|
|
if (should_insert(pair.second)) {
|
|
insert_buffer(&ch_config, pair.first);
|
|
}
|
|
}
|
|
last_periodic_insert = now;
|
|
}
|
|
if (static_cast<uint64_t>(DURATION_MS(last_stats_print, now)) >= STATS_PRINT_INTERVAL_MS) {
|
|
for (auto &pair : g_event_buffers) {
|
|
EventBuffer &buf = pair.second;
|
|
if (buf.rows_since_print > 0)
|
|
logi(" {:<31} rows: {:8d} total: {:10d}", buf.event_name, buf.rows_since_print, buf.total_rows);
|
|
buf.rows_since_print = 0;
|
|
}
|
|
last_stats_print = now;
|
|
logi("");
|
|
}
|
|
}
|
|
|
|
if (fullread(socket_fd, &length, 4) == -1)
|
|
goto read_error;
|
|
if (ebuf.omaxsize < length + 4) {
|
|
ebuf.omaxsize = (length + 4 + 65535) & ~65535;
|
|
ebuf.obuf = (char *)realloc(ebuf.obuf, ebuf.omaxsize);
|
|
if (ebuf.obuf == NULL) {
|
|
loge("out of memory");
|
|
exit(1);
|
|
}
|
|
}
|
|
v = ebuf.obuf;
|
|
memcpy(v + vpos, &length, 4);
|
|
vpos += 4;
|
|
#ifdef T_SEND_TIME
|
|
if (length < (int32_t)sizeof(struct timespec))
|
|
goto read_error;
|
|
if (fullread(socket_fd, v + vpos, sizeof(struct timespec)) == -1)
|
|
goto read_error;
|
|
// memcpy(v+vpos, &ts, sizeof(struct timespec));
|
|
vpos += sizeof(struct timespec);
|
|
length -= sizeof(struct timespec);
|
|
#endif
|
|
if (length < (int32_t)sizeof(int))
|
|
goto read_error;
|
|
if (fullread(socket_fd, &type, sizeof(int)) == -1)
|
|
goto read_error;
|
|
memcpy(v + vpos, &type, sizeof(int));
|
|
vpos += sizeof(int);
|
|
length -= sizeof(int);
|
|
if (fullread(socket_fd, v + vpos, length) == -1)
|
|
goto read_error;
|
|
|
|
if (type == -1) {
|
|
append_received_config_chunk(v + vpos, length);
|
|
} else if (type == -2) {
|
|
verify_config();
|
|
} else if (type >= 0 && type < number_of_events && is_on[type]) {
|
|
if (insert_event(&ch_config, type, v + vpos, length) != 0) {
|
|
loge("Failed to insert event type {}", type);
|
|
}
|
|
}
|
|
}
|
|
|
|
read_error:
|
|
// Flush all remaining buffers before exit
|
|
logi("Inserting all buffers before exit...");
|
|
insert_all_buffers(&ch_config);
|
|
|
|
// Print final statistics
|
|
size_t total_rows = 0;
|
|
for (const auto& pair : g_event_buffers) {
|
|
total_rows += pair.second.row_count;
|
|
}
|
|
logi("Final buffer state: {} events with {} uninserted rows",
|
|
g_event_buffers.size(), total_rows);
|
|
|
|
// Client cleanup is automatic via std::unique_ptr
|
|
logi("Exiting...");
|
|
|
|
// Flush and close fmtlog
|
|
fmtlog::poll();
|
|
fmtlog::closeLogFile();
|
|
|
|
return 0;
|
|
}
|