agents: add nsolid-version and add it to metadata

Also, cache the metadata so it's not generated multiple times per
different rpc's.
Finally, resolve having both `nsolid-saas` and `nsolid-saas-token`
header name and settle for the first.

Author: santigimeno

agents/grpc/src/asset_stream.cc

@@ -17,15 +17,11 @@ AssetStream::AssetStream(
     NSolidService::StubInterface* stub,
     AssetStor&& stor,
     std::weak_ptr<AssetStreamObserver> observer,
-    const std::string& agent_id,
-    const std::string& saas,
+    const GrpcMetadata& metadata,
     AssetStreamRpcType rpc_type): observer_(observer),
                                   stor_(std::move(stor)) {
   ASSERT_EQ(0, lock_.init(true));
-  context_.AddMetadata("nsolid-agent-id", agent_id);
-  if (!saas.empty()) {
-    context_.AddMetadata("nsolid-saas-token", saas);
-  }
+  GrpcClient::AddMetadata(&context_, metadata);
 
   // Call the appropriate RPC method based on the rpc_type parameter
   if (rpc_type == EXPORT_CONTINUOUS_PROFILE) {

agents/grpc/src/asset_stream.h

@@ -2,6 +2,7 @@
 #define AGENTS_GRPC_SRC_ASSET_STREAM_H_
 
 #include "./proto/nsolid_service.grpc.pb.h"
+#include "grpc_client.h"
 #include "grpcpp/grpcpp.h"
 #include "nsolid/thread_safe.h"
 #include "../../src/profile_collector.h"
@@ -54,8 +55,7 @@ class AssetStream: public ::grpc::ClientWriteReactor<grpcagent::Asset> {
   explicit AssetStream(grpcagent::NSolidService::StubInterface* stub,
                        AssetStor&& stor,
                        std::weak_ptr<AssetStreamObserver> observer,
-                       const std::string& agent_id,
-                       const std::string& saas,
+                       const GrpcMetadata& metadata,
                        AssetStreamRpcType rpc_type = EXPORT_ASSET);
 
   ~AssetStream();

agents/grpc/src/command_stream.cc

@@ -12,14 +12,11 @@ namespace grpc {
 
 CommandStream::CommandStream(NSolidService::StubInterface* stub,
                              std::weak_ptr<CommandStreamObserver> observer,
-                             const std::string& agent_id,
-                             const std::string& saas): observer_(observer) {
+                             const GrpcMetadata& metadata):
+                                observer_(observer) {
   ASSERT_EQ(0, lock_.init(true));
   ASSERT_EQ(0, uv_cond_init(&on_done_cond_));
-  context_.AddMetadata("nsolid-agent-id", agent_id);
-  if (!saas.empty()) {
-    context_.AddMetadata("nsolid-saas-token", saas);
-  }
+  GrpcClient::AddMetadata(&context_, metadata);
   context_.set_wait_for_ready(true);
   stub->async()->Command(&context_, this);
   StartRead(&server_request_);

agents/grpc/src/command_stream.h

@@ -2,6 +2,7 @@
 #define AGENTS_GRPC_SRC_COMMAND_STREAM_H_
 
 #include "./proto/nsolid_service.grpc.pb.h"
+#include "grpc_client.h"
 #include "grpcpp/grpcpp.h"
 #include "nsolid/thread_safe.h"
 #include "nsuv-inl.h"
@@ -31,8 +32,7 @@ class CommandStream:
  public:
   explicit CommandStream(grpcagent::NSolidService::StubInterface* stub,
                          std::weak_ptr<CommandStreamObserver> observer,
-                         const std::string& agent_id,
-                         const std::string& saas);
+                         const GrpcMetadata& metadata);
 
   ~CommandStream();
 

agents/grpc/src/grpc_agent.cc

@@ -21,6 +21,7 @@
 #include "opentelemetry/exporters/otlp/otlp_grpc_metric_exporter.h"
 #include "opentelemetry/exporters/otlp/otlp_grpc_metric_exporter_factory.h"
 #include "opentelemetry/exporters/otlp/otlp_metric_utils.h"
+#include "../../src/node_version.h"
 
 using std::chrono::duration_cast;
 using std::chrono::nanoseconds;
@@ -122,6 +123,23 @@ ErrorType translate_error(int err) {
   }
 }
 
+GrpcMetadata GrpcAgent::BuildRpcMetadata(const std::string& agent_id,
+                                         const std::string& saas) {
+  GrpcMetadata metadata = {
+    {"nsolid-agent-id", agent_id},
+    {"nsolid-version", NSOLID_VERSION_STRING},
+  };
+  if (!saas.empty()) {
+    metadata.emplace("nsolid-saas", saas);
+  }
+
+  return metadata;
+}
+
+void GrpcAgent::RefreshRpcMetadata() {
+  rpc_metadata_ = BuildRpcMetadata(agent_id_, saas());
+}
+
 void PopulateCommon(grpcagent::CommonResponse* common,
                     const std::string& command,
                     const char* req_id) {
@@ -431,6 +449,7 @@ GrpcAgent::GrpcAgent(): hooks_init_(false),
                         proc_prev_stor_(),
                         config_(json::object()),
                         agent_id_(GetAgentId()),
+                        rpc_metadata_(BuildRpcMetadata(agent_id_, "")),
                         auth_retries_(0),
                         unauthorized_(false),
                         assets_enabled_(true),
@@ -501,8 +520,7 @@ void GrpcAgent::reset_command_stream() {
   Debug("Resetting command stream\n");
   command_stream_ = std::make_unique<CommandStream>(nsolid_service_stub_.get(),
                                                     weak_from_this(),
-                                                    agent_id_,
-                                                    saas());
+                                                    rpc_metadata());
 }
 
 void GrpcAgent::set_asset_cb(SharedEnvInst envinst,
@@ -1018,21 +1036,24 @@ int GrpcAgent::config(const json& config) {
   json old_config = config_;
   config_ = config;
   json diff = json::diff(old_config, config_);
+  const bool saas_changed = utils::find_any_fields_in_diff(diff, { "/saas" });
+  const bool grpc_changed = utils::find_any_fields_in_diff(diff, { "/grpc" });
   DebugJSON("Old Config: \n%s\n", old_config);
   DebugJSON("NewConfig: \n%s\n", config_);
   DebugJSON("Diff: \n%s\n", diff);
 
-  if (utils::find_any_fields_in_diff(diff, { "/saas" })) {
+  if (saas_changed) {
     auto it = config_.find("saas");
     saas_.reset();
     if (it != config_.end()) {
       parse_saas_token(*it);
     }
+    RefreshRpcMetadata();
   }
 
-  if (utils::find_any_fields_in_diff(diff, { "/grpc" })) {
+  if (grpc_changed || saas_changed) {
     auto it = config_.find("grpc");
-    if (it != config_.end()) {
+    if (it != config_.end() || saas_) {
       // Setup the client/s
 
       bool insecure = false;
@@ -1053,8 +1074,7 @@ int GrpcAgent::config(const json& config) {
       OtlpGrpcClientOptions opts;
       opts.compression = "gzip";
       opts.endpoint = endpoint;
-      opts.metadata = {{"nsolid-agent-id", agent_id_},
-                       {"nsolid-saas", saas()}};
+      opts.metadata = rpc_metadata();
       opts.timeout = DEFAULT_GRPC_TIMEOUT;
       // Make sure the client is initialized. We set it to the same
       // default value as ax_concurrent_requests as the exporters.
@@ -1086,8 +1106,7 @@ int GrpcAgent::config(const json& config) {
         OtlpGrpcExporterOptions options;
         options.compression = "gzip";
         options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
+        options.metadata = rpc_metadata();
         options.timeout = DEFAULT_GRPC_TIMEOUT;
         if (!insecure) {
           options.credentials = opts.credentials;
@@ -1099,8 +1118,7 @@ int GrpcAgent::config(const json& config) {
         OtlpGrpcMetricExporterOptions options;
         options.compression = "gzip";
         options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
+        options.metadata = rpc_metadata();
         options.timeout = DEFAULT_GRPC_TIMEOUT;
         if (!insecure) {
           options.credentials = opts.credentials;
@@ -1113,8 +1131,7 @@ int GrpcAgent::config(const json& config) {
         OtlpGrpcLogRecordExporterOptions options;
         options.compression = "gzip";
         options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
+        options.metadata = rpc_metadata();
         options.timeout = DEFAULT_GRPC_TIMEOUT;
         if (!insecure) {
           options.credentials = opts.credentials;
@@ -1534,8 +1551,7 @@ void GrpcAgent::got_continuous_profile(
     AssetStream* stream = new AssetStream(nsolid_service_stub_.get(),
                                           AssetStor{stor.type, thread_id},
                                           weak_from_this(),
-                                          agent_id_,
-                                          saas(),
+                                          rpc_metadata(),
                                           EXPORT_CONTINUOUS_PROFILE);
     it = cont_profile_stor_map_.emplace(thread_id, ProfileStor{
       utils::generate_unique_id(),
@@ -1794,7 +1810,7 @@ void GrpcAgent::send_blocked_loop_event(BlockedLoopStor&& stor) {
     Arena::Create<grpcagent::BlockedLoopEvent>(arena.get());
   PopulateBlockedLoopEvent(event, stor);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::BlockedLoopEvent>(
     nsolid_service_stub_.get(),
@@ -1831,7 +1847,7 @@ void GrpcAgent::send_exit() {
     exit_body->set_profile(cpu_profile_state.last_main_profile);
   }
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
   uv_cond_t cond;
   uv_mutex_t lock;
   bool signaled = false;
@@ -1881,7 +1897,7 @@ void GrpcAgent::send_info_event(const char* req_id) {
     PopulateInfoEvent(info_event, info, req_id);
   }
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::InfoEvent>(
     nsolid_service_stub_.get(),
@@ -1909,7 +1925,7 @@ void GrpcAgent::send_metrics_event(const char* req_id) {
                        thr_metrics_cache_,
                        req_id);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::MetricsEvent>(
     nsolid_service_stub_.get(),
@@ -1932,7 +1948,7 @@ void GrpcAgent::send_packages_event(const char* req_id) {
   auto packages_event = Arena::Create<grpcagent::PackagesEvent>(arena.get());
   PopulatePackagesEvent(packages_event, req_id);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::PackagesEvent>(
     nsolid_service_stub_.get(),
@@ -1956,7 +1972,7 @@ void GrpcAgent::send_reconfigure_event(const char* req_id) {
     Arena::Create<grpcagent::ReconfigureEvent>(arena.get());
   PopulateReconfigureEvent(reconfigure_event, req_id);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::ReconfigureEvent>(
     nsolid_service_stub_.get(),
@@ -2007,7 +2023,7 @@ void GrpcAgent::send_source_code_event(const grpcagent::CommandRequest& req) {
     }
   }
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::SourceCodeEvent>(
     nsolid_service_stub_.get(),
@@ -2030,7 +2046,7 @@ void GrpcAgent::send_startup_times_event(const char* req_id) {
   auto st_event = Arena::Create<grpcagent::StartupTimesEvent>(arena.get());
   PopulateStartupTimesEvent(st_event, req_id);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::StartupTimesEvent>(
     nsolid_service_stub_.get(),
@@ -2055,7 +2071,7 @@ void GrpcAgent::send_unblocked_loop_event(BlockedLoopStor&& stor) {
     Arena::Create<grpcagent::UnblockedLoopEvent>(arena.get());
   PopulateUnblockedLoopEvent(event, stor);
 
-  auto context = GrpcClient::MakeClientContext(agent_id_, saas());
+  auto context = GrpcClient::MakeClientContext(rpc_metadata());
 
   GrpcClient::DelegateAsyncExport<grpcagent::UnblockedLoopEvent>(
     nsolid_service_stub_.get(),
@@ -2170,8 +2186,7 @@ ErrorType GrpcAgent::do_start_prof_end(ErrorType err,
   auto stream = new AssetStream(nsolid_service_stub_.get(),
                                 AssetStor{type, thread_id},
                                 weak_from_this(),
-                                agent_id_,
-                                saas(),
+                                rpc_metadata(),
                                 EXPORT_ASSET);
   if (err != ErrorType::ESuccess) {
     send_asset_error(type, req_id, std::move(opts), stream, err);

agents/grpc/src/grpc_agent.h

@@ -110,6 +110,8 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
     return saas_ ? saas_->token : empty;
   }
 
+  const GrpcMetadata& rpc_metadata() const { return rpc_metadata_; }
+
  private:
   struct CommandRequestStor {
     grpcagent::CommandRequest request;
@@ -256,6 +258,11 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
 
   void parse_saas_token(const std::string& token);
 
+  static GrpcMetadata BuildRpcMetadata(const std::string& agent_id,
+                                       const std::string& saas);
+
+  void RefreshRpcMetadata();
+
   bool pending_profiles() const;
 
   void reconfigure(const grpcagent::CommandRequest& config);
@@ -335,6 +342,7 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
   TSQueue<nlohmann::json> config_msg_q_;
   nlohmann::json config_;
   std::string agent_id_;
+  GrpcMetadata rpc_metadata_;
   std::unique_ptr<SaaSInfo> saas_;
 
   nsuv::ns_timer auth_timer_;

agents/grpc/src/grpc_client.cc

@@ -82,17 +82,20 @@ std::shared_ptr<Channel>
   * Create gRPC client context to call RPC.
   */
 std::unique_ptr<ClientContext>
-GrpcClient::MakeClientContext(const std::string& agent_id,
-                              const std::string& saas) {
+GrpcClient::MakeClientContext(const GrpcMetadata& metadata) {
   std::unique_ptr<ClientContext> context = std::make_unique<ClientContext>();
-  context->AddMetadata("nsolid-agent-id", agent_id);
-  if (!saas.empty()) {
-    context->AddMetadata("nsolid-saas-token", saas);
-  }
+  AddMetadata(context.get(), metadata);
 
   return context;
 }
 
+void GrpcClient::AddMetadata(ClientContext* context,
+                             const GrpcMetadata& metadata) {
+  for (const auto& [key, value] : metadata) {
+    context->AddMetadata(key, value);
+  }
+}
+
 /**
   * Create N|Solid service stub to communicate with the N|Solid Console.
   */

agents/grpc/src/grpc_client.h

@@ -9,6 +9,7 @@
 #include "google/protobuf/util/json_util.h"
 #include "grpc_utils.h"
 #include "grpcpp/grpcpp.h"
+#include "opentelemetry/exporters/otlp/otlp_environment.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -21,6 +22,7 @@ OPENTELEMETRY_END_NAMESPACE
 
 using google::protobuf::Arena;
 using opentelemetry::v1::exporter::otlp::OtlpGrpcClientOptions;
+using GrpcMetadata = opentelemetry::v1::exporter::otlp::OtlpHeaders;
 
 namespace node {
 namespace nsolid {
@@ -74,7 +76,10 @@ class GrpcClient {
    * Create gRPC client context to call RPC.
    */
   static std::unique_ptr<::grpc::ClientContext>
-    MakeClientContext(const std::string& agent_id, const std::string& saas);
+    MakeClientContext(const GrpcMetadata& metadata);
+
+  static void AddMetadata(::grpc::ClientContext* context,
+                          const GrpcMetadata& metadata);
 
   /**
    * Create N|Solid service stub to communicate with the N|Solid Console.

test/agents/test-grpc-basic.mjs

@@ -4,6 +4,7 @@ import fixtures from '../common/fixtures.js';
 import assert from 'node:assert';
 import {
   checkExitData,
+  checkRpcMetadata,
   GRPCServer,
   TestClient,
 } from '../common/nsolid-grpc-agent/index.js';
@@ -165,8 +166,12 @@ tests.push({
           config.saas = correctEnv.NSOLID_SAAS;
         }
 
-        grpcServer.on('command', mustCall(async ({ agentId }) => {
+        grpcServer.on('command', mustCall(async ({ agentId, metadata }) => {
           // Verify the CommandStream is working by sending a command from server to client
+          checkRpcMetadata(metadata, agentId);
+          if (correctEnv.NSOLID_SAAS) {
+            assert.strictEqual(metadata['nsolid-saas'][0], correctEnv.NSOLID_SAAS);
+          }
           const infoResult = await grpcServer.info(agentId);
           assert.ok(infoResult);
           const exit = await child.shutdown(0);