agents,lib,src: implement OTel metrics API

agents/grpc/src/grpc_agent.cc

@@ -1,5 +1,7 @@
 #include "grpc_agent.h"
 
+#include <algorithm>
+
 #include "asserts-cpp/asserts.h"
 #include "nsolid/nsolid_api.h"
 #include "nsolid/continuous_profiler.h"
@@ -39,6 +41,7 @@ using LogsRecordable = opentelemetry::sdk::logs::Recordable;
 using opentelemetry::sdk::metrics::MetricData;
 using opentelemetry::sdk::metrics::ResourceMetrics;
 using opentelemetry::sdk::metrics::ScopeMetrics;
+using opentelemetry::sdk::instrumentationscope::InstrumentationScope;
 using opentelemetry::sdk::resource::Resource;
 using opentelemetry::sdk::resource::ResourceAttributes;
 using opentelemetry::sdk::trace::Recordable;
@@ -296,8 +299,7 @@ void PopulateMetricsEvent(grpcagent::MetricsEvent* metrics_event,
     otlp::fill_env_metrics(metrics, env_metrics_stor, false);
   }
 
-  data.scope_metric_data_ =
-    std::vector<ScopeMetrics>{{otlp::GetScope(), metrics}};
+  data.scope_metric_data_.push_back({otlp::GetScope(), std::move(metrics)});
   OtlpMetricUtils::PopulateResourceMetrics(
     data, metrics_event->mutable_body()->mutable_resource_metrics()->Add());
 }
@@ -616,7 +618,7 @@ int GrpcAgent::start() {
   return r;
 }
 
