Cyclic buffer telemetry exporter moved to use plug-in infrastructure (ai-dynamo#1088)

* Cyclic buffer telemetry exporter moved to use plug-in infrastructure

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Removing redundant headers and try/catch routines

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Minor fixes

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Moving buffer to be class member

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Removing fall back routine

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Allowing NIXL_TELEMETRY_DIR to be not set

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Compilance with previous functionality. If NIXL_TELEMETRY_DIR is not set, exporter will not be created

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Removing redaundant logging

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

* Not letting to load default telemetry plug-in if NIXL_TELEMETRY_DIR is not set

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

---------

Signed-off-by: Aleksandr Bilkovskii <alexanderb@nvidia.com>

Author: alexanderbilk

meson.build

@@ -312,7 +312,7 @@ if get_option('buildtype') == 'debug'
     run_command('truncate', '-s 0', plugfile, check: true)
 endif
 
-nixl_inc_dirs = include_directories('src/api/cpp', 'src/api/cpp/backend', 'src/infra', 'src/core')
+nixl_inc_dirs = include_directories('src/api/cpp', 'src/api/cpp/backend', 'src/infra', 'src/core', 'src/core/telemetry')
 nixl_gpu_inc_dirs = include_directories('src/api/gpu/ucx')
 plugins_inc_dirs = include_directories('src/plugins')
 utils_inc_dirs = include_directories('src/utils')

src/api/cpp/telemetry/telemetry_exporter.h

@@ -23,14 +23,12 @@
 #include <string>
 
 inline constexpr char telemetryExporterVar[] = "NIXL_TELEMETRY_EXPORTER";
-inline constexpr char telemetryExporterOutputPathVar[] = "NIXL_TELEMETRY_EXPORTER_OUTPUT_PATH";
 
 /**
  * @struct nixlTelemetryExporterInitParams
  * @brief Initialization parameters for telemetry exporters
  */
 struct nixlTelemetryExporterInitParams {
-    std::string outputPath; // Output path (file path, URL, etc.)
     std::string agentName;
     size_t maxEventsBuffered;
 };

src/core/meson.build

@@ -57,7 +57,9 @@ nixl_lib = library('nixl',
                    'nixl_agent.cpp',
                    'nixl_plugin_manager.cpp',
                    'nixl_listener.cpp',
-                   'telemetry.cpp',
+                   'telemetry/telemetry.cpp',
+                   'telemetry/buffer_exporter.cpp',
+                   'telemetry/buffer_plugin.cpp',
                    include_directories: [ nixl_inc_dirs, utils_inc_dirs ],
                    link_args: ['-lstdc++fs'],
                    dependencies: nixl_lib_deps,

src/core/nixl_agent.cpp

@@ -32,7 +32,6 @@
 #include "telemetry_event.h"
 
 constexpr char TELEMETRY_ENABLED_VAR[] = "NIXL_TELEMETRY_ENABLE";
-constexpr char TELEMETRY_DIR_VAR[] = "NIXL_TELEMETRY_DIR";
 static const std::vector<std::vector<std::string>> illegal_plugin_combinations = {
     {"GDS", "GDS_MT"},
 };
@@ -126,19 +125,12 @@ nixlAgentData::nixlAgentData(const std::string &name, const nixlAgentConfig &cfg
 
     memorySection = new nixlLocalSection();
     const char *telemetry_env_val = std::getenv(TELEMETRY_ENABLED_VAR);
-    const char *telemetry_env_dir = std::getenv(TELEMETRY_DIR_VAR);
 
     if (telemetry_env_val != nullptr) {
         if (!strcasecmp(telemetry_env_val, "y") || !strcasecmp(telemetry_env_val, "1") ||
             !strcasecmp(telemetry_env_val, "yes") || !strcasecmp(telemetry_env_val, "on")) {
             telemetryEnabled = true;
-            if (telemetry_env_dir != nullptr) {
-                std::string telemetry_file = std::string(telemetry_env_dir) + "/" + name;
-                telemetry_ = std::make_unique<nixlTelemetry>(telemetry_file, backendEngines);
-                NIXL_DEBUG << "NIXL telemetry is enabled with output file: " << telemetry_file;
-            } else {
-                NIXL_DEBUG << "NIXL telemetry is enabled without an output file";
-            }
+            telemetry_ = std::make_unique<nixlTelemetry>(name, backendEngines);
         } else if (cfg.captureTelemetry) {
             telemetryEnabled = true;
             NIXL_WARN << "NIXL telemetry is enabled through config, "

src/core/nixl_plugin_manager.cpp

@@ -674,4 +674,5 @@ void nixlPluginManager::registerBuiltinPlugins() {
 #ifdef STATIC_PLUGIN_HF3FS
     NIXL_REGISTER_STATIC_PLUGIN(Backend, HF3FS)
 #endif
+    NIXL_REGISTER_STATIC_PLUGIN(Telemetry, BUFFER)
 }

src/core/telemetry/buffer_exporter.cpp

@@ -0,0 +1,45 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+#include "buffer_exporter.h"
+#include "common/nixl_log.h"
+
+namespace {
+static std::filesystem::path
+getFilePath(const nixlTelemetryExporterInitParams &init_params) {
+    // if we reach here, we ensured env var is set
+    auto telemetry_dir = std::getenv(telemetryDirVar);
+    return std::filesystem::path(telemetry_dir) / init_params.agentName.data();
+}
+} // namespace
+
+nixlTelemetryBufferExporter::nixlTelemetryBufferExporter(
+    const nixlTelemetryExporterInitParams &init_params)
+    : nixlTelemetryExporter(init_params),
+      filePath_(getFilePath(init_params)),
+      buffer_(filePath_.string(), true, TELEMETRY_VERSION, getMaxEventsBuffered()) {
+    NIXL_INFO << "Telemetry enabled, using buffer path: " << filePath_.string()
+              << " with size: " << getMaxEventsBuffered();
+}
+
+nixl_status_t
+nixlTelemetryBufferExporter::exportEvent(const nixlTelemetryEvent &event) {
+    if (!buffer_.push(event)) {
+        return NIXL_ERR_UNKNOWN;
+    }
+
+    return NIXL_SUCCESS;
+}

src/core/telemetry/buffer_exporter.h

@@ -0,0 +1,52 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+#ifndef _TELEMETRY_BUFFER_EXPORTER_H
+#define _TELEMETRY_BUFFER_EXPORTER_H
+
+#include "common/cyclic_buffer.h"
+#include "telemetry/telemetry_exporter.h"
+#include "telemetry_event.h"
+#include "nixl_types.h"
+
+#include <filesystem>
+
+constexpr const char telemetryDirVar[] = "NIXL_TELEMETRY_DIR";
+
+/**
+ * @class nixlTelemetryBufferExporter
+ * @brief Shared memory buffer based telemetry exporter implementation
+ *
+ * This class implements the telemetry exporter interface to export
+ * telemetry events to a shared memory buffer.
+ */
+class nixlTelemetryBufferExporter : public nixlTelemetryExporter {
+public:
+    /**
+     * @brief Constructor using init params (plugin-compatible)
+     * @param init_params Initialization parameters
+     */
+    explicit nixlTelemetryBufferExporter(const nixlTelemetryExporterInitParams &init_params);
+
+    nixl_status_t
+    exportEvent(const nixlTelemetryEvent &event) override;
+
+private:
+    std::filesystem::path filePath_;
+    sharedRingBuffer<nixlTelemetryEvent> buffer_;
+};
+
+#endif // _TELEMETRY_BUFFER_EXPORTER_H

src/core/telemetry/buffer_plugin.cpp

@@ -0,0 +1,28 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ */
+
+#include "buffer_exporter.h"
+#include "telemetry/telemetry_plugin.h"
+#include "telemetry/telemetry_exporter.h"
+
+// Plugin type alias for convenience
+using buffer_exporter_plugin_t = nixlTelemetryPluginCreator<nixlTelemetryBufferExporter>;
+
+nixlTelemetryPlugin *
+createStaticBUFFERPlugin() {
+    return buffer_exporter_plugin_t::create(nixlTelemetryPluginApiVersionV1, "buffer", "1.0.0");
+}

src/core/telemetry.cpp src/core/telemetry/telemetry.cppsrc/core/telemetry.cpp renamed to src/core/telemetry/telemetry.cpp

@@ -27,20 +27,23 @@
 #include "telemetry.h"
 #include "telemetry_event.h"
 #include "util.h"
+#include "plugin_manager.h"
+#include "buffer_exporter.h"
 
 using namespace std::chrono_literals;
 namespace fs = std::filesystem;
 
 constexpr std::chrono::milliseconds DEFAULT_TELEMETRY_RUN_INTERVAL = 100ms;
 constexpr size_t DEFAULT_TELEMETRY_BUFFER_SIZE = 4096;
+constexpr const char *defaultTelemetryPlugin = "BUFFER";
 
-nixlTelemetry::nixlTelemetry(const std::string &file_path, backend_map_t &backend_map)
+nixlTelemetry::nixlTelemetry(const std::string &agent_name, backend_map_t &backend_map)
     : pool_(1),
       writeTask_(pool_.get_executor(), DEFAULT_TELEMETRY_RUN_INTERVAL, false),
-      file_(file_path),
+      agentName_(agent_name),
       backendMap_(backend_map) {
-    if (file_path.empty()) {
-        throw std::invalid_argument("Telemetry file path cannot be empty");
+    if (agent_name.empty()) {
+        throw std::invalid_argument("Expected non-empty agent name in nixl telemetry create");
     }
     initializeTelemetry();
 }
@@ -69,17 +72,38 @@ nixlTelemetry::initializeTelemetry() {
         std::stoul(std::getenv(TELEMETRY_BUFFER_SIZE_VAR)) :
         DEFAULT_TELEMETRY_BUFFER_SIZE;
 
-    auto full_file_path = fs::path(file_);
-
     if (buffer_size == 0) {
         throw std::invalid_argument("Telemetry buffer size cannot be 0");
     }
 
-    NIXL_INFO << "Telemetry enabled, using buffer path: " << full_file_path
-              << " with size: " << buffer_size;
+    const char *exporter_name = std::getenv(telemetryExporterVar);
+
+    if (!exporter_name) {
+        NIXL_INFO << "No telemetry exporter was specified, using default: "
+                  << defaultTelemetryPlugin;
+        exporter_name = defaultTelemetryPlugin;
+        if (!std::getenv(telemetryDirVar)) {
+            NIXL_DEBUG << telemetryDirVar
+                       << " is not set, NIXL telemetry is enabled without any exporter";
+            return;
+        }
+    }
+    auto &plugin_manager = nixlPluginManager::getInstance();
+    std::shared_ptr<const nixlTelemetryPluginHandle> plugin_handle =
+        plugin_manager.loadTelemetryPlugin(exporter_name);
+
+    if (plugin_handle == nullptr) {
+        throw std::runtime_error("Failed to load telemetry plugin: " + std::string(exporter_name));
+    }
+
+    const nixlTelemetryExporterInitParams init_params{agentName_, buffer_size};
+    exporter_ = plugin_handle->createExporter(init_params);
+    if (!exporter_) {
+        NIXL_ERROR << "Failed to create telemetry exporter: " << exporter_name;
+        return;
+    }
 
-    buffer_ = std::make_unique<sharedRingBuffer<nixlTelemetryEvent>>(
-        full_file_path, true, TELEMETRY_VERSION, buffer_size);
+    NIXL_DEBUG << "NIXL telemetry is enabled with " << exporter_name << "exporter";
 
     auto run_interval = std::getenv(TELEMETRY_RUN_INTERVAL_VAR) ?
         std::chrono::milliseconds(std::stoul(std::getenv(TELEMETRY_RUN_INTERVAL_VAR))) :
@@ -96,14 +120,14 @@ bool
 nixlTelemetry::writeEventHelper() {
     std::vector<nixlTelemetryEvent> next_queue;
     // assume next buffer will be the same size as the current one
-    next_queue.reserve(buffer_->capacity());
+    next_queue.reserve(exporter_->getMaxEventsBuffered());
     {
         std::lock_guard<std::mutex> lock(mutex_);
         events_.swap(next_queue);
     }
     for (auto &event : next_queue) {
         // if full, ignore
-        buffer_->push(event);
+        exporter_->exportEvent(event);
     }
     // collect all events and sort them by timestamp
     std::vector<nixlTelemetryEvent> all_events;
@@ -122,7 +146,7 @@ nixlTelemetry::writeEventHelper() {
                   return a.timestampUs_ < b.timestampUs_;
               });
     for (auto &event : all_events) {
-        buffer_->push(event);
+        exporter_->exportEvent(event);
     }
     return true;
 }

src/core/telemetry.h src/core/telemetry/telemetry.hsrc/core/telemetry.h renamed to src/core/telemetry/telemetry.h

@@ -18,6 +18,7 @@
 #define _TELEMETRY_H
 
 #include "common/cyclic_buffer.h"
+#include "telemetry/telemetry_exporter.h"
 #include "telemetry_event.h"
 #include "mem_section.h"
 #include "nixl_types.h"
@@ -49,7 +50,7 @@ struct periodicTask {
 
 class nixlTelemetry {
 public:
-    nixlTelemetry(const std::string &file_path, backend_map_t &backend_map);
+    nixlTelemetry(const std::string &agent_name, backend_map_t &backend_map);
 
     ~nixlTelemetry();
 
@@ -81,12 +82,13 @@ class nixlTelemetry {
     updateData(const std::string &event_name, nixl_telemetry_category_t category, uint64_t value);
     bool
     writeEventHelper();
+    std::unique_ptr<nixlTelemetryExporter> exporter_;
     std::unique_ptr<sharedRingBuffer<nixlTelemetryEvent>> buffer_;
     std::vector<nixlTelemetryEvent> events_;
     std::mutex mutex_;
     asio::thread_pool pool_;
     periodicTask writeTask_;
-    std::string file_;
+    std::string agentName_;
     backend_map_t &backendMap_;
 };