diff --git a/docs/telemetry.md b/docs/telemetry.md index f231b8073a..668a9ddad3 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -18,7 +18,6 @@ Custom telemetry exporter plug-ins can be created according to [src/plugins/tele ### Event Structure Each telemetry event contains: -- **Timestamp**: Microsecond precision timestamp - **Category**: Event category for filtering and aggregation - **Event Name**: Descriptive name/identifier for the event - **Value**: Numeric value associated with the event @@ -61,7 +60,7 @@ The **Shared Memory Buffer** plug-in, contains the data per transaction event, w ### Telemetry Details -- Timestamp is done after the operation completion. +- Producers append events under a mutex, consumers read them in **ring insertion order**. - Current design allows silent telemetry loss. - Current design does not support selective telemetry(e.g per category). All the telemetry events could be either ON or OFF. @@ -128,14 +127,12 @@ Both readers produce similar formatted output: ``` === NIXL Telemetry Event === -Timestamp: 2025-01-15 14:30:25.123456 Category: TRANSFER Event: agent_tx_bytes Value: 1048576 =========================== === NIXL Telemetry Event === -Timestamp: 2025-01-15 14:30:25.124567 Category: MEMORY Event: agent_memory_registered Value: 4096 diff --git a/examples/cpp/telemetry_reader.cpp b/examples/cpp/telemetry_reader.cpp index 2274f2f371..ba745c3109 100644 --- a/examples/cpp/telemetry_reader.cpp +++ b/examples/cpp/telemetry_reader.cpp @@ -44,21 +44,6 @@ signal_handler(int signal) { } } -std::string -format_timestamp(uint64_t timestamp_us) { - auto time_point = - std::chrono::system_clock::time_point(std::chrono::microseconds(timestamp_us)); - auto time_t = std::chrono::system_clock::to_time_t(time_point); - - std::stringstream ss; - ss << std::put_time(std::localtime(&time_t), "%Y-%m-%d %H:%M:%S"); - - auto microseconds = timestamp_us % 1000000; - ss << "." << std::setfill('0') << std::setw(6) << microseconds; - - return ss.str(); -} - std::string format_bytes(uint64_t bytes) { const char *units[] = {"B", "KB", "MB", "GB", "TB"}; @@ -80,7 +65,6 @@ print_telemetry_event(const nixlTelemetryEvent &event) { // Can be extended to more general ostream if needed // friend std::ostream &operator<<(std::ostream &os, const nixlTelemetryEvent &event) std::cout << "\n=== NIXL Telemetry Event ===" << std::endl; - std::cout << "Timestamp: " << format_timestamp(event.timestampUs_) << std::endl; std::cout << "Category: " << nixlEnumStrings::telemetryCategoryStr(event.category_) << std::endl; std::cout << "Event name: " << event.eventName_ << std::endl; diff --git a/examples/python/telemetry_reader.py b/examples/python/telemetry_reader.py index 4eb7e0fd92..4733db1ccb 100755 --- a/examples/python/telemetry_reader.py +++ b/examples/python/telemetry_reader.py @@ -23,7 +23,6 @@ import signal import sys import time -from datetime import datetime # Set up logging logging.basicConfig( @@ -34,7 +33,7 @@ logger = logging.getLogger(__name__) # Constants from telemetry_event.h -TELEMETRY_VERSION = 1 +TELEMETRY_VERSION = 2 MAX_EVENT_NAME_LEN = 32 # NIXL telemetry categories @@ -65,7 +64,6 @@ class NixlTelemetryEvent(ctypes.Structure): _pack_ = 1 _fields_ = [ - ("timestamp_us", ctypes.c_uint64), ("category", ctypes.c_int), ("event_name", ctypes.c_char * MAX_EVENT_NAME_LEN), ("_padding", ctypes.c_uint32), @@ -90,7 +88,7 @@ class BufferHeader(ctypes.Structure): class SharedRingBuffer: """Python wrapper for the C++ SharedRingBuffer class""" - def __init__(self, file_path, version=1): + def __init__(self, file_path, version=TELEMETRY_VERSION): self.file_path = file_path self.version = version self.file_fd = -1 @@ -122,11 +120,12 @@ def _map_header_only(self): temp_header = BufferHeader.from_buffer(header_mmap) - if temp_header.version != self.version: + actual_version = temp_header.version + if actual_version != self.version: del temp_header header_mmap.close() raise RuntimeError( - f"Version mismatch: expected {self.version}, got {temp_header.version}" + f"Version mismatch: expected {self.version}, got {actual_version}" ) self.buffer_size = temp_header.capacity @@ -197,13 +196,6 @@ def __del__(self): os.close(self.file_fd) -def format_timestamp(timestamp_us): - """Format timestamp in microseconds to readable format""" - dt = datetime.fromtimestamp(timestamp_us / 1_000_000) - microseconds = timestamp_us % 1_000_000 - return f"{dt.strftime('%Y-%m-%d %H:%M:%S')}.{microseconds:06d}" - - def format_bytes(bytes_val): """Format bytes to human readable format""" units = ["B", "KB", "MB", "GB", "TB"] @@ -235,7 +227,6 @@ def get_telemetry_category_string(category): def print_telemetry_event(event): """Print telemetry event in a formatted way""" logger.info("\n=== NIXL Telemetry Event ===") - logger.info("Timestamp: %s", format_timestamp(event.timestamp_us)) # Decode event name event_name = event.event_name.decode("utf-8").rstrip("\x00") diff --git a/src/api/cpp/backend/backend_engine.h b/src/api/cpp/backend/backend_engine.h index d7a1cb7293..b62d284dd7 100644 --- a/src/api/cpp/backend/backend_engine.h +++ b/src/api/cpp/backend/backend_engine.h @@ -66,12 +66,8 @@ class nixlBackendEngine { if (!enableTelemetry_) return; if (telemetryEvents_.size() >= MAX_TELEMETRY_QUEUE_SIZE) return; std::lock_guard lock(telemetryEventsMutex_); - telemetryEvents_.emplace_back(std::chrono::duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count(), - nixl_telemetry_category_t::NIXL_TELEMETRY_BACKEND, - event_name, - value); + telemetryEvents_.emplace_back( + nixl_telemetry_category_t::NIXL_TELEMETRY_BACKEND, event_name, value); } public: diff --git a/src/api/cpp/telemetry/telemetry_plugin.h b/src/api/cpp/telemetry/telemetry_plugin.h index fa038d97dc..bffa176109 100644 --- a/src/api/cpp/telemetry/telemetry_plugin.h +++ b/src/api/cpp/telemetry/telemetry_plugin.h @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -24,10 +24,10 @@ #include #include -enum class nixl_telemetry_plugin_api_version : unsigned int { V1 = 1 }; - -inline constexpr nixl_telemetry_plugin_api_version nixlTelemetryPluginApiVersionV1 = - nixl_telemetry_plugin_api_version::V1; +enum class nixl_telemetry_plugin_api_version : unsigned int { + V1 = 1, + V2 = 2, +}; // Type alias for exporter creation function using exporter_creator_fn_t = diff --git a/src/core/nixl_plugin_manager.cpp b/src/core/nixl_plugin_manager.cpp index fc351d14de..96c1ff6953 100644 --- a/src/core/nixl_plugin_manager.cpp +++ b/src/core/nixl_plugin_manager.cpp @@ -186,11 +186,10 @@ telemetryLoader(void *handle, const std::string &plugin_path) { return nullptr; } - // Check API version - if (plugin->api_version != nixlTelemetryPluginApiVersionV1) { + if (plugin->api_version != nixl_telemetry_plugin_api_version::V2) { NIXL_ERROR << "Plugin API version mismatch for " << plugin_path << ": expected " - << static_cast(nixlTelemetryPluginApiVersionV1) << ", got " - << static_cast(plugin->api_version); + << static_cast(nixl_telemetry_plugin_api_version::V2) << ", got " + << static_cast(plugin->api_version); dlclose(handle); return nullptr; } diff --git a/src/core/telemetry/buffer_plugin.cpp b/src/core/telemetry/buffer_plugin.cpp index 76491c9dfd..2359d018dc 100644 --- a/src/core/telemetry/buffer_plugin.cpp +++ b/src/core/telemetry/buffer_plugin.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -24,5 +24,6 @@ using buffer_exporter_plugin_t = nixlTelemetryPluginCreator= maxBufferedEvents_) { return; } - events_.emplace_back(std::chrono::duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count(), - category, - event_name, - value); + events_.emplace_back(category, event_name, value); } // The next 4 methods might be removed, as addXferTime covers them. @@ -232,22 +227,15 @@ nixlTelemetry::addXferTime(std::chrono::microseconds xfer_time, bool is_write, u const char *bytes_name = is_write ? "agent_tx_bytes" : "agent_rx_bytes"; const char *requests_name = is_write ? "agent_tx_requests_num" : "agent_rx_requests_num"; - const auto time = std::chrono::duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - const std::lock_guard lock(mutex_); if (events_.size() + 3 > maxBufferedEvents_) { return; } - events_.emplace_back(time, - nixl_telemetry_category_t::NIXL_TELEMETRY_PERFORMANCE, + events_.emplace_back(nixl_telemetry_category_t::NIXL_TELEMETRY_PERFORMANCE, "agent_xfer_time", xfer_time.count()); - events_.emplace_back( - time, nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, bytes_name, bytes); - events_.emplace_back( - time, nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, requests_name, 1); + events_.emplace_back(nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, bytes_name, bytes); + events_.emplace_back(nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, requests_name, 1); } void diff --git a/src/core/telemetry/telemetry_event.h b/src/core/telemetry/telemetry_event.h index 308a26797d..e5a26c9c56 100644 --- a/src/core/telemetry/telemetry_event.h +++ b/src/core/telemetry/telemetry_event.h @@ -27,7 +27,7 @@ constexpr char TELEMETRY_BUFFER_SIZE_VAR[] = "NIXL_TELEMETRY_BUFFER_SIZE"; constexpr char TELEMETRY_RUN_INTERVAL_VAR[] = "NIXL_TELEMETRY_RUN_INTERVAL"; -constexpr inline int TELEMETRY_VERSION = 1; +constexpr inline int TELEMETRY_VERSION = 2; constexpr inline size_t MAX_EVENT_NAME_LEN = 32; /** @@ -55,29 +55,25 @@ telemetryCategoryStr(const nixl_telemetry_category_t &category); * @brief A structure to hold individual telemetry event data for cyclic buffer storage */ struct nixlTelemetryEvent { - uint64_t timestampUs_; nixl_telemetry_category_t category_; // Main event category for filtering char eventName_[MAX_EVENT_NAME_LEN]; // Detailed event name/identifier uint64_t value_; // Numeric value associated with the event nixlTelemetryEvent() noexcept = default; - nixlTelemetryEvent(uint64_t timestamp_us, - nixl_telemetry_category_t category, + nixlTelemetryEvent(nixl_telemetry_category_t category, const char *event_name, uint64_t value) noexcept - : timestampUs_(timestamp_us), - category_(category), + : category_(category), value_(value) { strncpy(eventName_, event_name, MAX_EVENT_NAME_LEN - 1); eventName_[MAX_EVENT_NAME_LEN - 1] = '\0'; } - nixlTelemetryEvent(uint64_t timestamp_us, - nixl_telemetry_category_t category, + nixlTelemetryEvent(nixl_telemetry_category_t category, const std::string &event_name, uint64_t value) noexcept - : nixlTelemetryEvent(timestamp_us, category, event_name.c_str(), value) {} + : nixlTelemetryEvent(category, event_name.c_str(), value) {} }; #endif diff --git a/src/plugins/telemetry/README.md b/src/plugins/telemetry/README.md index d65c2761ea..f25ef6ecd0 100644 --- a/src/plugins/telemetry/README.md +++ b/src/plugins/telemetry/README.md @@ -1,5 +1,5 @@