diff --git a/src/api/cpp/telemetry/telemetry_exporter.h b/src/api/cpp/telemetry/telemetry_exporter.h index 274e61a210..e18ab950c6 100644 --- a/src/api/cpp/telemetry/telemetry_exporter.h +++ b/src/api/cpp/telemetry/telemetry_exporter.h @@ -22,6 +22,7 @@ #include #include +#include inline constexpr char telemetryExporterVar[] = "NIXL_TELEMETRY_EXPORTER"; @@ -75,7 +76,58 @@ class nixlTelemetryExporter { virtual nixl_status_t exportEvent(const nixlTelemetryEvent &event) = 0; + /** + * @brief Runs @p body with a batch open around it. + * + * Exporters may use the batch boundary to share per-batch work across the + * exportEvent() calls made inside @p body (e.g. a single timestamp). The + * batch begins before @p body runs and ends when it returns (including on + * exception). Scopes may nest; the lifecycle hooks fire only for the + * outermost scope, so a nested call reuses the surrounding batch. + * + * @tparam Body Callable invocable with no arguments. + * @param body Callable run once while the batch is open. + */ + template + void + withBatch(Body &&body) { + const batchScope scope{*this}; + std::forward(body)(); + } + private: + class batchScope { + public: + explicit batchScope(nixlTelemetryExporter &exporter) noexcept : exporter_(exporter) { + if (exporter_.batchDepth_++ == 0) { + exporter_.onBatchBegin(); + } + } + + ~batchScope() { + if (--exporter_.batchDepth_ == 0) { + exporter_.onBatchEnd(); + } + } + + batchScope(const batchScope &) = delete; + batchScope(batchScope &&) = delete; + batchScope & + operator=(const batchScope &) = delete; + batchScope & + operator=(batchScope &&) = delete; + + private: + nixlTelemetryExporter &exporter_; + }; + + virtual void + onBatchBegin() noexcept {} + + virtual void + onBatchEnd() noexcept {} + + unsigned batchDepth_ = 0; const size_t maxEventsBuffered_; }; diff --git a/src/core/telemetry/telemetry.cpp b/src/core/telemetry/telemetry.cpp index 7dab1b8177..392a39936d 100644 --- a/src/core/telemetry/telemetry.cpp +++ b/src/core/telemetry/telemetry.cpp @@ -279,13 +279,15 @@ nixlTelemetry::flushPendingEvents() { events_.swap(next_queue); } - for (auto &event : next_queue) { - // Deactivated metrics never enter the queue (filtered at the source), so - // everything here is already exportable. - exporter_->exportEvent(event); - } + exporter_->withBatch([&] { + for (auto &event : next_queue) { + // Deactivated metrics never enter the queue (filtered at the source), + // so everything here is already exportable. + exporter_->exportEvent(event); + } - exportDroppedEvents(); + exportDroppedEvents(); + }); return true; } diff --git a/src/plugins/telemetry/doca/doca_exporter.cpp b/src/plugins/telemetry/doca/doca_exporter.cpp index 0e55e7f2da..76ef0560ae 100644 --- a/src/plugins/telemetry/doca/doca_exporter.cpp +++ b/src/plugins/telemetry/doca/doca_exporter.cpp @@ -356,8 +356,32 @@ nixlTelemetryDocaExporter::~nixlTelemetryDocaExporter() { ctx_.reset(); } +uint64_t +nixlTelemetryDocaExporter::currentTimestamp() noexcept { + if (!inBatch_) { + return docaTimestamp(); + } + if (!batchTimestamp_.has_value()) { + batchTimestamp_ = docaTimestamp(); + } + return *batchTimestamp_; +} + +void +nixlTelemetryDocaExporter::onBatchBegin() noexcept { + inBatch_ = true; + batchTimestamp_.reset(); +} + +void +nixlTelemetryDocaExporter::onBatchEnd() noexcept { + inBatch_ = false; + batchTimestamp_.reset(); +} + doca_error_t nixlTelemetryDocaExporter::appendCounterSample(const nixlTelemetryEvent &event, + uint64_t timestamp, const char *counter_name, const char *label_values[]) { #pragma GCC diagnostic push @@ -365,23 +389,20 @@ nixlTelemetryDocaExporter::appendCounterSample(const nixlTelemetryEvent &event, // Counter events carry a per-operation delta; increment so the exported // counter is a monotonic cumulative total (add_counter would instead push // each delta as an absolute value, yielding a non-monotonic series). - return doca_telemetry_exporter_metrics_add_counter_increment(ctx_->source, - docaTimestamp(), - counter_name, - event.value_, - ctx_->label_set_id, - label_values); + return doca_telemetry_exporter_metrics_add_counter_increment( + ctx_->source, timestamp, counter_name, event.value_, ctx_->label_set_id, label_values); #pragma GCC diagnostic pop } doca_error_t nixlTelemetryDocaExporter::appendErrorCounterSample(const nixlTelemetryEvent &event, + uint64_t timestamp, const char *status) { #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wdeprecated-declarations" const char *label_values[] = {agent_name_.c_str(), status}; return doca_telemetry_exporter_metrics_add_counter_increment(ctx_->source, - docaTimestamp(), + timestamp, "agent_errors_total", event.value_, ctx_->error_label_set_id, @@ -391,12 +412,13 @@ nixlTelemetryDocaExporter::appendErrorCounterSample(const nixlTelemetryEvent &ev doca_error_t nixlTelemetryDocaExporter::appendGaugeSample(const nixlTelemetryEvent &event, + uint64_t timestamp, const char *metric_name, const char *label_values[]) { #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wdeprecated-declarations" return doca_telemetry_exporter_metrics_add_gauge( - ctx_->source, docaTimestamp(), metric_name, event.value_, ctx_->label_set_id, label_values); + ctx_->source, timestamp, metric_name, event.value_, ctx_->label_set_id, label_values); #pragma GCC diagnostic pop } @@ -426,10 +448,11 @@ nixlTelemetryDocaExporter::exportEvent(const nixlTelemetryEvent &event) { try { const std::lock_guard lock(g_metrics_mutex); const char *label_values[] = {agent_name_.c_str()}; + const uint64_t ts = currentTimestamp(); if (const char *const status = nixlEnumStrings::telemetryErrorStatusLabel(event.eventType_)) { - const auto result = appendErrorCounterSample(event, status); + const auto result = appendErrorCounterSample(event, ts, status); if (result != DOCA_SUCCESS) { NIXL_ERROR << "Failed to add error counter: " << result; return NIXL_ERR_UNKNOWN; @@ -442,7 +465,7 @@ nixlTelemetryDocaExporter::exportEvent(const nixlTelemetryEvent &event) { // Idempotent gauge first, non-idempotent counter increment last. doca_error_t gauge_result = DOCA_SUCCESS; if (descriptor.gaugeName != nullptr) { - gauge_result = appendGaugeSample(event, descriptor.gaugeName, label_values); + gauge_result = appendGaugeSample(event, ts, descriptor.gaugeName, label_values); if (gauge_result != DOCA_SUCCESS) { NIXL_ERROR << "Failed to add gauge: " << gauge_result; } @@ -458,7 +481,7 @@ nixlTelemetryDocaExporter::exportEvent(const nixlTelemetryEvent &event) { doca_error_t counter_result = DOCA_SUCCESS; if (descriptor.counterName != nullptr) { - counter_result = appendCounterSample(event, descriptor.counterName, label_values); + counter_result = appendCounterSample(event, ts, descriptor.counterName, label_values); if (counter_result != DOCA_SUCCESS) { NIXL_ERROR << "Failed to add counter: " << counter_result; } diff --git a/src/plugins/telemetry/doca/doca_exporter.h b/src/plugins/telemetry/doca/doca_exporter.h index 1eee71e615..630f412fd2 100644 --- a/src/plugins/telemetry/doca/doca_exporter.h +++ b/src/plugins/telemetry/doca/doca_exporter.h @@ -22,7 +22,9 @@ #include "nixl_types.h" #include +#include #include +#include #include struct DocaSharedContext; @@ -45,21 +47,35 @@ class nixlTelemetryDocaExporter : public nixlTelemetryExporter { private: const std::string agent_name_; std::shared_ptr ctx_; + bool inBatch_ = false; + std::optional batchTimestamp_; + + void + onBatchBegin() noexcept override; + void + onBatchEnd() noexcept override; + + [[nodiscard]] uint64_t + currentTimestamp() noexcept; // Appends a counter sample to the time-series. DOCA accumulates the value // (add_counter_increment), so repeated per-operation deltas produce a // monotonic cumulative total, matching the Prometheus exporter. [[nodiscard]] doca_error_t appendCounterSample(const nixlTelemetryEvent &event, + uint64_t timestamp, const char *counter_name, const char *label_values[]); [[nodiscard]] doca_error_t - appendErrorCounterSample(const nixlTelemetryEvent &event, const char *status); + appendErrorCounterSample(const nixlTelemetryEvent &event, + uint64_t timestamp, + const char *status); // Appends a gauge sample to the time-series (absolute last-operation value). [[nodiscard]] doca_error_t appendGaugeSample(const nixlTelemetryEvent &event, + uint64_t timestamp, const char *metric_name, const char *label_values[]); diff --git a/test/doca-telemetry/telemetry_doca_nixl_test.cpp b/test/doca-telemetry/telemetry_doca_nixl_test.cpp index 65dc1df170..f05aa60cfd 100644 --- a/test/doca-telemetry/telemetry_doca_nixl_test.cpp +++ b/test/doca-telemetry/telemetry_doca_nixl_test.cpp @@ -21,6 +21,7 @@ #include "telemetry_event.h" #include +#include #include #include #include @@ -28,6 +29,7 @@ #include #include #include +#include #include @@ -494,6 +496,186 @@ TEST_F(docaNixlExporterTest, DISABLED_IpcDeliversToLiveDts) { << " -- IPC delivery or DTS Prometheus re-export failed"; } +// All samples exported inside one withBatch scope must carry a single shared +// timestamp, even when wall-clock time advances between events. +// The events are spaced apart by more than DOCA's millisecond exposition +// resolution, so per-event clock reads (the pre-batch behavior) would stamp them +// with different timestamps; a shared batch timestamp keeps them identical. +// Histogram series are timestamped at flush (not per observe), so they are +// excluded. +TEST_F(docaNixlExporterTest, BatchSharesOneTimestampAcrossEvents) { + constexpr char agentName[] = "nixl_doca_batch_ts_test"; + const nixlTelemetryExporterInitParams params{agentName, 4096}; + nixlTelemetryDocaExporter exporter(params); + + constexpr std::array types{ + nixl_telemetry_event_type_t::AGENT_TX_BYTES, + nixl_telemetry_event_type_t::AGENT_RX_BYTES, + nixl_telemetry_event_type_t::AGENT_MEMORY_REGISTERED, + nixl_telemetry_event_type_t::AGENT_ERR_INVALID_PARAM, + }; + exporter.withBatch([&] { + for (const auto type : types) { + EXPECT_EQ(exporter.exportEvent(nixlTelemetryEvent(type, 100)), NIXL_SUCCESS); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + }); + ASSERT_EQ(exporter.flush(), NIXL_SUCCESS); + + const nixl::doca_test::labelSet errLabels{{"agent_name", agentName}, + {"status", "invalid_param"}}; + const auto metrics = + scrapeUntilValue(port_, "agent_errors_total", 100.0, std::chrono::seconds(12), errLabels); + + const auto endsWith = [](const std::string &s, const std::string &suffix) { + return s.size() >= suffix.size() && + s.compare(s.size() - suffix.size(), suffix.size(), suffix) == 0; + }; + + std::optional shared; + size_t counted = 0; + for (const auto &[id, seriesSamples] : metrics.series()) { + const auto agent = id.labels.find("agent_name"); + if (agent == id.labels.end() || agent->second != agentName || seriesSamples.empty()) { + continue; + } + if (endsWith(id.name, "_bucket") || endsWith(id.name, "_sum") || + endsWith(id.name, "_count")) { + continue; + } + const std::optional ts = seriesSamples.back().timestamp; + ASSERT_TRUE(ts.has_value()) << id.name << " must carry a timestamp in the exposition"; + if (!shared.has_value()) { + shared = ts; + } + EXPECT_EQ(*ts, *shared) << id.name << " must share the single batch timestamp"; + ++counted; + } + EXPECT_GE(counted, 2u) << "expected several non-histogram series produced within the batch"; +} + +// Without a surrounding withBatch each exportEvent stamps its samples with a +// fresh clock read: two events spaced past DOCA's millisecond exposition +// resolution must land on distinct timestamps (the counterpart to the shared +// batch timestamp above), and their exposed values must be correct. +TEST_F(docaNixlExporterTest, StandaloneExportStampsEachEventFreshly) { + constexpr char agentName[] = "nixl_doca_standalone_ts_test"; + const nixlTelemetryExporterInitParams params{agentName, 4096}; + nixlTelemetryDocaExporter exporter(params); + + const std::string txCounter = + nixlEnumStrings::telemetryMetricDescriptor(nixl_telemetry_event_type_t::AGENT_TX_BYTES) + .counterName; + const std::string rxCounter = + nixlEnumStrings::telemetryMetricDescriptor(nixl_telemetry_event_type_t::AGENT_RX_BYTES) + .counterName; + const nixl::doca_test::labelSet labels{{"agent_name", agentName}}; + + ASSERT_EQ( + exporter.exportEvent(nixlTelemetryEvent(nixl_telemetry_event_type_t::AGENT_TX_BYTES, 42)), + NIXL_SUCCESS); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + ASSERT_EQ( + exporter.exportEvent(nixlTelemetryEvent(nixl_telemetry_event_type_t::AGENT_RX_BYTES, 7)), + NIXL_SUCCESS); + ASSERT_EQ(exporter.flush(), NIXL_SUCCESS); + + const auto metrics = scrapeUntilValue(port_, rxCounter, 7.0, std::chrono::seconds(12), labels); + EXPECT_EQ(metrics.latestValue(txCounter, labels), std::optional(42.0)); + EXPECT_EQ(metrics.latestValue(rxCounter, labels), std::optional(7.0)); + + const auto seriesTimestamp = [&metrics](const std::string &name, + const nixl::doca_test::labelSet &where) { + std::optional ts; + for (const auto &[id, seriesSamples] : metrics.series()) { + if (id.name != name || seriesSamples.empty()) { + continue; + } + const auto agent = id.labels.find("agent_name"); + if (agent != id.labels.end() && agent->second == where.at("agent_name")) { + ts = seriesSamples.back().timestamp; + } + } + return ts; + }; + + const std::optional txTs = seriesTimestamp(txCounter, labels); + const std::optional rxTs = seriesTimestamp(rxCounter, labels); + ASSERT_TRUE(txTs.has_value() && rxTs.has_value()); + EXPECT_NE(*txTs, *rxTs) << "standalone events spaced apart must get independent timestamps"; +} + +// Nested withBatch scopes must behave as one batch: the exporter's batch hooks +// fire only for the outermost scope, so an inner scope's exit must not reset the +// shared timestamp. Events emitted before, inside, and after the nested scope -- +// spaced past DOCA's millisecond exposition resolution -- must all carry the one +// outer-batch timestamp. +TEST_F(docaNixlExporterTest, NestedBatchScopesShareOuterTimestamp) { + constexpr char agentName[] = "nixl_doca_nested_batch_ts_test"; + const nixlTelemetryExporterInitParams params{agentName, 4096}; + nixlTelemetryDocaExporter exporter(params); + + exporter.withBatch([&] { + EXPECT_EQ(exporter.exportEvent( + nixlTelemetryEvent(nixl_telemetry_event_type_t::AGENT_TX_BYTES, 100)), + NIXL_SUCCESS); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + exporter.withBatch([&] { + EXPECT_EQ(exporter.exportEvent( + nixlTelemetryEvent(nixl_telemetry_event_type_t::AGENT_RX_BYTES, 100)), + NIXL_SUCCESS); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + }); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + EXPECT_EQ(exporter.exportEvent(nixlTelemetryEvent( + nixl_telemetry_event_type_t::AGENT_MEMORY_REGISTERED, 100)), + NIXL_SUCCESS); + }); + ASSERT_EQ(exporter.flush(), NIXL_SUCCESS); + + const nixl::doca_test::labelSet labels{{"agent_name", agentName}}; + const auto metrics = scrapeUntilValue( + port_, "agent_memory_registered_last_bytes", 100.0, std::chrono::seconds(12), labels); + + const auto seriesTimestamp = [&](const std::string &name) -> std::optional { + for (const auto &[id, seriesSamples] : metrics.series()) { + if (id.name != name || seriesSamples.empty()) { + continue; + } + const auto agent = id.labels.find("agent_name"); + if (agent != id.labels.end() && agent->second == agentName) { + return seriesSamples.back().timestamp; + } + } + return std::nullopt; + }; + + // One series produced from each nesting position: before (outer), inside + // (nested), and after the nested scope closes. Their exports are spaced past + // DOCA's ms resolution, so a per-event or per-scope clock read would give the + // "after" series a later timestamp; the shared outer batch timestamp keeps + // all three identical. + const std::string txCounter = + nixlEnumStrings::telemetryMetricDescriptor(nixl_telemetry_event_type_t::AGENT_TX_BYTES) + .counterName; + const std::string rxCounter = + nixlEnumStrings::telemetryMetricDescriptor(nixl_telemetry_event_type_t::AGENT_RX_BYTES) + .counterName; + const std::string memoryGauge = "agent_memory_registered_last_bytes"; + + const std::optional beforeTs = seriesTimestamp(txCounter); + const std::optional insideTs = seriesTimestamp(rxCounter); + const std::optional afterTs = seriesTimestamp(memoryGauge); + + ASSERT_TRUE(beforeTs.has_value()) << txCounter << " (before nesting) missing or untimestamped"; + ASSERT_TRUE(insideTs.has_value()) << rxCounter << " (inside nesting) missing or untimestamped"; + ASSERT_TRUE(afterTs.has_value()) << memoryGauge << " (after nesting) missing or untimestamped"; + + EXPECT_EQ(*insideTs, *beforeTs) << "nested-scope event must share the outer batch timestamp"; + EXPECT_EQ(*afterTs, *beforeTs) + << "event after the nested scope closes must still share the outer batch timestamp"; +} + int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv);