agents: implement custom OTLP metrics serializer

Introduce custom OTLP metrics serialization in gRPC agent,
replacing SDK with BatchedMetricData and direct protobuf
generation. Provides per-data-point timestamps, custom batching,
and improved performance for high-frequency metrics.

Add debug logging, fix thread ID type mismatches, implement
robust batch processing for partial metrics, and update test
expectations to resolve inconsistent processing and hangs.

Author: santigimeno

agents/grpc/src/grpc_agent.cc

@@ -265,9 +265,15 @@ void PopulateMetricsEvent(grpcagent::MetricsEvent* metrics_event,
     otlp::fill_env_metrics(metrics, env_metrics_stor, false);
   }
 
+  auto batched = metrics.DumpMetricsAndReset();
+  std::vector<opentelemetry::sdk::metrics::MetricData> metric_data;
+  for (const auto& bm : batched) {
+    metric_data.push_back(otlp::BatchedMetricToMetricData(bm));
+  }
+
   data.scope_metric_data_ =
     std::vector<ScopeMetrics>{{ otlp::GetScope(),
-                                metrics.DumpMetricsAndReset() }};
+                                std::move(metric_data) }};
   OtlpMetricUtils::PopulateResourceMetrics(
     data, metrics_event->mutable_body()->mutable_resource_metrics()->Add());
 }
@@ -946,11 +952,7 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
   if (agent->thr_metrics_batch_.ShouldFlush()) {
     auto metric_data = agent->thr_metrics_batch_.DumpMetricsAndReset();
     if (!metric_data.empty()) {
-      ResourceMetrics data;
-      data.resource_ = otlp::GetResource();
-      data.scope_metric_data_ =
-        std::vector<ScopeMetrics>{{ otlp::GetScope(), std::move(metric_data) }};
-      agent->metrics_exporter_->enqueue(std::move(data));
+      agent->metrics_exporter_->enqueue(std::move(metric_data));
     }
   }
 }
