Skip to content

Commit 2e4a6ef

Browse files
santigimenosantigimeno
committed
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.
1 parent fcab189 commit 2e4a6ef

10 files changed

Lines changed: 237 additions & 104 deletions

File tree

‎agents/grpc/src/asset_stream.cc‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "asset_stream.h"
2+
#include "grpc_client.h"
23

34
#include <cinttypes>
45

@@ -8,6 +9,8 @@
89

910
using grpc::Status;
1011
using grpcagent::NSolidService;
12+
using std::chrono::system_clock;
13+
using std::chrono::seconds;
1114

1215
namespace node {
1316
namespace nsolid {
@@ -27,6 +30,10 @@ AssetStream::AssetStream(
2730
context_.AddMetadata("nsolid-saas-token", saas);
2831
}
2932

33+
// Use configurable deadline for asset streams
34+
context_.set_deadline(system_clock::now() +
35+
seconds(get_grpc_deadline_seconds()));
36+
3037
// Call the appropriate RPC method based on the rpc_type parameter
3138
if (rpc_type == EXPORT_CONTINUOUS_PROFILE) {
3239
stub->async()->ExportContinuousProfile(&context_, &event_response_, this);

‎agents/grpc/src/grpc_agent.cc‎

Lines changed: 85 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "grpc_agent.h"
2+
#include "../../src/nsolid/nsolid_util.h"
23

34
#include "asserts-cpp/asserts.h"
45
#include "nsolid/nsolid_api.h"
@@ -62,6 +63,11 @@ using ThreadMetricsMap = std::map<uint64_t, ThreadMetrics::MetricsStor>;
6263

6364
constexpr uint64_t span_timer_interval = 1000;
6465
constexpr size_t span_msg_q_min_size = 1000;
66+
// At class or file scope
67+
constexpr size_t retry_max_attempts = 5;
68+
constexpr auto retry_initial_backoff = std::chrono::milliseconds(500);
69+
constexpr auto retry_max_backoff = std::chrono::seconds(5);
70+
constexpr float retry_backoff_multiplier = 2.0f;
6571

6672
const char* const kNSOLID_GRPC_INSECURE = "NSOLID_GRPC_INSECURE";
6773
const char* const kNSOLID_GRPC_CERTS = "NSOLID_GRPC_CERTS";
@@ -998,7 +1004,81 @@ void GrpcAgent::check_exit_on_profile() {
9981004
}
9991005
}
10001006

1001-
int GrpcAgent::config(const json& config) {
1007+
void GrpcAgent::configure_grpc_exporters(const std::string& endpoint,
1008+
bool insecure) {
1009+
Debug("GrpcAgent configured. Endpoint: %s. Insecure: %d\n",
1010+
endpoint.c_str(), static_cast<unsigned>(insecure));
1011+
1012+
OtlpGrpcClientOptions opts;
1013+
opts.compression = "gzip";
1014+
opts.endpoint = endpoint;
1015+
opts.metadata = {{"nsolid-agent-id", agent_id_},
1016+
{"nsolid-saas", saas()}};
1017+
// Make sure the client is initialized. We set it to the same
1018+
// default value as max_concurrent_requests as the exporters.
1019+
opts.max_concurrent_requests = 64;
1020+
opts.use_ssl_credentials = !insecure;
1021+
if (!insecure) {
1022+
if (!custom_certs_.empty()) {
1023+
opts.ssl_credentials_cacert_as_string = custom_certs_;
1024+
} else {
1025+
opts.ssl_credentials_cacert_as_string = cacert_;
1026+
}
1027+
}
1028+
1029+
opts.retry_policy_max_attempts = retry_max_attempts;
1030+
opts.retry_policy_initial_backoff = retry_initial_backoff;
1031+
opts.retry_policy_max_backoff = retry_max_backoff;
1032+
opts.retry_policy_backoff_multiplier = retry_backoff_multiplier;
1033+
1034+
nsolid_service_stub_ = GrpcClient::MakeNSolidServiceStub(opts);
1035+
// CommandStream needs to be created before the OTLP client to avoid
1036+
// a race condition with abseil mutexes.
1037+
reset_command_stream();
1038+
1039+
std::shared_ptr<OtlpGrpcClient> client =
1040+
OtlpGrpcClientFactory::Create(opts);
1041+
1042+
auto set_common_options = [&](auto& options) {
1043+
options.compression = "gzip";
1044+
options.endpoint = endpoint;
1045+
options.metadata = {{"nsolid-agent-id", agent_id_},
1046+
{"nsolid-saas", saas()}};
1047+
if (!insecure) {
1048+
options.use_ssl_credentials = true;
1049+
if (!custom_certs_.empty()) {
1050+
options.ssl_credentials_cacert_as_string = custom_certs_;
1051+
} else {
1052+
options.ssl_credentials_cacert_as_string = cacert_;
1053+
}
1054+
}
1055+
1056+
options.retry_policy_max_attempts = retry_max_attempts;
1057+
options.retry_policy_initial_backoff = retry_initial_backoff;
1058+
options.retry_policy_max_backoff = retry_max_backoff;
1059+
options.retry_policy_backoff_multiplier = retry_backoff_multiplier;
1060+
};
1061+
1062+
{
1063+
OtlpGrpcExporterOptions options;
1064+
set_common_options(options);
1065+
trace_exporter_ = std::make_unique<OtlpGrpcExporter>(options, client);
1066+
}
1067+
{
1068+
OtlpGrpcMetricExporterOptions options;
1069+
set_common_options(options);
1070+
metrics_exporter_ =
1071+
std::make_unique<OtlpGrpcMetricExporter>(options, client);
1072+
}
1073+
{
1074+
OtlpGrpcLogRecordExporterOptions options;
1075+
set_common_options(options);
1076+
log_exporter_ =
1077+
std::make_unique<OtlpGrpcLogRecordExporter>(options, client);
1078+
}
1079+
}
1080+
1081+
int GrpcAgent::config(const nlohmann::json& config) {
10021082
int ret = 0;
10031083
json old_config = config_;
10041084
config_ = config;
@@ -1026,93 +1106,13 @@ int GrpcAgent::config(const json& config) {
10261106
// Only parse the insecure flag in non SaaS mode.
10271107
if (insecure_str.has_value() && (!saas_ || saas_->testing)) {
10281108
// insecure = std::stoull(insecure_str.value());
1029-
insecure = std::stoi(insecure_str.value());
1109+
insecure = utils::parse_env_var_int(kNSOLID_GRPC_INSECURE, 0, 0, 1);
10301110
}
10311111

10321112
const std::string& endpoint = !saas_ ?
10331113
it->get<std::string>() :
10341114
saas_->endpoint;
1035-
Debug("GrpcAgent configured. Endpoint: %s. Insecure: %d\n",
1036-
endpoint.c_str(), static_cast<unsigned>(insecure));
1037-
1038-
OtlpGrpcClientOptions opts;
1039-
opts.compression = "gzip";
1040-
opts.endpoint = endpoint;
1041-
opts.metadata = {{"nsolid-agent-id", agent_id_},
1042-
{"nsolid-saas", saas()}};
1043-
// Make sure the client is initialized. We set it to the same
1044-
// default value as ax_concurrent_requests as the exporters.
1045-
opts.max_concurrent_requests = 64;
1046-
opts.use_ssl_credentials = !insecure;
1047-
if (!insecure) {
1048-
if (!custom_certs_.empty()) {
1049-
opts.ssl_credentials_cacert_as_string = custom_certs_;
1050-
} else {
1051-
opts.ssl_credentials_cacert_as_string = cacert_;
1052-
}
1053-
}
1054-
1055-
nsolid_service_stub_ = GrpcClient::MakeNSolidServiceStub(opts);
1056-
// CommandStream needs to be created before the OTLP client to avoid
1057-
// a race condition with abseil mutexes.
1058-
reset_command_stream();
1059-
1060-
std::shared_ptr<OtlpGrpcClient> client =
1061-
OtlpGrpcClientFactory::Create(opts);
1062-
1063-
{
1064-
OtlpGrpcExporterOptions options;
1065-
options.compression = "gzip";
1066-
options.endpoint = endpoint;
1067-
options.metadata = {{"nsolid-agent-id", agent_id_},
1068-
{"nsolid-saas", saas()}};
1069-
if (!insecure) {
1070-
options.use_ssl_credentials = true;
1071-
if (!custom_certs_.empty()) {
1072-
options.ssl_credentials_cacert_as_string = custom_certs_;
1073-
} else {
1074-
options.ssl_credentials_cacert_as_string = cacert_;
1075-
}
1076-
}
1077-
1078-
trace_exporter_ = std::make_unique<OtlpGrpcExporter>(options, client);
1079-
}
1080-
{
1081-
OtlpGrpcMetricExporterOptions options;
1082-
options.compression = "gzip";
1083-
options.endpoint = endpoint;
1084-
options.metadata = {{"nsolid-agent-id", agent_id_},
1085-
{"nsolid-saas", saas()}};
1086-
if (!insecure) {
1087-
options.use_ssl_credentials = true;
1088-
if (!custom_certs_.empty()) {
1089-
options.ssl_credentials_cacert_as_string = custom_certs_;
1090-
} else {
1091-
options.ssl_credentials_cacert_as_string = cacert_;
1092-
}
1093-
}
1094-
1095-
metrics_exporter_ =
1096-
std::make_unique<OtlpGrpcMetricExporter>(options, client);
1097-
}
1098-
{
1099-
OtlpGrpcLogRecordExporterOptions options;
1100-
options.compression = "gzip";
1101-
options.endpoint = endpoint;
1102-
options.metadata = {{"nsolid-agent-id", agent_id_},
1103-
{"nsolid-saas", saas()}};
1104-
if (!insecure) {
1105-
options.use_ssl_credentials = true;
1106-
if (!custom_certs_.empty()) {
1107-
options.ssl_credentials_cacert_as_string = custom_certs_;
1108-
} else {
1109-
options.ssl_credentials_cacert_as_string = cacert_;
1110-
}
1111-
}
1112-
1113-
log_exporter_ =
1114-
std::make_unique<OtlpGrpcLogRecordExporter>(options, client);
1115-
}
1115+
configure_grpc_exporters(endpoint, insecure);
11161116
}
11171117
}
11181118

@@ -1821,7 +1821,8 @@ void GrpcAgent::send_exit() {
18211821
exit_body->set_profile(cpu_profile_state.last_main_profile);
18221822
}
18231823

1824-
auto context = GrpcClient::MakeClientContext(agent_id_, saas());
1824+
// 1 second for fast exit
1825+
auto context = GrpcClient::MakeClientContext(agent_id_, saas(), 1);
18251826
uv_cond_t cond;
18261827
uv_mutex_t lock;
18271828
bool signaled = false;

‎agents/grpc/src/grpc_agent.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,8 @@ class GrpcAgent: public std::enable_shared_from_this<GrpcAgent>,
202202

203203
void command_stream_closed(const ::grpc::Status& s);
204204

205+
void configure_grpc_exporters(const std::string& endpoint, bool insecure);
206+
205207
int config(const nlohmann::json& config);
206208

207209
void do_start();

‎agents/grpc/src/grpc_client.cc‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#include "grpc_client.h"
22
#include "debug_utils-inl.h"
3+
#include "nsolid/nsolid_util.h"
34
#include "opentelemetry/exporters/otlp/otlp_grpc_client_options.h"
45

56
using grpc::Channel;
@@ -11,11 +12,24 @@ using grpc::SslCredentials;
1112
using grpc::SslCredentialsOptions;
1213
using grpcagent::NSolidService;
1314
using opentelemetry::v1::exporter::otlp::OtlpGrpcClientOptions;
15+
using std::chrono::system_clock;
16+
using std::chrono::seconds;
1417

1518
namespace node {
1619
namespace nsolid {
20+
21+
using utils::parse_env_var_int;
22+
1723
namespace grpc {
1824

25+
const char* const kNSOLID_GRPC_DEADLINE = "NSOLID_GRPC_DEADLINE";
26+
27+
int get_grpc_deadline_seconds() {
28+
static const int cached_deadline =
29+
parse_env_var_int(kNSOLID_GRPC_DEADLINE, 10, 1, INT_MAX);
30+
return cached_deadline;
31+
}
32+
1933
/**
2034
* Create gRPC channel.
2135
*/
@@ -32,6 +46,48 @@ std::shared_ptr<Channel>
3246
grpc_arguments.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 15 * 1000 /* 15 sec*/);
3347
grpc_arguments.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
3448
grpc_arguments.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
49+
50+
51+
static const auto kServiceConfigJson = std::string_view{R"(
52+
{
53+
"methodConfig": [
54+
{
55+
"name": [{}],
56+
"retryPolicy": {
57+
"maxAttempts": %0000000000u,
58+
"initialBackoff": "%0000000000.1fs",
59+
"maxBackoff": "%0000000000.1fs",
60+
"backoffMultiplier": %0000000000.1f,
61+
"retryableStatusCodes": [
62+
"CANCELLED",
63+
"DEADLINE_EXCEEDED",
64+
"ABORTED",
65+
"OUT_OF_RANGE",
66+
"DATA_LOSS",
67+
"UNAVAILABLE"
68+
]
69+
}
70+
}
71+
]
72+
})"};
73+
74+
// Allocate string with buffer large enough to hold the formatted json config
75+
auto service_config = std::string(kServiceConfigJson.size(), '\0');
76+
// Prior to C++17, need to explicitly cast away constness from `data()` buffer
77+
float initial_backoff = options.retry_policy_initial_backoff.count();
78+
float max_backoff = options.retry_policy_max_backoff.count();
79+
float backoff_multiplier = options.retry_policy_backoff_multiplier;
80+
std::snprintf(
81+
const_cast<decltype(service_config)::value_type *>(service_config.data()),
82+
service_config.size(),
83+
kServiceConfigJson.data(),
84+
options.retry_policy_max_attempts,
85+
std::min(std::max(initial_backoff, 0.f), 999999999.f),
86+
std::min(std::max(max_backoff, 0.f), 999999999.f),
87+
std::min(std::max(backoff_multiplier, 0.f), 999999999.f));
88+
89+
grpc_arguments.SetServiceConfigJSON(service_config);
90+
3591
if (!options.use_ssl_credentials) {
3692
channel = CreateCustomChannel(options.endpoint,
3793
InsecureChannelCredentials(),
@@ -55,7 +111,19 @@ std::shared_ptr<Channel>
55111
std::unique_ptr<ClientContext>
56112
GrpcClient::MakeClientContext(const std::string& agent_id,
57113
const std::string& saas) {
114+
// Forward to the version with custom deadline using environment variable
115+
return MakeClientContext(agent_id, saas, get_grpc_deadline_seconds());
116+
}
117+
118+
/**
119+
* Create gRPC client context to call RPC with custom deadline.
120+
*/
121+
std::unique_ptr<ClientContext>
122+
GrpcClient::MakeClientContext(const std::string& agent_id,
123+
const std::string& saas,
124+
int deadline_seconds) {
58125
std::unique_ptr<ClientContext> context = std::make_unique<ClientContext>();
126+
context->set_deadline(system_clock::now() + seconds(deadline_seconds));
59127
context->AddMetadata("nsolid-agent-id", agent_id);
60128
if (!saas.empty()) {
61129
context->AddMetadata("nsolid-saas-token", saas);

‎agents/grpc/src/grpc_client.h‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,14 @@ class GrpcClient {
6868
static std::unique_ptr<::grpc::ClientContext>
6969
MakeClientContext(const std::string& agent_id, const std::string& saas);
7070

71+
/**
72+
* Create gRPC client context to call RPC with custom deadline.
73+
*/
74+
static std::unique_ptr<::grpc::ClientContext>
75+
MakeClientContext(const std::string& agent_id,
76+
const std::string& saas,
77+
int deadline_seconds);
78+
7179
/**
7280
* Create N|Solid service stub to communicate with the N|Solid Console.
7381
*/
@@ -148,6 +156,9 @@ class GrpcClient {
148156
}
149157
};
150158

159+
// Helper function to get configurable gRPC deadline
160+
int get_grpc_deadline_seconds();
161+
151162
} // namespace grpc
152163
} // namespace nsolid
153164
} // namespace node

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
'BUILDING_LIBCURL',
6464
'ENABLE_ASYNC_EXPORT',
6565
'OPENTELEMETRY_STL_VERSION=2020',
66+
'ENABLE_OTLP_RETRY_PREVIEW',
6667
],
6768
'dependencies': [
6869
'../protobuf/protobuf.gyp:protobuf',
@@ -75,6 +76,7 @@
7576
'defines': [
7677
'ENABLE_ASYNC_EXPORT',
7778
'OPENTELEMETRY_STL_VERSION=2020',
79+
'ENABLE_OTLP_RETRY_PREVIEW',
7880
],
7981
'include_dirs': [
8082
'api/include',

0 commit comments

Comments
 (0)