Solve concurrency issue when telemetry endpoint is unreachable

Author: pablomartinezbernardo

src/datadog/telemetry/telemetry_impl.cpp

@@ -226,6 +226,7 @@ void Telemetry::schedule_tasks() {
 }
 
 Telemetry::~Telemetry() {
+  shutting_down_.store(true, std::memory_order_release);
   if (!tasks_.empty()) {
     cancel_tasks(tasks_);
     capture_metrics();
@@ -736,6 +737,9 @@ void Telemetry::capture_metrics() {
 
 void Telemetry::log(std::string message, telemetry::LogLevel level,
                     Optional<std::string> stacktrace) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   auto timestamp = std::chrono::duration_cast<std::chrono::seconds>(
                        clock_().wall.time_since_epoch())
                        .count();
@@ -750,6 +754,9 @@ void Telemetry::increment_counter(const Counter& id) {
 
 void Telemetry::increment_counter(const Counter& id,
                                   const std::vector<std::string>& tags) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   std::lock_guard l{counter_mutex_};
   counters_[{id, tags}] += 1;
 }
@@ -760,6 +767,9 @@ void Telemetry::decrement_counter(const Counter& id) {
 
 void Telemetry::decrement_counter(const Counter& id,
                                   const std::vector<std::string>& tags) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   std::lock_guard l{counter_mutex_};
   auto& v = counters_[{id, tags}];
   if (v > 0) v -= 1;
@@ -772,6 +782,9 @@ void Telemetry::set_counter(const Counter& id, uint64_t value) {
 void Telemetry::set_counter(const Counter& id,
                             const std::vector<std::string>& tags,
                             uint64_t value) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   std::lock_guard l{counter_mutex_};
   counters_[{id, tags}] = value;
 }
@@ -782,6 +795,9 @@ void Telemetry::set_rate(const Rate& id, uint64_t value) {
 
 void Telemetry::set_rate(const Rate& id, const std::vector<std::string>& tags,
                          uint64_t value) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   std::lock_guard l{rate_mutex_};
   rates_[{id, tags}] = value;
 }
@@ -793,6 +809,9 @@ void Telemetry::add_datapoint(const Distribution& id, uint64_t value) {
 void Telemetry::add_datapoint(const Distribution& id,
                               const std::vector<std::string>& tags,
                               uint64_t value) {
+  if (shutting_down_.load(std::memory_order_acquire)) {
+    return;
+  }
   std::lock_guard l{distributions_mutex_};
   distributions_[{id, tags}].emplace_back(value);
 }

src/datadog/telemetry/telemetry_impl.h

@@ -9,6 +9,7 @@
 #include <datadog/telemetry/metrics.h>
 #include <datadog/tracer_signature.h>
 
+#include <atomic>
 #include <mutex>
 
 #include "json.hpp"
@@ -39,6 +40,7 @@ class Telemetry final {
   std::shared_ptr<tracing::HTTPClient> http_client_;
   tracing::Clock clock_;
   std::shared_ptr<tracing::EventScheduler> scheduler_;
+  std::atomic<bool> shutting_down_{false};
 
   /// Counter
   std::mutex counter_mutex_;