From 1d69eb0e1ef6cc6f4856f9e441a77f40ca5c25f5 Mon Sep 17 00:00:00 2001 From: Robert Schmidt Date: Mon, 16 Feb 2026 09:12:44 +0100 Subject: [PATCH] Add T tracer to record configured events to database This extends the functionality provided by record in that it allows collection of data into a Clickhouse database, allowing aggregation of data from gNB and UE for functions like channel estimation for PUSCH transmissions that did not decode successfully at the gNB. It is intended to be used in conjunction with Aerial Data Lake but is not limited to it. Data access is simpler than in !3462, but could be improved using the metadata approach from !3462. The integration uses logfmt for fast logging, and a clickhouse DB client, both integrated using CPM. This change set also adds a docker-compose file for a UE on 100MHz, to be used with Aerial configurations, and builds the tracer for Aerial and nrUE docker files. --- .../sa_gnb_aerial/docker-compose-ue.yaml | 61 + common/utils/T/T_messages.txt | 2 +- common/utils/T/tracer/CMakeLists.txt | 54 + common/utils/T/tracer/Makefile | 12 +- common/utils/T/tracer/record_db.cpp | 1324 +++++++++++++++++ docker/Dockerfile.base.ubuntu | 2 + .../Dockerfile.build.fhi72.native_arm.ubuntu | 3 +- docker/Dockerfile.build.fhi72.ubuntu | 3 +- docker/Dockerfile.build.ubuntu | 26 +- docker/Dockerfile.gNB.aerial.ubuntu | 14 +- docker/Dockerfile.nrORU.fhi72.ubuntu | 6 - docker/Dockerfile.nrUE.ubuntu | 11 + 12 files changed, 1492 insertions(+), 26 deletions(-) create mode 100644 ci-scripts/yaml_files/sa_gnb_aerial/docker-compose-ue.yaml create mode 100644 common/utils/T/tracer/record_db.cpp diff --git a/ci-scripts/yaml_files/sa_gnb_aerial/docker-compose-ue.yaml b/ci-scripts/yaml_files/sa_gnb_aerial/docker-compose-ue.yaml new file mode 100644 index 0000000000..15f2f177f4 --- /dev/null +++ b/ci-scripts/yaml_files/sa_gnb_aerial/docker-compose-ue.yaml @@ -0,0 +1,61 @@ +services: + oai-nr-ue: + image: ${REGISTRY-oaisoftwarealliance/}${NRUE_IMG:-oai-nr-ue}:${UE_TAG:-${TAG:-develop}} + logging: + options: + max-file: "1" + network_mode: host + container_name: oai-nr-ue + privileged: true + cpuset: 4,6,8,10,12,14,16,18,20,22,24,26,28 + cap_drop: + - ALL + cap_add: + - NET_ADMIN # for interface bringup + - NET_RAW # for ping + - SYS_NICE # for performance + - IPC_LOCK # for memory locking + - SYS_PTRACE # for debugging + devices: + - /dev/net/tun:/dev/net/tun + ulimits: + core: -1 # for core dumps + #entrypoint: /bin/bash -c "sleep infinity" + environment: + TZ: UTC + #DISPLAY: $DISPLAY #Uncomment for GUI + USE_ADDITIONAL_OPTIONS: | + --thread-pool 4,6,8,10,12,14,16,18,20,22,24,26 + --usrp-args type=x4xx,addr=192.168.10.2,clock_source=external,time_source=external + --ssb 1478 -r 273 --numerology 1 --band 78 -C 3750000000 + --num-dl-actors 20 + --ue-nb-ant-tx 1 + --ue-nb-ant-rx 1 + --ue-rxgain 25 + --ue-txgain 13 + --uicc0.imsi 001010000000003 + --ue-fo-compensation -A 90 + --usrp-tx-thread-config 1 + --log_config.global_log_options level,nocolor,utc_time + # Unused options: + # --T_stdout 2 + # --imscope + # --ssb 1478 -r 273 --numerology 1 --band 78 -C 3750000000 + # --usrp-tx-thread-config 1 + # --log_config.mac_log_level debug + # --log_config.nr_mac_log_level debug + # --usrp-tx-thread-config 1 -d + # --ssb 194 -r 51 --numerology 1 --band 78 -C 3750000000 #20MHz + # --ssb 236 -r 106 --numerology 1 --band 78 -C 3750000000 #40MHz + volumes: + - ../../conf_files/ue.sa.conf:/opt/oai-nr-ue/etc/nr-ue.conf + # UL throughput is worse with 256-qam table + - ../../../targets/PROJECTS/GENERIC-NR-5GC/CONF/uecap_ports1.xml:/opt/oai-nr-ue/uecap_ports1.xml + #For GUI + - ${HOME}/.Xauthority:/root/.Xauthority:rw + - /tmp/.X11-unix:/tmp/.X11-unix + healthcheck: + test: /bin/bash -c "pgrep nr-uesoftmodem" + interval: 10s + timeout: 5s + retries: 5 diff --git a/common/utils/T/T_messages.txt b/common/utils/T/T_messages.txt index d2405f9033..18e98be0b6 100644 --- a/common/utils/T/T_messages.txt +++ b/common/utils/T/T_messages.txt @@ -1404,7 +1404,7 @@ ID = UE_PHY_PDCCH_ENERGY ID = UE_PHY_PDSCH_IQ DESC = UE PDSCH received IQ data GROUP = ALL:PHY:GRAPHIC:HEAVY:UE - FORMAT = int,eNB_ID : int,frame : int,subframe : int,nb_rb : int,N_RB_UL : int,symbols_per_tti : buffer,pusch_comp + FORMAT = int,eNB_ID : int,frame : int,subframe : int,nb_rb : int,N_RB_UL : int,symbols_per_tti : buffer,pdsch_comp ID = UE_PHY_PDSCH_ENERGY DESC = UE PDSCH 1 energy and threshold GROUP = ALL:PHY:GRAPHIC:HEAVY:UE diff --git a/common/utils/T/tracer/CMakeLists.txt b/common/utils/T/tracer/CMakeLists.txt index e347552d31..d64ea840a9 100644 --- a/common/utils/T/tracer/CMakeLists.txt +++ b/common/utils/T/tracer/CMakeLists.txt @@ -37,11 +37,65 @@ add_subdirectory(filter) add_subdirectory(logger) add_subdirectory(view) +# Option to enable record_db tracer tool with ClickHouse database support +add_boolean_option(T_RECORD_DB OFF "Compile record_db tracer tool with ClickHouse support" OFF) + +if(T_RECORD_DB) + # Save current BUILD_SHARED_LIBS setting + set(SAVED_BUILD_SHARED_LIBS ${BUILD_SHARED_LIBS}) + + # Configure options for clickhouse-cpp + set(BUILD_SHARED_LIBS ON CACHE BOOL "" FORCE) + + # Save current compiler flags and disable -Werror for zstd + set(SAVED_CMAKE_C_FLAGS "${CMAKE_C_FLAGS}") + + # Check if system zstd library is available + find_library(ZSTD_LIBRARY NAMES zstd) + if(ZSTD_LIBRARY) + # Use system zstd library to avoid building legacy zstd code with warnings + set(WITH_SYSTEM_ZSTD ON CACHE BOOL "" FORCE) + else() + # Fall back to bundled zstd and suppress warnings from legacy code + # https://github.com/ClickHouse/clickhouse-cpp/blob/master/contrib/zstd/zstd/legacy/zstd_v01.c#L1935 + set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-maybe-uninitialized -Wno-uninitialized") + endif() + + CPMAddPackage( + NAME clickhouse-cpp + GITHUB_REPOSITORY ClickHouse/clickhouse-cpp + GIT_TAG master + ) + + # Restore original compiler flags and BUILD_SHARED_LIBS + set(CMAKE_C_FLAGS "${SAVED_CMAKE_C_FLAGS}") + set(BUILD_SHARED_LIBS ${SAVED_BUILD_SHARED_LIBS} CACHE BOOL "" FORCE) + + CPMAddPackage( + NAME fmtlog + GITHUB_REPOSITORY MengRao/fmtlog + GIT_TAG main + GIT_SHALLOW FALSE + ) + + add_executable(record_db record_db.cpp) + + target_include_directories(record_db PRIVATE ${fmtlog_SOURCE_DIR} ${fmtlog_SOURCE_DIR}/fmt/include) + target_link_libraries(record_db PRIVATE tracer_utils clickhouse-cpp-lib fmtlog-shared fmt) + +endif() # T_RECORD_DB + add_custom_target(T_tools) add_dependencies(T_tools record replay extract_config extract_input_subframe extract_output_subframe extract macpdu2wireshark multi textlog csv) + +# Add record_db to T_tools if it was built +if(TARGET record_db) + add_dependencies(T_tools record_db) +endif() + add_dependencies(nr-softmodem T_tools) add_dependencies(nr-uesoftmodem T_tools) add_dependencies(lte-softmodem T_tools) diff --git a/common/utils/T/tracer/Makefile b/common/utils/T/tracer/Makefile index cedf8fdbcf..5d0aed4029 100644 --- a/common/utils/T/tracer/Makefile +++ b/common/utils/T/tracer/Makefile @@ -1,18 +1,23 @@ CC=gcc +CXX=g++ CFLAGS=-Wall -g -pthread -DT_TRACER -I. -I../incgen +CXXFLAGS=-Wall -g -pthread -DT_TRACER -I. -I../incgen -std=c++11 #CFLAGS += -O3 -ffast-math -fomit-frame-pointer LIBS=-lm XLIBS=-lX11 -lpng -lXft -all: record replay extract_config textlog enb ue vcd macpdu2wireshark \ +all: record record_db replay extract_config textlog enb ue vcd macpdu2wireshark \ extract_input_subframe extract_output_subframe to_vcd extract multi \ gnb t_tracer_app_gnb t_tracer_app_ue gnb_mac csv record: utils.o record.o database.o configuration.o $(CC) $(CFLAGS) -o record $^ $(LIBS) +record_db: utils.o record_db.o database.o configuration.o + $(CXX) $(CXXFLAGS) -o record_db $^ $(LIBS) -lclickhouse-cpp-lib $(FMTLOG_LIBS) + replay: utils.o replay.o $(CC) $(CFLAGS) -o replay $^ $(LIBS) @@ -102,8 +107,11 @@ filter/filter.a: %.o: %.c $(CC) $(CFLAGS) -c -o $@ $< +%.o: %.cpp + $(CXX) $(CXXFLAGS) -c -o $@ $< + clean: - rm -f *.o core tracer_remote textlog enb ue vcd record replay + rm -f *.o core tracer_remote textlog enb ue vcd record record_db replay rm -f extract_config macpdu2wireshark extract_input_subframe rm -f extract_output_subframe to_vcd extract multi gnb rm -f t_tracer_app_gnb t_tracer_app_ue gnb_mac csv diff --git a/common/utils/T/tracer/record_db.cpp b/common/utils/T/tracer/record_db.cpp new file mode 100644 index 0000000000..b450f617db --- /dev/null +++ b/common/utils/T/tracer/record_db.cpp @@ -0,0 +1,1324 @@ +/* + * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The OpenAirInterface Software Alliance licenses this file to You under + * the OAI Public License, Version 1.1 (the "License"); you may not use this file + * except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.openairinterface.org/?page_id=698 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + *------------------------------------------------------------------------------- + * For more information about the OpenAirInterface (OAI) Software Alliance: + * contact@openairinterface.org + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "fmtlog.h" + +namespace ch = clickhouse; + +extern "C" { // For connect_to(), socket_send(), parse_database(), etc. +#include +#include +#include +#include +#include +#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((end) - (start)).count()) +#define DURATION_US(start, end) (std::chrono::duration_cast((end) - (start)).count()) +#define DURATION_MS(start, end) (std::chrono::duration_cast((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 this option is mandatory\n" + " -on turn log ON for given GROUP or ID\n" + " -off 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 connect to given IP address (default %s)\n" + " -p connect to given port (default %d)\n" + " -ch-host ClickHouse host (default %s)\n" + " -ch-port ClickHouse native port (default %d)\n" + " -ch-db ClickHouse database name (default %s)\n" + " -ch-user ClickHouse user (optional)\n" + " -ch-pass 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..."); + 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 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 inner_col; // Reusable inner column for ClickHouse + std::shared_ptr 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(); + auto &inner_data = inner_col->GetWritableData(); + inner_data.reserve(size); + + offsets_col = std::make_shared(); + 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 columns; // Ordered list of columns + std::vector 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> data_buffers; + + // Cached typed column pointers to avoid expensive As<> casts on every insert + std::shared_ptr ts_sw_col; + std::shared_ptr ts_tx_col; + std::vector> int_cols; + std::vector> ulong_cols; + std::vector> string_cols; + std::unordered_map 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 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(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(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(); + 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(); + 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(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(); + 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(); + 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(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>(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(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(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, + void *database, + int event_id, + const char *event_data, + int data_length, + std::chrono::high_resolution_clock::time_point t_start) +{ + 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(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(std::chrono::duration_cast(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(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; + + auto t_start = std::chrono::high_resolution_clock::now(); + + 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, database, type, v + vpos, length, t_start) != 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; +} diff --git a/docker/Dockerfile.base.ubuntu b/docker/Dockerfile.base.ubuntu index b0eb5c4d32..f13aa1f917 100644 --- a/docker/Dockerfile.base.ubuntu +++ b/docker/Dockerfile.base.ubuntu @@ -49,6 +49,8 @@ RUN apt-get update && \ bison \ flex \ m4 \ + # libzstd-dev for ClickHouse C++ library + libzstd-dev \ # python3-pip for conf template generation python3-pip \ wget \ diff --git a/docker/Dockerfile.build.fhi72.native_arm.ubuntu b/docker/Dockerfile.build.fhi72.native_arm.ubuntu index 96f50fa485..30c1df5962 100644 --- a/docker/Dockerfile.build.fhi72.native_arm.ubuntu +++ b/docker/Dockerfile.build.fhi72.native_arm.ubuntu @@ -81,8 +81,7 @@ RUN /bin/sh oaienv && \ --ninja --gNB \ --build-lib "telnetsrv enbscope uescope nrscope" \ -t oran_fhlib_5g --cmake-opt -Dxran_LOCATION=/opt/phy/fhi_lib/lib \ - --build-e2 --cmake-opt -DXAPP_MULTILANGUAGE=OFF --cmake-opt -DKPM_VERSION=$KPM_VERSION --cmake-opt -DE2AP_VERSION=$E2AP_VERSION \ - $BUILD_OPTION && \ + --build-e2 --cmake-opt -DXAPP_MULTILANGUAGE=OFF --cmake-opt -DKPM_VERSION=$KPM_VERSION --cmake-opt -DE2AP_VERSION=$E2AP_VERSION && \ # Mainly to see if the sanitize option was perfectly executed ldd ran_build/build/nr-softmodem && \ ldd ran_build/build/liboran_fhlib_5g.so && \ diff --git a/docker/Dockerfile.build.fhi72.ubuntu b/docker/Dockerfile.build.fhi72.ubuntu index 3d483618cd..63ec8a440e 100644 --- a/docker/Dockerfile.build.fhi72.ubuntu +++ b/docker/Dockerfile.build.fhi72.ubuntu @@ -71,8 +71,7 @@ RUN /bin/sh oaienv && \ --ninja --gNB --nrRU \ --build-lib "telnetsrv enbscope uescope nrscope" \ -t oran_fhlib_5g --cmake-opt -Dxran_LOCATION=/opt/phy/fhi_lib/lib \ - --build-e2 --cmake-opt -DKPM_VERSION=$KPM_VERSION --cmake-opt -DE2AP_VERSION=$E2AP_VERSION \ - $BUILD_OPTION && \ + --build-e2 --cmake-opt -DKPM_VERSION=$KPM_VERSION --cmake-opt -DE2AP_VERSION=$E2AP_VERSION && \ # Mainly to see if the sanitize option was perfectly executed ldd ran_build/build/nr-softmodem && \ ldd ran_build/build/liboran_fhlib_5g.so && \ diff --git a/docker/Dockerfile.build.ubuntu b/docker/Dockerfile.build.ubuntu index bc2ed577a6..b919a1aefc 100644 --- a/docker/Dockerfile.build.ubuntu +++ b/docker/Dockerfile.build.ubuntu @@ -42,15 +42,23 @@ RUN \ # Mount CPM package cache --mount=type=cache,target=/root/.cache/cpm/ \ /bin/sh oaienv && \ - mkdir -p cmake_targets/ran_build/build && \ - cd cmake_targets/ran_build/build && \ - cmake ../../../ -GNinja \ - -DENABLE_TELNETSRV=ON -DENABLE_ENBSCOPE=ON -DENABLE_UESCOPE=ON -DENABLE_NRSCOPE=ON \ - -DOAI_USRP=ON -DOAI_BLADERF=ON -DOAI_ETHERNET=ON -DOAI_VRTSIM_TAPS_CLIENT=ON \ - -DE2_AGENT=ON -DKPM_VERSION=KPM_V3_00 -DE2AP_VERSION=E2AP_V3 \ - -DAVX512=OFF && \ - # -DCMAKE_C_FLAGS=-Werror -DCMAKE_CXX_FLAGS=-Werror && \ - ninja + cd cmake_targets && \ + mkdir -p log && \ + ./build_oai -c --ninja \ + --eNB --gNB --RU --UE --nrUE \ + --build-lib "telnetsrv enbscope uescope nrscope" \ + -w USRP -t Ethernet -w BLADERF \ + --build-e2 --cmake-opt -DKPM_VERSION=$KPM_VERSION --cmake-opt -DE2AP_VERSION=$E2AP_VERSION \ + --cmake-opt -DOAI_VRTSIM_TAPS_CLIENT=ON \ + --noavx512 \ + --build-tool-opt -k10 \ + --cmake-opt -DT_RECORD_DB=ON \ + --cmake-opt -DCMAKE_C_FLAGS="-Werror" --cmake-opt -DCMAKE_CXX_FLAGS="-Werror" $BUILD_OPTION && \ + # Mainly to see if the sanitize option was perfectly executed + echo "---- ldd on executables ----" && \ + ldd ran_build/build/*softmodem* ran_build/build/nr-cuup ran_build/build/oairu && \ + echo "---- ldd on shared libraries ----" && \ + ldd ran_build/build/*.so ## Build FlexRIC for SM Models RUN \ diff --git a/docker/Dockerfile.gNB.aerial.ubuntu b/docker/Dockerfile.gNB.aerial.ubuntu index b79ab78fa8..20a453f652 100644 --- a/docker/Dockerfile.gNB.aerial.ubuntu +++ b/docker/Dockerfile.gNB.aerial.ubuntu @@ -55,13 +55,15 @@ RUN /bin/sh oaienv && \ cd cmake_targets && \ mkdir -p log && \ ./build_oai -c -w AERIAL --gNB --ninja --build-tool-opt -k10 \ - --cmake-opt -DCMAKE_C_FLAGS="-Werror" --cmake-opt -DCMAKE_CXX_FLAGS="-Werror" + --cmake-opt -DCMAKE_C_FLAGS="-Werror" --cmake-opt -DCMAKE_CXX_FLAGS="-Werror" \ + --cmake-opt -DT_RECORD_DB=ON #start from scratch for target executable FROM ubuntu:noble AS oai-gnb-aerial ENV DEBIAN_FRONTEND=noninteractive ENV TZ=Europe +ENV PATH="/opt/oai-gnb/bin:$PATH" RUN apt-get update && \ apt-get upgrade --yes && \ @@ -85,17 +87,20 @@ RUN apt-get update && \ WORKDIR /opt/oai-gnb/bin COPY --from=ran-build \ - /oai-ran/cmake_targets/ran_build/build/nr-softmodem ./ + /oai-ran/cmake_targets/ran_build/build/nr-softmodem \ + /oai-ran/cmake_targets/ran_build/build/common/utils/T/tracer/record* \ + ./ + COPY ./docker/scripts/gnb_entrypoint.sh ./entrypoint.sh COPY --from=ran-build \ /oai-ran/cmake_targets/ran_build/build/libparams_libconfig.so \ /oai-ran/cmake_targets/ran_build/build/libparams_yaml.so \ + /oai-ran/cmake_targets/ran_build/build/_deps*/fmtlog-build*/libfmtlog-shared*.so \ + /oai-ran/cmake_targets/ran_build/build/_deps*/clickhouse-cpp-build*/clickhouse/libclickhouse*.so \ /usr/local/lib/ -#copy files from nvIPC COPY --from=ran-build \ - /usr/local/lib/libfmtlog-shared.so \ /usr/local/lib/libnvlog.so \ /usr/local/lib/libnvipc.so \ /usr/local/lib/ @@ -113,5 +118,6 @@ WORKDIR /opt/oai-gnb COPY --from=ran-build /usr/local/bin/yq /usr/local/bin/yq COPY --from=ran-build /tini /tini +COPY --from=ran-build /oai-ran/common/utils/T/T_messages.txt ./ ENTRYPOINT ["/tini", "-v", "--", "/opt/oai-gnb/bin/entrypoint.sh"] CMD ["/opt/oai-gnb/bin/nr-softmodem"] diff --git a/docker/Dockerfile.nrORU.fhi72.ubuntu b/docker/Dockerfile.nrORU.fhi72.ubuntu index 0db881c734..9c4e1b3971 100644 --- a/docker/Dockerfile.nrORU.fhi72.ubuntu +++ b/docker/Dockerfile.nrORU.fhi72.ubuntu @@ -25,7 +25,6 @@ FROM ran-build-fhi72:latest AS ran-build #start from scratch for target executable FROM ubuntu:noble AS oai-nr-oru -ARG BUILD_OPTION ENV DEBIAN_FRONTEND=noninteractive ENV TZ=Europe @@ -50,11 +49,6 @@ RUN apt-get update && \ xz-utils \ libnuma-dev \ libyaml-cpp-dev && \ -# if the --sanitize option was used to build, additional packages are required - /bin/bash -c 'if [[ "$BUILD_OPTION" = "--sanitize" ]]; then DEBIAN_FRONTEND=noninteractive apt-get install --yes \ - libasan8 \ - libubsan1 \ - liblapacke; fi' && \ rm -rf /var/lib/apt/lists/* WORKDIR /opt/nr-oru/bin diff --git a/docker/Dockerfile.nrUE.ubuntu b/docker/Dockerfile.nrUE.ubuntu index 7c8b15d013..1cd151db53 100644 --- a/docker/Dockerfile.nrUE.ubuntu +++ b/docker/Dockerfile.nrUE.ubuntu @@ -35,6 +35,7 @@ FROM ubuntu:noble AS oai-nr-ue ARG BUILD_OPTION ENV DEBIAN_FRONTEND=noninteractive ENV TZ=Europe/Paris +ENV PATH="/opt/oai-nr-ue/bin:$PATH" RUN apt-get update && \ apt-get upgrade --yes && \ @@ -67,6 +68,7 @@ RUN apt-get update && \ iperf3 \ iperf \ libyaml-cpp-dev \ + libzstd1 \ libnanomsg5 && \ # if the --sanitize option was used to build, additional packages are required /bin/bash -c 'if [[ "$BUILD_OPTION" = "--sanitize" ]]; then DEBIAN_FRONTEND=noninteractive apt-get install --yes \ @@ -78,6 +80,7 @@ RUN apt-get update && \ WORKDIR /opt/oai-nr-ue/bin COPY --from=nr-ue-build \ /oai-ran/cmake_targets/ran_build/build/nr-uesoftmodem \ + /oai-ran/cmake_targets/ran_build/build/common/utils/T/tracer/record* \ ./ COPY ./docker/scripts/nr_ue_entrypoint.sh ./entrypoint.sh @@ -94,10 +97,17 @@ COPY --from=nr-ue-build \ /oai-ran/cmake_targets/ran_build/build/libtelnetsrv.so \ /oai-ran/cmake_targets/ran_build/build/libtelnetsrv_ciUE.so \ /oai-ran/cmake_targets/ran_build/build/libtelnetsrv_5Gue.so \ + /oai-ran/cmake_targets/ran_build/build/libnrscope.so \ + /oai-ran/cmake_targets/ran_build/build/libtelnetsrv.so \ /oai-ran/cmake_targets/ran_build/build/libparams_yaml.so \ /oai-ran/cmake_targets/ran_build/build/libvrtsim.so \ + # Wildcards so it doesn't fail if not present + /oai-ran/cmake_targets/ran_build/build/_deps*/fmtlog-build*/libfmtlog-shared*.so \ + /oai-ran/cmake_targets/ran_build/build/_deps*/clickhouse-cpp-build*/clickhouse/libclickhouse*.so \ /usr/local/lib/ +# /oai-ran/cmake_targets/ran_build/build/libimscope.so \ + # Now we are copying from builder-image the UHD files. COPY --from=nr-ue-base /usr/local/bin/uhd_find_devices /usr/local/bin COPY --from=nr-ue-base /usr/local/lib/libuhd.so.4.8.0 /usr/local/lib @@ -124,5 +134,6 @@ WORKDIR /opt/oai-nr-ue COPY --from=nr-ue-base /usr/local/bin/yq /usr/local/bin/yq COPY --from=nr-ue-base /tini /tini +COPY --from=nr-ue-build /oai-ran/common/utils/T/T_messages.txt ./ CMD ["/opt/oai-nr-ue/bin/nr-uesoftmodem"] ENTRYPOINT ["/tini", "-v", "--", "/opt/oai-nr-ue/bin/entrypoint.sh"]