@@ -969,24 +971,15 @@ void GrpcAgent::on_metrics_timer() {
   if (metrics_paused_) {
     auto metric_data = proc_metrics_batch_.DumpMetricsAndReset();
     if (!metric_data.empty()) {
-      ResourceMetrics data;
-      data.resource_ = otlp::GetResource();
-      data.scope_metric_data_ =
-        std::vector<ScopeMetrics>{{ otlp::GetScope(), std::move(metric_data) }};
-      metrics_exporter_->enqueue(std::move(data));
+      metrics_exporter_->enqueue(std::move(metric_data));
     }
 
     metric_data = thr_metrics_batch_.DumpMetricsAndReset();
     if (!metric_data.empty()) {
-      ResourceMetrics data;
-      data.resource_ = otlp::GetResource();
-      data.scope_metric_data_ =
-        std::vector<ScopeMetrics>{{ otlp::GetScope(), std::move(metric_data) }};
-      metrics_exporter_->enqueue(std::move(data));
+      metrics_exporter_->enqueue(std::move(metric_data));
     }
 
     metrics_exporter_->flush();
-    return;
   }
 
   got_proc_metrics();
@@ -1459,11 +1452,7 @@ void GrpcAgent::got_proc_metrics() {
   if (proc_metrics_batch_.ShouldFlush()) {
     auto metric_data = proc_metrics_batch_.DumpMetricsAndReset();
     if (!metric_data.empty()) {
-      ResourceMetrics data;
-      data.resource_ = otlp::GetResource();
-      data.scope_metric_data_ =
-        std::vector<ScopeMetrics>{{ otlp::GetScope(), std::move(metric_data) }};
-      metrics_exporter_->enqueue(std::move(data));
+      metrics_exporter_->enqueue(std::move(metric_data));
     }
   }
 }

agents/grpc/src/grpc_metrics_exporter.cc

@@ -1,6 +1,7 @@
 #include "grpc_metrics_exporter.h"
 #include "grpc_client.h"
 
+#include "../../otlp/src/otlp_common.h"
 #include "opentelemetry/exporters/otlp/otlp_grpc_client.h"
 #include "opentelemetry/exporters/otlp/otlp_metric_utils.h"
 
@@ -52,7 +53,8 @@ void GrpcMetricsExporter::init() {
 GrpcMetricsExporter::~GrpcMetricsExporter() {
 }
 
-void GrpcMetricsExporter::enqueue(ResourceMetrics&& metrics) {
+void GrpcMetricsExporter::enqueue(
+    std::vector<otlp::BatchedMetricData>&& metrics) {
   metrics_q_.push(std::move(metrics));
   if (!in_flight_) {
     export_current();
@@ -78,7 +80,8 @@ void GrpcMetricsExporter::export_current() {
 
   auto* request =
     google::protobuf::Arena::Create<ExportMetricsServiceRequest>(arena.get());
-  OtlpMetricUtils::PopulateRequest(data, request);
+
+  otlp::PopulateRequest(data, otlp::GetResource(), otlp::GetScope(), request);
 
   auto context = OtlpGrpcClient::MakeClientContext(options_);
   ::grpc::Status immediate =

agents/grpc/src/grpc_metrics_exporter.h

@@ -5,6 +5,7 @@
 #include <memory>
 #include <string>
 
+#include "../../otlp/src/batched_metric_data.h"
 #include "nsolid/async_ts_queue.h"
 #include "nsolid/nsolid_util.h"
 #include "opentelemetry/exporters/otlp/otlp_grpc_client.h"
@@ -45,7 +46,7 @@ class GrpcMetricsExporter:
 
   void init();
 
-  void enqueue(opentelemetry::sdk::metrics::ResourceMetrics&& metrics);
+  void enqueue(std::vector<otlp::BatchedMetricData>&& metrics);
 
   void flush();
 
@@ -63,7 +64,7 @@ class GrpcMetricsExporter:
   opentelemetry::v1::exporter::otlp::OtlpGrpcMetricExporterOptions options_;
   std::shared_ptr<opentelemetry::v1::exporter::otlp::OtlpGrpcClient> client_;
   std::unique_ptr<MetricsServiceStub> metrics_service_stub_;
-  utils::RingBuffer<opentelemetry::sdk::metrics::ResourceMetrics> metrics_q_;
+  utils::RingBuffer<std::vector<otlp::BatchedMetricData>> metrics_q_;
   bool in_flight_;
   std::shared_ptr<AsyncTSQueue<::grpc::Status>> metrics_completion_q_;
 };

agents/otlp/src/batched_metric_data.h

@@ -0,0 +1,51 @@
+#ifndef AGENTS_OTLP_SRC_BATCHED_METRIC_DATA_H_
+#define AGENTS_OTLP_SRC_BATCHED_METRIC_DATA_H_
+
+#include <vector>
+
+#include "opentelemetry/nostd/variant.h"
+#include "opentelemetry/sdk/common/attribute_utils.h"
+#include "opentelemetry/sdk/metrics/data/point_data.h"
+#include "opentelemetry/sdk/metrics/instruments.h"
+#include "opentelemetry/version.h"
+
+namespace node {
+namespace nsolid {
+namespace otlp {
+
+using opentelemetry::sdk::metrics::SumPointData;
+using opentelemetry::sdk::metrics::HistogramPointData;
+using opentelemetry::sdk::metrics::Base2ExponentialHistogramPointData;
+using opentelemetry::sdk::metrics::LastValuePointData;
+using opentelemetry::sdk::metrics::DropPointData;
+using opentelemetry::sdk::metrics::SummaryPointData;
+using opentelemetry::sdk::metrics::AggregationTemporality;
+using opentelemetry::sdk::metrics::InstrumentDescriptor;
+using PointAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
+using PointType =
+    opentelemetry::nostd::variant<SumPointData,
+                                  HistogramPointData,
+                                  Base2ExponentialHistogramPointData,
+                                  LastValuePointData,
+                                  DropPointData,
+                                  SummaryPointData>;
+
+struct TimedPointDataAttributes {
+  opentelemetry::common::SystemTimestamp start_ts;
+  opentelemetry::common::SystemTimestamp end_ts;
+  PointAttributes attributes;
+  PointType point_data;
+};
+
+class BatchedMetricData {
+ public:
+  InstrumentDescriptor instrument_descriptor;
+  AggregationTemporality aggregation_temporality;
+  std::vector<TimedPointDataAttributes> point_data_attr_;
+};
+
+}  // namespace otlp
+}  // namespace nsolid
+}  // namespace node
+
+#endif  // AGENTS_OTLP_SRC_BATCHED_METRIC_DATA_H_