-int GrpcAgent::stop(bool profile_stopped) {
+int GrpcAgent::stop(bool profile_stopped, OtelMetricsStor final_metrics) {
   Debug("Stopping gRPC Agent: %d\n", profile_stopped);
   bool expected = false;
   if (exiting_.compare_exchange_strong(expected, true) == false) {
@@ -625,6 +627,10 @@ int GrpcAgent::stop(bool profile_stopped) {
     return 0;
   }
 
+  if (!final_metrics.empty()) {
+    final_otel_metrics_q_.enqueue(std::move(final_metrics));
+  }
+
   if (utils::are_threads_equal(thread_.base(), uv_thread_self())) {
     do_stop();
   } else {
@@ -791,7 +797,14 @@ int GrpcAgent::start_heap_snapshot_from_js(
     return;
   }
 
-  agent->stop(profile_stopped);
+  OtelMetricsStor final_metrics;
+  if (!on_signal) {
+    SharedEnvInst envinst = EnvInst::GetInst(EnvList::Inst()->main_thread_id());
+    if (envinst != nullptr) {
+      final_metrics = envinst->GetOtelMetrics();
+    }
+  }
+  agent->stop(profile_stopped, std::move(final_metrics));
 }
 
 /*static*/ void GrpcAgent::shutdown_cb_(nsuv::ns_async*,
@@ -894,6 +907,7 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
     return;
   }
 
+  queue_final_otel_metrics_(envinst, agent_wp);
   agent->env_msg_q_.enqueue({ envinst, false });
   if (!agent->env_msg_.is_closing()) {
     ASSERT_EQ(0, agent->env_msg_.send());
@@ -910,18 +924,20 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
   std::tuple<SharedEnvInst, bool> tup;
   while (agent->env_msg_q_.dequeue(tup)) {
     SharedEnvInst envinst = std::get<0>(tup);
-    EnvInst::Scope scp(envinst);
-    if (scp.Success()) {
-      bool creation = std::get<1>(tup);
-      if (creation) {
+    const bool creation = std::get<1>(tup);
+    const uint64_t thread_id = GetThreadId(envinst);
+    if (creation) {
+      EnvInst::Scope scp(envinst);
+      if (scp.Success()) {
         auto pair = agent->env_metrics_map_.emplace(
           std::piecewise_construct,
-          std::forward_as_tuple(GetThreadId(envinst)),
+          std::forward_as_tuple(thread_id),
           std::forward_as_tuple(envinst));
         ASSERT(pair.second);
-      } else {
-        agent->env_metrics_map_.erase(GetThreadId(envinst));
       }
+    } else {
+      agent->env_metrics_map_.erase(thread_id);
+      metrics_msg_cb_(nullptr, agent_wp);
     }
   }
 }
@@ -983,23 +999,67 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
     return;
   }
 
-  ResourceMetrics data;
-  auto resource = otlp::GetMetricsResource();
-  data.resource_ = resource.get();
-  std::vector<MetricData> metrics;
+  ThreadMetricsSnapshot snapshot;
+  while (agent->thr_metrics_msg_q_.dequeue(snapshot)) {
+    // A retiring provider's final snapshot supersedes queued periodic copies.
+    if (agent->exiting_ ||
+        EnvInst::GetInst(snapshot.metrics.thread_id) == nullptr) {
+      snapshot.otel_metrics.clear();
+    }
+    agent->export_metrics_(snapshot.otel_metrics, &snapshot.metrics);
+    agent->thr_metrics_cache_.insert_or_assign(snapshot.metrics.thread_id,
+                                               std::move(snapshot.metrics));
+  }
+
+  OtelMetricsStor final_metrics;
+  while (agent->final_otel_metrics_q_.dequeue(final_metrics)) {
+    agent->export_metrics_(final_metrics);
+  }
+}
 
-  ThreadMetricsStor stor;
-  while (agent->thr_metrics_msg_q_.dequeue(stor)) {
-    otlp::fill_env_metrics(metrics, stor, false);
-    agent->thr_metrics_cache_.insert_or_assign(stor.thread_id, std::move(stor));
+void GrpcAgent::export_metrics_(
+    const OtelMetricsStor& instruments,
+    const ThreadMetrics::MetricsStor* thread_metrics) {
+  if (metrics_exporter_ == nullptr ||
+      (thread_metrics == nullptr && instruments.empty())) {
+    return;
   }
 
-  data.scope_metric_data_ =
-    std::vector<ScopeMetrics>{{otlp::GetScope(), metrics}};
-  auto result = agent->metrics_exporter_->Export(data);
+  auto resource = otlp::GetMetricsResource();
+  ResourceMetrics data;
+  data.resource_ = resource.get();
+  if (thread_metrics != nullptr) {
+    std::vector<MetricData> metrics;
+    otlp::fill_env_metrics(metrics, *thread_metrics, false);
+    data.scope_metric_data_.push_back({otlp::GetScope(), std::move(metrics)});
+  }
+
+  // Each environment owns an independent provider and export batch.
+  otlp::OtelMetricsMap otel_metrics;
+  otlp::fill_otel_instruments(otel_metrics, instruments);
+  std::vector<std::unique_ptr<InstrumentationScope>> scopes;
+  for (auto& [identity, scope_metrics] : otel_metrics) {
+    scopes.push_back(InstrumentationScope::Create(std::get<0>(identity),
+                                                  std::get<1>(identity),
+                                                  std::get<2>(identity)));
+    data.scope_metric_data_.push_back({scopes.back().get(),
+                                      std::move(scope_metrics)});
+  }
+  auto result = metrics_exporter_->Export(data);
   Debug("# ThreadMetrics Exported. Result: %d\n", static_cast<int>(result));
 }
 
+/*static*/void GrpcAgent::queue_final_otel_metrics_(SharedEnvInst envinst,
+                                                   WeakGrpcAgent agent_wp) {
+  SharedGrpcAgent agent = agent_wp.lock();
+  if (agent == nullptr || agent->exiting_) return;
+
+  OtelMetricsStor snapshot = envinst->GetOtelMetrics();
+  if (!snapshot.empty()) {
+    agent->final_otel_metrics_q_.enqueue(std::move(snapshot));
+  }
+}
+
 /*static*/void GrpcAgent::metrics_timer_cb_(nsuv::ns_timer*,
                                             WeakGrpcAgent agent_wp) {
   SharedGrpcAgent agent = agent_wp.lock();
@@ -1040,7 +1100,16 @@ void GrpcAgent::env_deletion_cb_(SharedEnvInst envinst,
     return;
   }
 
-  if (agent->thr_metrics_msg_q_.enqueue(metrics->Get()) == 1) {
+  OtelMetricsStor otel_metrics;
+  SharedEnvInst envinst = EnvInst::GetInst(metrics->thread_id());
+  if (envinst != nullptr) {
+    otel_metrics = envinst->GetOtelMetrics();
+  }
+  ThreadMetricsSnapshot snapshot = {
+    metrics->Get(),
+    std::move(otel_metrics),
+  };
+  if (agent->thr_metrics_msg_q_.enqueue(std::move(snapshot)) == 1) {
     ASSERT_EQ(0, uv_async_send(&agent->metrics_msg_));
   }
 }
@@ -1347,6 +1416,11 @@ void GrpcAgent::do_start() {
 }
 
 void GrpcAgent::do_stop() {
+  // Drain the last synchronous snapshots before closing the exporter.
+  metrics_msg_cb_(nullptr, weak_from_this());
+  if (metrics_exporter_ != nullptr) {
+    USE(metrics_exporter_->ForceFlush(DEFAULT_GRPC_TIMEOUT));
+  }
   // Don't try to send the exit event when grpc connection wasn't authorized or
   // the agent wasn't fully configured.
   if (!unauthorized_ && nsolid_service_stub_) {
@@ -1440,8 +1514,7 @@ void GrpcAgent::got_proc_metrics() {
   ResourceMetrics data;
   auto resource = otlp::GetMetricsResource();
   data.resource_ = resource.get();
-  data.scope_metric_data_ =
-    std::vector<ScopeMetrics>{{otlp::GetScope(), metrics}};
+  data.scope_metric_data_.push_back({otlp::GetScope(), std::move(metrics)});
   auto result = metrics_exporter_->Export(data);
   Debug("# ProcessMetrics Exported. Result: %d\n", static_cast<int>(result));
   proc_prev_stor_ = stor;

agents/grpc/src/grpc_agent.h

@@ -1,8 +1,9 @@
 #ifndef AGENTS_GRPC_SRC_GRPC_AGENT_H_
 #define AGENTS_GRPC_SRC_GRPC_AGENT_H_
 
-#include <nsolid.h>
 #include <nsolid/async_ts_queue.h>
+#include <nsolid/nsolid_api.h>
+#include <nsolid/nsolid_otel_metrics_types.h>
 #include <nsolid/thread_safe.h>
 #include <memory>
 #include "grpcpp/grpcpp.h"
@@ -101,7 +102,7 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
 
   int start_heap_snapshot_from_js(const grpcagent::CommandRequest& req);
 
-  int stop(bool profile_stopped = false);
+  int stop(bool profile_stopped = false, OtelMetricsStor final_metrics = {});
 
   const std::string& agent_id() const { return agent_id_; }
 
@@ -197,6 +198,10 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
   static void start_profiling_msg_cb(nsuv::ns_async*, WeakGrpcAgent);
 
   static void thr_metrics_cb_(SharedThreadMetrics, WeakGrpcAgent);
+  static void queue_final_otel_metrics_(SharedEnvInst, WeakGrpcAgent);
+
+  void export_metrics_(const OtelMetricsStor&,
+                       const ThreadMetrics::MetricsStor* = nullptr);
 
   static void trace_hook_(Tracer*, const Tracer::SpanStor&, WeakGrpcAgent);
 
@@ -300,6 +305,11 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
   void update_tracer(uint32_t flags);
 
  private:
+  struct ThreadMetricsSnapshot {
+    ThreadMetrics::MetricsStor metrics;
+    OtelMetricsStor otel_metrics;
+  };
+
   uv_loop_t loop_;
   nsuv::ns_thread thread_;
   nsuv::ns_async shutdown_;
@@ -333,7 +343,8 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
   ProcessMetrics::MetricsStor proc_prev_stor_;
   std::map<uint64_t, JSThreadMetrics> env_metrics_map_;
   nsuv::ns_async metrics_msg_;
-  TSQueue<ThreadMetrics::MetricsStor> thr_metrics_msg_q_;
+  TSQueue<ThreadMetricsSnapshot> thr_metrics_msg_q_;
+  TSQueue<OtelMetricsStor> final_otel_metrics_q_;
   nsuv::ns_timer metrics_timer_;
   std::unique_ptr<opentelemetry::v1::exporter::otlp::OtlpGrpcMetricExporter>
     metrics_exporter_;

agents/otlp/src/metrics_exporter.h

@@ -5,6 +5,7 @@
 #include <vector>
 
 #include "nsolid.h"
+#include "nsolid/nsolid_otel_metrics_types.h"
 
 namespace node {
 namespace nsolid {
@@ -25,6 +26,9 @@ class MetricsExporter {
       const ProcessMetrics::MetricsStor& prev_stor) = 0;
 
   virtual void got_thr_metrics(const std::vector<ThrMetricsStor>&) = 0;
+
+  // Vendor-specific adapters only support the built-in runtime metrics.
+  virtual void got_otel_metrics(const OtelMetricsStor&) {}
 };
 
 }  // namespace otlp