Telemetry exporter plug-in infra

DRAFT: Prometheus exporter implemetation
Signed-off-by: Aleksandr Bilkovskii <[email protected]>

Author: alexanderbilk

src/api/cpp/telemetry/telemetry_exporter.h

@@ -0,0 +1,74 @@
+/*
+ * 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_EXPORTER_H
+#define _TELEMETRY_EXPORTER_H
+
+#include "nixl_types.h"
+#include "telemetry_event.h"
+#include "common/cyclic_buffer.h"
+
+#include <string>
+#include <vector>
+#include <fstream>
+
+constexpr char TELEMETRY_EXPORTER_VAR[] = "NIXL_TELEMETRY_EXPORTER";
+constexpr char TELEMETRY_EXPORTER_OUTPUT_PATH_VAR[] = "NIXL_TELEMETRY_EXPORTER_OUTPUT_PATH";
+
+/**
+ * @struct nixlTelemetryExporterInitParams
+ * @brief Initialization parameters for telemetry exporters
+ */
+struct nixlTelemetryExporterInitParams {
+    std::string outputPath; // Output path (file path, URL, etc.)
+    uint32_t eventLimit; // Maximum number of events to buffer
+};
+
+/**
+ * @class nixlTelemetryExporter
+ * @brief Abstract base class for telemetry exporters
+ *
+ * This class defines the interface that all telemetry exporters must implement.
+ * It provides the core functionality for reading telemetry events and exporting
+ * them to various destinations.
+ */
+class nixlTelemetryExporter {
+protected:
+    uint32_t eventLimit_;
+
+public:
+    explicit nixlTelemetryExporter(const nixlTelemetryExporterInitParams *init_params)
+        : eventLimit_(init_params->eventLimit) {};
+    nixlTelemetryExporter(nixlTelemetryExporter &&) = delete;
+    nixlTelemetryExporter(const nixlTelemetryExporter &) = delete;
+
+    void
+    operator=(nixlTelemetryExporter &&) = delete;
+    void
+    operator=(const nixlTelemetryExporter &) = delete;
+
+    virtual ~nixlTelemetryExporter() = default;
+
+    uint32_t
+    getEventLimit() const noexcept {
+        return eventLimit_;
+    }
+
+    virtual nixl_status_t
+    exportEvent(const nixlTelemetryEvent &event) = 0;
+};
+
+#endif // _TELEMETRY_EXPORTER_H

src/api/cpp/telemetry/telemetry_plugin.h

