agents: implement gRPC retry policies

Enable them in opentelemetry-cpp by defining
`ENABLE_OTLP_RETRY_PREVIEW` and also them to our GrpcClient.
The values we're using with a default deadline of 10 seconds are:
  - max_attemps: 5
  - initial_backoff: 500ms
  - max_backoff: 5s
  - backoff_multiplier: 2

Add NSOLID_GRPC_DEADLINE env var for configurable gRPC timeouts.
Specially useful for testing.

Refactor and dry code when creating grpc exporters inside the new
GrpcAgent::configure_grpc_exporters() helper function.

Create parse_env_var_int() utility in nsolid_util.h for safe env
parsing.

Author: santigimeno

agents/grpc/src/asset_stream.cc

@@ -1,4 +1,5 @@
 #include "asset_stream.h"
+#include "grpc_client.h"
 
 #include <cinttypes>
 
@@ -8,6 +9,8 @@
 
 using grpc::Status;
 using grpcagent::NSolidService;
+using std::chrono::system_clock;
+using std::chrono::seconds;
 
 namespace node {
 namespace nsolid {
@@ -27,6 +30,10 @@ AssetStream::AssetStream(
     context_.AddMetadata("nsolid-saas-token", saas);
   }
 
+  // Use configurable deadline for asset streams
+  context_.set_deadline(system_clock::now() +
+                        seconds(get_grpc_deadline_seconds()));
+
   // Call the appropriate RPC method based on the rpc_type parameter
   if (rpc_type == EXPORT_CONTINUOUS_PROFILE) {
     stub->async()->ExportContinuousProfile(&context_, &event_response_, this);

agents/grpc/src/grpc_agent.cc

@@ -1,4 +1,5 @@
 #include "grpc_agent.h"
+#include "../../src/nsolid/nsolid_util.h"
 
 #include "asserts-cpp/asserts.h"
 #include "nsolid/nsolid_api.h"
@@ -992,7 +993,81 @@ void GrpcAgent::check_exit_on_profile() {
   }
 }
 
-int GrpcAgent::config(const json& config) {
+void GrpcAgent::configure_grpc_exporters(const std::string& endpoint,
+                                         bool insecure) {
+  Debug("GrpcAgent configured. Endpoint: %s. Insecure: %d\n",
+        endpoint.c_str(), static_cast<unsigned>(insecure));
+
+  OtlpGrpcClientOptions opts;
+  opts.compression = "gzip";
+  opts.endpoint = endpoint;
+  opts.metadata = {{"nsolid-agent-id", agent_id_},
+                   {"nsolid-saas", saas()}};
+  // Make sure the client is initialized. We set it to the same
+  // default value as max_concurrent_requests as the exporters.
+  opts.max_concurrent_requests = 64;
+  opts.use_ssl_credentials = !insecure;
+  if (!insecure) {
+    if (!custom_certs_.empty()) {
+      opts.ssl_credentials_cacert_as_string = custom_certs_;
+    } else {
+      opts.ssl_credentials_cacert_as_string = cacert_;
+    }
+  }
+
+  opts.retry_policy_max_attempts = 5;
+  opts.retry_policy_initial_backoff = std::chrono::milliseconds(500);
+  opts.retry_policy_max_backoff = std::chrono::seconds(5);
+  opts.retry_policy_backoff_multiplier = 2.0f;
+
+  nsolid_service_stub_ = GrpcClient::MakeNSolidServiceStub(opts);
+  // CommandStream needs to be created before the OTLP client to avoid
+  // a race condition with abseil mutexes.
+  reset_command_stream();
+
+  std::shared_ptr<OtlpGrpcClient> client =
+      OtlpGrpcClientFactory::Create(opts);
+
+  auto set_common_options = [&](auto& options) {
+    options.compression = "gzip";
+    options.endpoint = endpoint;
+    options.metadata = {{"nsolid-agent-id", agent_id_},
+                        {"nsolid-saas", saas()}};
+    if (!insecure) {
+      options.use_ssl_credentials = true;
+      if (!custom_certs_.empty()) {
+        options.ssl_credentials_cacert_as_string = custom_certs_;
+      } else {
+        options.ssl_credentials_cacert_as_string = cacert_;
+      }
+    }
+
+    options.retry_policy_max_attempts = 5;
+    options.retry_policy_initial_backoff = std::chrono::milliseconds(500);
+    options.retry_policy_max_backoff = std::chrono::seconds(5);
+    options.retry_policy_backoff_multiplier = 2.0f;
+  };
+
+  {
+    OtlpGrpcExporterOptions options;
+    set_common_options(options);
+    trace_exporter_ = std::make_unique<OtlpGrpcExporter>(options, client);
+  }
+  {
+    OtlpGrpcMetricExporterOptions options;
+    set_common_options(options);
+    metrics_exporter_ =
+      std::make_unique<OtlpGrpcMetricExporter>(options, client);
+  }
+  {
+    OtlpGrpcLogRecordExporterOptions options;
+    set_common_options(options);
+    log_exporter_ =
+      std::make_unique<OtlpGrpcLogRecordExporter>(options, client);
+  }
+}
+
+int GrpcAgent::config(const nlohmann::json& config) {
   int ret = 0;
   json old_config = config_;
   config_ = config;
@@ -1020,93 +1095,13 @@ int GrpcAgent::config(const json& config) {
       // Only parse the insecure flag in non SaaS mode.
       if (insecure_str.has_value() && (!saas_ || saas_->testing)) {
         // insecure = std::stoull(insecure_str.value());
-        insecure = std::stoi(insecure_str.value());
+        insecure = utils::parse_env_var_int(kNSOLID_GRPC_INSECURE, 0, 0, 1);
       }
 
       const std::string& endpoint = !saas_ ?
                                     it->get<std::string>() :
                                     saas_->endpoint;
-      Debug("GrpcAgent configured. Endpoint: %s. Insecure: %d\n",
-            endpoint.c_str(), static_cast<unsigned>(insecure));
-
-      OtlpGrpcClientOptions opts;
-      opts.compression = "gzip";
-      opts.endpoint = endpoint;
-      opts.metadata = {{"nsolid-agent-id", agent_id_},
-                       {"nsolid-saas", saas()}};
-      // Make sure the client is initialized. We set it to the same
-      // default value as ax_concurrent_requests as the exporters.
-      opts.max_concurrent_requests = 64;
-      opts.use_ssl_credentials = !insecure;
-      if (!insecure) {
-        if (!custom_certs_.empty()) {
-          opts.ssl_credentials_cacert_as_string = custom_certs_;
-        } else {
-          opts.ssl_credentials_cacert_as_string = cacert_;
-        }
-      }
-
-      nsolid_service_stub_ = GrpcClient::MakeNSolidServiceStub(opts);
-      // CommandStream needs to be created before the OTLP client to avoid
-      // a race condition with abseil mutexes.
-      reset_command_stream();
-
-      std::shared_ptr<OtlpGrpcClient> client =
-          OtlpGrpcClientFactory::Create(opts);
-
-      {
-        OtlpGrpcExporterOptions options;
-        options.compression = "gzip";
-        options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
-        if (!insecure) {
-          options.use_ssl_credentials = true;
-          if (!custom_certs_.empty()) {
-            options.ssl_credentials_cacert_as_string = custom_certs_;
-          } else {
-            options.ssl_credentials_cacert_as_string = cacert_;
-          }
-        }
-
-        trace_exporter_ = std::make_unique<OtlpGrpcExporter>(options, client);
-      }
-      {
-        OtlpGrpcMetricExporterOptions options;
-        options.compression = "gzip";
-        options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
-        if (!insecure) {
-          options.use_ssl_credentials = true;
-          if (!custom_certs_.empty()) {
-            options.ssl_credentials_cacert_as_string = custom_certs_;
-          } else {
-            options.ssl_credentials_cacert_as_string = cacert_;
-          }
-        }
-
-        metrics_exporter_ =
-          std::make_unique<OtlpGrpcMetricExporter>(options, client);
-      }
-      {
-        OtlpGrpcLogRecordExporterOptions options;
-        options.compression = "gzip";
-        options.endpoint = endpoint;
-        options.metadata = {{"nsolid-agent-id", agent_id_},
-                            {"nsolid-saas", saas()}};
-        if (!insecure) {
-          options.use_ssl_credentials = true;
-          if (!custom_certs_.empty()) {
-            options.ssl_credentials_cacert_as_string = custom_certs_;
-          } else {
-            options.ssl_credentials_cacert_as_string = cacert_;
-          }
-        }
-
-        log_exporter_ =
-          std::make_unique<OtlpGrpcLogRecordExporter>(options, client);
-      }
+      configure_grpc_exporters(endpoint, insecure);
     }
   }
 
@@ -1804,7 +1799,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(agent_id_, saas(), 1); // 1 second for fast exit
   uv_cond_t cond;
   uv_mutex_t lock;
   bool signaled = false;

agents/grpc/src/grpc_agent.h

@@ -202,6 +202,8 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
 
   void command_stream_closed(const ::grpc::Status& s);
 
+  void configure_grpc_exporters(const std::string& endpoint, bool insecure);
+
   int config(const nlohmann::json& config);
 
   void do_start();

agents/grpc/src/grpc_client.cc

@@ -2,6 +2,10 @@
 #include "debug_utils-inl.h"
 #include "opentelemetry/exporters/otlp/otlp_grpc_client_options.h"
 
+#include <cerrno>
+#include <cstdlib>
+#include <limits>
+
 using grpc::Channel;
 using grpc::ChannelArguments;
 using grpc::ClientContext;
@@ -11,11 +15,23 @@ using grpc::SslCredentials;
 using grpc::SslCredentialsOptions;
 using grpcagent::NSolidService;
 using opentelemetry::v1::exporter::otlp::OtlpGrpcClientOptions;
+using std::chrono::system_clock;
+using std::chrono::seconds;
 
 namespace node {
 namespace nsolid {
 namespace grpc {
 
+const char* const kNSOLID_GRPC_DEADLINE = "NSOLID_GRPC_DEADLINE";
+
+int get_grpc_deadline_seconds() {
+  static int cached_deadline = -1;
+  if (cached_deadline == -1) {
+    cached_deadline = parse_env_var_int(kNSOLID_GRPC_DEADLINE, 10, 1, INT_MAX);
+  }
+  return cached_deadline;
+}
+
 /**
   * Create gRPC channel.
   */
@@ -32,6 +48,48 @@ std::shared_ptr<Channel>
   grpc_arguments.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 15 * 1000 /* 15 sec*/);
   grpc_arguments.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
   grpc_arguments.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
+
+
+  static const auto kServiceConfigJson = std::string_view{R"(
+  {
+    "methodConfig": [
+      {
+        "name": [{}],
+        "retryPolicy": {
+          "maxAttempts": %0000000000u,
+          "initialBackoff": "%0000000000.1fs",
+          "maxBackoff": "%0000000000.1fs",
+          "backoffMultiplier": %0000000000.1f,
+          "retryableStatusCodes": [
+            "CANCELLED",
+            "DEADLINE_EXCEEDED",
+            "ABORTED",
+            "OUT_OF_RANGE",
+            "DATA_LOSS",
+            "UNAVAILABLE"
+          ]
+        }
+      }
+    ]
+  })"};
+
+  // Allocate string with buffer large enough to hold the formatted json config
+  auto service_config = std::string(kServiceConfigJson.size(), '\0');
+  // Prior to C++17, need to explicitly cast away constness from `data()` buffer
+  float initial_backoff = options.retry_policy_initial_backoff.count();
+  float max_backoff = options.retry_policy_max_backoff.count();
+  float backoff_multiplier = options.retry_policy_backoff_multiplier;
+  std::snprintf(
+    const_cast<decltype(service_config)::value_type *>(service_config.data()),
+    service_config.size(),
+    kServiceConfigJson.data(),
+    options.retry_policy_max_attempts,
+    std::min(std::max(initial_backoff, 0.f), 999999999.f),
+    std::min(std::max(max_backoff, 0.f), 999999999.f),
+    std::min(std::max(backoff_multiplier, 0.f), 999999999.f));
+
+  grpc_arguments.SetServiceConfigJSON(service_config);
+
   if (!options.use_ssl_credentials) {
     channel = CreateCustomChannel(options.endpoint,
                                   InsecureChannelCredentials(),
@@ -55,7 +113,19 @@ std::shared_ptr<Channel>
 std::unique_ptr<ClientContext>
 GrpcClient::MakeClientContext(const std::string& agent_id,
                               const std::string& saas) {
+  // Forward to the version with custom deadline using environment variable
+  return MakeClientContext(agent_id, saas, get_grpc_deadline_seconds());
+}
+
+/**
+  * Create gRPC client context to call RPC with custom deadline.
+  */
+std::unique_ptr<ClientContext>
+GrpcClient::MakeClientContext(const std::string& agent_id,
+                              const std::string& saas,
+                              int deadline_seconds) {
   std::unique_ptr<ClientContext> context = std::make_unique<ClientContext>();
+  context->set_deadline(system_clock::now() + seconds(deadline_seconds));
   context->AddMetadata("nsolid-agent-id", agent_id);
   if (!saas.empty()) {
     context->AddMetadata("nsolid-saas-token", saas);

agents/grpc/src/grpc_client.h

@@ -68,6 +68,14 @@ class GrpcClient {
   static std::unique_ptr<::grpc::ClientContext>
     MakeClientContext(const std::string& agent_id, const std::string& saas);
 
+  /**
+   * Create gRPC client context to call RPC with custom deadline.
+   */
+  static std::unique_ptr<::grpc::ClientContext>
+    MakeClientContext(const std::string& agent_id,
+                      const std::string& saas, 
+                      int deadline_seconds);
+
   /**
    * Create N|Solid service stub to communicate with the N|Solid Console.
    */
@@ -148,6 +156,9 @@ class GrpcClient {
   }
 };
 
+// Helper function to get configurable gRPC deadline
+int get_grpc_deadline_seconds();
+
 }  // namespace grpc
 }  // namespace nsolid
 }  // namespace node

deps/opentelemetry-cpp/otlp-http-exporter.gyp

@@ -63,6 +63,7 @@
         'BUILDING_LIBCURL',
         'ENABLE_ASYNC_EXPORT',
         'OPENTELEMETRY_STL_VERSION=2020',
+        'ENABLE_OTLP_RETRY_PREVIEW',
       ],
       'dependencies': [
         '../protobuf/protobuf.gyp:protobuf',
@@ -75,6 +76,7 @@
         'defines': [
           'ENABLE_ASYNC_EXPORT',
           'OPENTELEMETRY_STL_VERSION=2020',
+          'ENABLE_OTLP_RETRY_PREVIEW',
         ],
         'include_dirs': [
           'api/include',