Skip to content
Merged
5 changes: 1 addition & 4 deletions docs/telemetry.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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
Expand Down
16 changes: 0 additions & 16 deletions examples/cpp/telemetry_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"};
Expand All @@ -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;
Expand Down
19 changes: 5 additions & 14 deletions examples/python/telemetry_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import signal
import sys
import time
from datetime import datetime

# Set up logging
logging.basicConfig(
Expand All @@ -34,7 +33,7 @@
logger = logging.getLogger(__name__)

# Constants from telemetry_event.h
TELEMETRY_VERSION = 1
TELEMETRY_VERSION = 2
Comment thread
coderabbitai[bot] marked this conversation as resolved.
MAX_EVENT_NAME_LEN = 32

# NIXL telemetry categories
Expand Down Expand Up @@ -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),
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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"]
Expand Down Expand Up @@ -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")
Expand Down
8 changes: 2 additions & 6 deletions src/api/cpp/backend/backend_engine.h
Original file line number Diff line number Diff line change
Expand Up @@ -66,12 +66,8 @@ class nixlBackendEngine {
if (!enableTelemetry_) return;
if (telemetryEvents_.size() >= MAX_TELEMETRY_QUEUE_SIZE) return;
std::lock_guard<std::mutex> lock(telemetryEventsMutex_);
telemetryEvents_.emplace_back(std::chrono::duration_cast<std::chrono::microseconds>(
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:
Expand Down
10 changes: 5 additions & 5 deletions src/api/cpp/telemetry/telemetry_plugin.h
Original file line number Diff line number Diff line change
@@ -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");
Expand All @@ -24,10 +24,10 @@
#include <string_view>
#include <memory>

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 =
Expand Down
7 changes: 3 additions & 4 deletions src/core/nixl_plugin_manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<int>(nixlTelemetryPluginApiVersionV1) << ", got "
<< static_cast<int>(plugin->api_version);
<< static_cast<unsigned int>(nixl_telemetry_plugin_api_version::V2) << ", got "
<< static_cast<unsigned int>(plugin->api_version);
dlclose(handle);
return nullptr;
}
Expand Down
5 changes: 3 additions & 2 deletions src/core/telemetry/buffer_plugin.cpp
Original file line number Diff line number Diff line change
@@ -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");
Expand All @@ -24,5 +24,6 @@ using buffer_exporter_plugin_t = nixlTelemetryPluginCreator<nixlTelemetryBufferE

nixlTelemetryPlugin *
createStaticBUFFERPlugin() {
return buffer_exporter_plugin_t::create(nixlTelemetryPluginApiVersionV1, "buffer", "1.0.0");
return buffer_exporter_plugin_t::create(
nixl_telemetry_plugin_api_version::V2, "buffer", "1.0.0");
}
20 changes: 4 additions & 16 deletions src/core/telemetry/telemetry.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -174,12 +174,7 @@ nixlTelemetry::updateData(const std::string &event_name,
if (events_.size() >= maxBufferedEvents_) {
return;
}
events_.emplace_back(std::chrono::duration_cast<std::chrono::microseconds>(
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.
Expand Down Expand Up @@ -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::microseconds>(
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
Expand Down
14 changes: 5 additions & 9 deletions src/core/telemetry/telemetry_event.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
constexpr inline size_t MAX_EVENT_NAME_LEN = 32;

/**
Expand Down Expand Up @@ -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) {}
Comment thread
BatshevaBlack marked this conversation as resolved.
};

#endif
9 changes: 4 additions & 5 deletions src/plugins/telemetry/README.md
Original file line number Diff line number Diff line change
@@ -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");
Expand Down Expand Up @@ -84,7 +84,7 @@ nixlTelemetryCsvExporter::nixlTelemetryCsvExporter(
}

// Write CSV header
file_ << "timestamp_us,category,event_name,value\n";
file_ << "category,event_name,value\n";
NIXL_INFO << "CSV exporter initialized: " << file_path;
}

Expand All @@ -95,8 +95,7 @@ nixlTelemetryCsvExporter::exportEvent(const nixlTelemetryEvent &event) {
}

try {
file_ << event.timestampUs_ << ","
<< static_cast<int>(event.category_) << ","
file_ << static_cast<int>(event.category_) << ","
<< event.eventName_ << ","
<< event.value_ << "\n";
file_.flush();
Expand All @@ -122,7 +121,7 @@ using csv_exporter_plugin_t = nixlTelemetryPluginCreator<nixlTelemetryCsvExporte
extern "C" NIXL_TELEMETRY_PLUGIN_EXPORT nixlTelemetryPlugin *
nixl_telemetry_plugin_init() {
return csv_exporter_plugin_t::create(
NIXL_TELEMETRY_PLUGIN_API_VERSION,
nixl_telemetry_plugin_api_version::V2,
"csv", // Plugin name
"1.0.0" // Plugin version
);
Expand Down
4 changes: 2 additions & 2 deletions src/plugins/telemetry/prometheus/prometheus_plugin.cpp
Original file line number Diff line number Diff line change
@@ -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");
Expand All @@ -26,7 +26,7 @@ using prometheus_exporter_plugin_t = nixlTelemetryPluginCreator<nixlTelemetryPro
extern "C" NIXL_TELEMETRY_PLUGIN_EXPORT nixlTelemetryPlugin *
nixl_telemetry_plugin_init() {
return prometheus_exporter_plugin_t::create(
nixlTelemetryPluginApiVersionV1, "prometheus", "1.0.0");
nixl_telemetry_plugin_api_version::V2, "prometheus", "1.0.0");
}

// Plugin cleanup function
Expand Down
4 changes: 1 addition & 3 deletions test/gtest/telemetry_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -185,10 +185,8 @@ TEST_F(telemetryTest, TransferBytesTracking) {
}

TEST_F(telemetryTest, TelemetryEventStructure) {
nixlTelemetryEvent event1(
1234567890, nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, "test_event", 42);
nixlTelemetryEvent event1(nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER, "test_event", 42);

EXPECT_EQ(event1.timestampUs_, 1234567890);
EXPECT_EQ(event1.category_, nixl_telemetry_category_t::NIXL_TELEMETRY_TRANSFER);
EXPECT_EQ(event1.value_, 42);
EXPECT_STREQ(event1.eventName_, "test_event");
Expand Down
Loading