@@ -0,0 +1,101 @@
+/*
+ * 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_PLUGIN_H
+#define __TELEMETRY_PLUGIN_H
+
+#include "telemetry/telemetry_exporter.h"
+//  #include "common/nixl_log.h"
+
+// Forward declarations for special engine types
+//  class nixlUcxEngine;
+
+// Define the plugin API version
+#define NIXL_TELEMETRY_PLUGIN_API_VERSION 1
+
+// Define the plugin interface class
+class nixlTelemetryPlugin {
+public:
+    int api_version;
+
+    // Function pointer for creating a new exporter instance
+    nixlTelemetryExporter *(*create_exporter)(const nixlTelemetryExporterInitParams *init_params);
+
+    // Function pointer for destroying an exporter instance
+    void (*destroy_exporter)(nixlTelemetryExporter *exporter);
+
+    // Function to get the plugin name
+    const char *(*get_plugin_name)();
+
+    // Function to get the plugin version
+    const char *(*get_plugin_version)();
+};
+
+// Macro to define exported C functions for the plugin
+#define NIXL_TELEMETRY_PLUGIN_EXPORT __attribute__((visibility("default")))
+
+// Template for creating backend plugins with minimal boilerplate
+template<typename ExporterType> class nixlTelemetryPluginCreator {
+public:
+    static nixlTelemetryPlugin *
+    create(int api_version, const char *name, const char *version) {
+
+        static const char *plugin_name = name;
+        static const char *plugin_version = version;
+
+        static nixlTelemetryPlugin plugin_instance = {api_version,
+                                                      createExporter,
+                                                      destroyExporter,
+                                                      []() { return plugin_name; },
+                                                      []() { return plugin_version; }};
+
+        return &plugin_instance;
+    }
+
+private:
+    [[nodiscard]] static nixlTelemetryExporter *
+    createExporter(const nixlTelemetryExporterInitParams *init_params) {
+        try {
+            return new ExporterType(init_params);
+        }
+        catch (const std::exception &e) {
+            NIXL_ERROR << "Failed to create exporter: " << e.what();
+            return nullptr;
+        }
+    }
+
+    static void
+    destroyExporter(nixlTelemetryExporter *exporter) {
+        delete exporter;
+    }
+};
+
+// Plugin must implement these functions for dynamic loading
+// Note: extern "C" is required for dynamic loading to avoid C++ name mangling
+extern "C" {
+// Initialize the plugin
+NIXL_TELEMETRY_PLUGIN_EXPORT
+nixlTelemetryPlugin *
+nixl_telemetry_plugin_init();
+
+// Cleanup the plugin
+NIXL_TELEMETRY_PLUGIN_EXPORT
+void
+nixl_telemetry_plugin_fini();
+}
+
+#endif // __TELEMETRY_PLUGIN_H

src/core/meson.build

@@ -62,6 +62,7 @@ nixl_lib = library('nixl',
                    'nixl_plugin_manager.cpp',
                    'nixl_listener.cpp',
                    'telemetry.cpp',
+                   'telemetry_plugin_manager.cpp',
                    include_directories: [ nixl_inc_dirs, utils_inc_dirs ],
                    link_args: ['-lstdc++fs'],
                    dependencies: nixl_lib_deps,

src/core/telemetry.cpp

@@ -26,6 +26,7 @@
 #include "common/nixl_log.h"
 #include "telemetry.h"
 #include "telemetry_event.h"
+#include "telemetry_plugin_manager.h"
 #include "util.h"
 
 using namespace std::chrono_literals;
@@ -75,23 +76,92 @@ nixlTelemetry::initializeTelemetry() {
         throw std::invalid_argument("Telemetry buffer size cannot be 0");
     }
 
-    NIXL_INFO << "Telemetry enabled, using buffer path: " << full_file_path
-              << " with size: " << buffer_size;
 
-    buffer_ = std::make_unique<sharedRingBuffer<nixlTelemetryEvent>>(
-        full_file_path, true, TELEMETRY_VERSION, buffer_size);
+    // Check if exporter is enabled and which type to use
+    const char *exporter_type_env = std::getenv(TELEMETRY_EXPORTER_VAR);
+    if (exporter_type_env) {
+        // Use plugin-based exporter
+        std::string exporter_type = exporter_type_env;
+
+        NIXL_INFO << "Telemetry exporter enabled, type: " << exporter_type;
+
+        // Prepare initialization parameters for the exporter
+        nixlTelemetryExporterInitParams init_params;
+        init_params.outputPath = std::getenv(TELEMETRY_EXPORTER_OUTPUT_PATH_VAR) ?
+            std::getenv(TELEMETRY_EXPORTER_OUTPUT_PATH_VAR) :
+            full_file_path.string();
+        init_params.eventLimit = buffer_size;
+
+        // Create exporter through plugin manager
+        auto &exporter_manager = nixlTelemetryPluginManager::getInstance();
+        exporter_ = exporter_manager.createExporter(exporter_type, &init_params);
+
+        if (!exporter_) {
+            NIXL_WARN << "Failed to create telemetry exporter '" << exporter_type
+                      << "', falling back to buffer mode";
+            // Fall back to buffer mode
+            buffer_ = std::make_unique<sharedRingBuffer<nixlTelemetryEvent>>(
+                full_file_path, true, TELEMETRY_VERSION, buffer_size);
+            writeTask_.callback_ = [this]() { return writeEventHelper(); };
+        } else {
+            writeTask_.callback_ = [this]() { return telemetryExporterHelper(); };
+        }
+    } else {
+        // Use buffer mode (default)
+        NIXL_INFO << "Telemetry enabled, using buffer path: " << full_file_path
+                  << " with size: " << buffer_size;
+
+        buffer_ = std::make_unique<sharedRingBuffer<nixlTelemetryEvent>>(
+            full_file_path, true, TELEMETRY_VERSION, buffer_size);
+
+        writeTask_.callback_ = [this]() { return writeEventHelper(); };
+    }
 
     auto run_interval = std::getenv(TELEMETRY_RUN_INTERVAL_VAR) ?
         std::chrono::milliseconds(std::stoul(std::getenv(TELEMETRY_RUN_INTERVAL_VAR))) :
         DEFAULT_TELEMETRY_RUN_INTERVAL;
 
     // Update write task interval and start it
-    writeTask_.callback_ = [this]() { return writeEventHelper(); };
+
     writeTask_.interval_ = run_interval;
     writeTask_.enabled_ = true;
     registerPeriodicTask(writeTask_);
 }
 
+bool
+nixlTelemetry::telemetryExporterHelper() {
+    std::vector<nixlTelemetryEvent> next_queue;
+    next_queue.reserve(exporter_->getEventLimit());
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        events_.swap(next_queue);
+    }
+    for (auto &event : next_queue) {
+        // if full, ignore
+        exporter_->exportEvent(event);
+    }
+    // collect all events and sort them by timestamp
+    std::vector<nixlTelemetryEvent> all_events;
+    for (auto &backend : backendMap_) {
+        auto backend_events = backend.second->getTelemetryEvents();
+        for (auto &event : backend_events) {
+            // don't trust enum value coming from backend,
+            // as it might be different from the one in agent
+            event.category_ = nixl_telemetry_category_t::NIXL_TELEMETRY_BACKEND;
+            all_events.push_back(event);
+        }
+    }
+    std::sort(all_events.begin(),
+              all_events.end(),
+              [](const nixlTelemetryEvent &a, const nixlTelemetryEvent &b) {
+                  return a.timestampUs_ < b.timestampUs_;
+              });
+    for (auto &event : all_events) {
+        exporter_->exportEvent(event);
+    }
+    return true;
+}
+
 bool
 nixlTelemetry::writeEventHelper() {
     std::vector<nixlTelemetryEvent> next_queue;

src/core/telemetry.h

@@ -19,6 +19,7 @@
 
 #include "common/cyclic_buffer.h"
 #include "telemetry_event.h"
+#include "telemetry/telemetry_exporter.h"
 #include "mem_section.h"
 #include "nixl_types.h"
 
@@ -81,8 +82,11 @@ class nixlTelemetry {
     updateData(const std::string &event_name, nixl_telemetry_category_t category, uint64_t value);
     bool
     writeEventHelper();
+    bool
+    telemetryExporterHelper();
     std::unique_ptr<sharedRingBuffer<nixlTelemetryEvent>> buffer_;
     std::vector<nixlTelemetryEvent> events_;
+    std::unique_ptr<nixlTelemetryExporter> exporter_;
     std::mutex mutex_;
     asio::thread_pool pool_;
     periodicTask writeTask_;