Skip to content

telemetry/doca: share one timestamp per export batch (NIX-1625) #1952

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
e-eygin wants to merge 4 commits into
base: main
Choose a base branch
from
Open

telemetry/doca: share one timestamp per export batch (NIX-1625) #1952

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8302691
telemetry/doca: share one timestamp per export batch (NIX-1625)
e-eygin Jul 16, 2026
b18b7ae
telemetry/doca: address review - lowerCamelCase batchScope, nesting-s…
e-eygin Jul 16, 2026
b555158
telemetry/doca: document withBatch and tighten nested-batch test
e-eygin Jul 16, 2026
2cca182
telemetry: add @tparam/@param tags to withBatch Doxygen
e-eygin Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions src/api/cpp/telemetry/telemetry_exporter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

#include <stdexcept>
#include <string>
#include <utility>

inline constexpr char telemetryExporterVar[] = "NIXL_TELEMETRY_EXPORTER";

Expand Down Expand Up @@ -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.
*/
Comment thread
coderabbitai[bot] marked this conversation as resolved.
template<typename Body>
void
withBatch(Body &&body) {
const batchScope scope{*this};
std::forward<Body>(body)();
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

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_;
};

Expand Down
14 changes: 8 additions & 6 deletions src/core/telemetry/telemetry.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
45 changes: 34 additions & 11 deletions src/plugins/telemetry/doca/doca_exporter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,32 +356,53 @@ 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();
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

doca_error_t
nixlTelemetryDocaExporter::appendCounterSample(const nixlTelemetryEvent &event,
uint64_t timestamp,
const char *counter_name,
const char *label_values[]) {
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
// 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,
Expand All @@ -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
}

Expand Down Expand Up @@ -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;
Expand All @@ -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;
}
Expand All @@ -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;
}
Expand Down
18 changes: 17 additions & 1 deletion src/plugins/telemetry/doca/doca_exporter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@
#include "nixl_types.h"

#include <doca_error.h>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>

struct DocaSharedContext;
Expand All @@ -45,21 +47,35 @@ class nixlTelemetryDocaExporter : public nixlTelemetryExporter {
private:
const std::string agent_name_;
std::shared_ptr<DocaSharedContext> ctx_;
bool inBatch_ = false;
std::optional<uint64_t> 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[]);

Expand Down
Loading
Loading