Merge branch 'main' into dmehala/origin-detection

Author: dmehala

.github/workflows/codeql-analysis.yml

@@ -25,7 +25,7 @@ jobs:
       uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
 
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@1b549b9259bda1cb5ddde3b41741a82a2d15a841 # v3.28.13
+      uses: github/codeql-action/init@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16
       with:
         languages: ${{ matrix.language }}
 
@@ -36,4 +36,4 @@ jobs:
       run: bin/cmake-build --preset=ci-codeql
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@1b549b9259bda1cb5ddde3b41741a82a2d15a841 # v3.28.13
+      uses: github/codeql-action/analyze@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16

src/datadog/curl.cpp

@@ -214,7 +214,7 @@ class CurlImpl {
   };
 
   void run();
-  void handle_message(const CURLMsg &, std::unique_lock<std::mutex> &);
+  void handle_message(const CURLMsg &);
   CURLcode log_on_error(CURLcode result);
   CURLMcode log_on_error(CURLMcode result);
 
@@ -386,16 +386,13 @@ Expected<void> CurlImpl::post(
         handle.get(), (url.scheme + "://" + url.authority + url.path).c_str()));
   }
 
-  std::list<CURL *> node;
-  node.push_back(handle.get());
   {
     std::lock_guard<std::mutex> lock(mutex_);
-    new_handles_.splice(new_handles_.end(), node);
-
-    (void)headers.release();
-    (void)handle.release();
-    (void)request.release();
+    new_handles_.emplace_back(handle.get());
   }
+  std::ignore = headers.release();
+  std::ignore = handle.release();
+  std::ignore = request.release();
 
   log_on_error(curl_.multi_wakeup(multi_handle_));
 
@@ -405,6 +402,8 @@ Expected<void> CurlImpl::post(
 }
 
 void CurlImpl::drain(std::chrono::steady_clock::time_point deadline) {
+  log_on_error(curl_.multi_wakeup(multi_handle_));
+
   std::unique_lock<std::mutex> lock(mutex_);
   no_requests_.wait_until(lock, deadline, [this]() {
     return num_active_handles_ == 0 && new_handles_.empty();
@@ -465,31 +464,32 @@ CURLMcode CurlImpl::log_on_error(CURLMcode result) {
 }
 
 void CurlImpl::run() {
-  int num_messages_remaining;
-  CURLMsg *message;
-  constexpr int max_wait_milliseconds = 10000;
-  std::unique_lock<std::mutex> lock(mutex_);
+  int num_active_handles = 0;
+  int num_messages_remaining = 0;
 
-  for (;;) {
-    log_on_error(curl_.multi_perform(multi_handle_, &num_active_handles_));
-    if (num_active_handles_ == 0) {
-      no_requests_.notify_all();
-    }
+  bool shutting_down = false;
+  CURLMsg *message = nullptr;
+  constexpr int max_wait_milliseconds = 10'000;
 
-    // If a request is done or errored out, curl will enqueue a "message" for
-    // us to handle.  Handle any pending messages.
-    while ((message = curl_.multi_info_read(multi_handle_,
-                                            &num_messages_remaining))) {
-      handle_message(*message, lock);
-    }
-    lock.unlock();
-    log_on_error(curl_.multi_poll(multi_handle_, nullptr, 0,
-                                  max_wait_milliseconds, nullptr));
+  std::unique_lock<std::mutex> lock(mutex_, std::defer_lock);
+  std::list<CURL *> handles_to_process;
+
+  while (true) {
     lock.lock();
+    shutting_down = shutting_down_;
+
+    handles_to_process.splice(handles_to_process.begin(), new_handles_);
+    assert(new_handles_.empty());
+
+    num_active_handles_ =
+        num_active_handles + static_cast<int>(handles_to_process.size());
+    lock.unlock();
+
+    no_requests_.notify_all();
 
     // New requests might have been added while we were sleeping.
-    for (; !new_handles_.empty(); new_handles_.pop_front()) {
-      CURL *handle = new_handles_.front();
+    for (; !handles_to_process.empty(); handles_to_process.pop_front()) {
+      CURL *handle = handles_to_process.front();
       char *user_data;
       if (log_on_error(curl_.easy_getinfo_private(handle, &user_data)) !=
           CURLE_OK) {
@@ -528,9 +528,18 @@ void CurlImpl::run() {
       request_handles_.insert(handle);
     }
 
-    if (shutting_down_) {
-      break;
+    if (shutting_down) break;
+
+    log_on_error(curl_.multi_perform(multi_handle_, &num_active_handles));
+
+    // If a request is done or errored out, curl will enqueue a "message" for
+    // us to handle. Handle any pending messages.
+    while ((message = curl_.multi_info_read(multi_handle_,
+                                            &num_messages_remaining))) {
+      handle_message(*message);
     }
+    log_on_error(curl_.multi_poll(multi_handle_, nullptr, 0,
+                                  max_wait_milliseconds, nullptr));
   }
 
   // We're shutting down.  Clean up any remaining request handles.
@@ -548,8 +557,7 @@ void CurlImpl::run() {
   request_handles_.clear();
 }
 
-void CurlImpl::handle_message(const CURLMsg &message,
-                              std::unique_lock<std::mutex> &lock) {
+void CurlImpl::handle_message(const CURLMsg &message) {
   if (message.msg != CURLMSG_DONE) {
     return;
   }
@@ -571,21 +579,17 @@ void CurlImpl::handle_message(const CURLMsg &message,
     error_message += curl_.easy_strerror(result);
     error_message += "): ";
     error_message += request.error_buffer;
-    lock.unlock();
     request.on_error(
         Error{Error::CURL_REQUEST_FAILURE, std::move(error_message)});
-    lock.lock();
   } else {
     long status;
     if (log_on_error(curl_.easy_getinfo_response_code(request_handle,
                                                       &status)) != CURLE_OK) {
       status = -1;
     }
     HeaderReader reader(&request.response_headers_lower);
-    lock.unlock();
     request.on_response(static_cast<int>(status), reader,
                         std::move(request.response_body));
-    lock.lock();
   }
 
   log_on_error(curl_.multi_remove_handle(multi_handle_, request_handle));

src/datadog/datadog_agent.cpp

@@ -246,8 +246,24 @@ void DatadogAgent::flush() {
     return;
   }
 
+  // Ideally:
+  /*auto [encode_result, duration] = mesure([&trace_chunks] {*/
+  /*  std::string body;*/
+  /*  msgpack_encode(body, trace_chunks);*/
+  /*});*/
+
   std::string body;
+
+  auto beg = std::chrono::steady_clock::now();
   auto encode_result = msgpack_encode(body, trace_chunks);
+  auto end = std::chrono::steady_clock::now();
+
+  telemetry::distribution::add(
+      metrics::tracer::trace_chunk_serialization_duration,
+      std::chrono::duration_cast<std::chrono::microseconds>(end - beg).count());
+  telemetry::distribution::add(metrics::tracer::trace_chunk_serialized_bytes,
+                               static_cast<uint64_t>(body.size()));
+
   if (auto* error = encode_result.if_error()) {
     logger_->log_error(*error);
     return;
@@ -335,11 +351,17 @@ void DatadogAgent::flush() {
   };
 
   telemetry::counter::increment(metrics::tracer::api::requests);
+  telemetry::distribution::add(metrics::tracer::api::bytes_sent,
+                               static_cast<uint64_t>(body.size()));
+
   auto post_result =
       http_client_->post(traces_endpoint_, std::move(set_request_headers),
                          std::move(body), std::move(on_response),
                          std::move(on_error), clock_().tick + request_timeout_);
   if (auto* error = post_result.if_error()) {
+    // NOTE(@dmehala): `technical` is a better kind of errors.
+    telemetry::counter::increment(metrics::tracer::api::errors,
+                                  {"type:network"});
     logger_->log_error(
         error->with_prefix("Unexpected error submitting traces: "));
   }

src/datadog/telemetry/telemetry_impl.cpp

@@ -17,6 +17,40 @@ using namespace datadog::tracing;
 using namespace std::chrono_literals;
 
 namespace datadog::telemetry {
+namespace internal_metrics {
+
+/// The number of logs created with a given log level. Useful for calculating
+/// impact for other features (automatic sending of logs). Levels should be one
+/// of `debug`, `info`, `warn`, `error`, `critical`.
+const telemetry::Counter logs_created{"logs_created", "general", true};
+
+/// The number of requests sent to the api endpoint in the agent that errored,
+/// tagged by the error type (e.g. `type:timeout`, `type:network`,
+/// `type:status_code`) and Endpoint (`endpoint:agent`, `endpoint:agentless`).
+const telemetry::Counter errors{"telemetry_api.errors", "telemetry", true};
+
+/// The number of requests sent to a telemetry endpoint, regardless of success,
+/// tagged by the endpoint (`endpoint:agent`, `endpoint:agentless`).
+const telemetry::Counter requests{"telemetry_api.requests", "telemetry", true};
+
+/// The number of responses received from the endpoint, tagged with status code
+/// (`status_code:200`, `status_code:404`) and endpoint (`endpoint:agent`,
+/// `endpoint:agentless`).
+const telemetry::Counter responses{"telemetry_api.responses", "telemetry",
+                                   true};
+
+/// The size of the payload sent to the stats endpoint in bytes, tagged by the
+/// endpoint (`endpoint:agent`, `endpoint:agentless`).
+const telemetry::Distribution bytes_sent{"telemetry_api.bytes", "telemetry",
+                                         true};
+
+/// The time it takes to send the payload sent to the endpoint in ms, tagged by
+/// the endpoint (`endpoint:agent`, `endpoint:agentless`).
+const telemetry::Distribution request_duration{"telemetry_api.ms", "telemetry",
+                                               true};
+
+}  // namespace internal_metrics
+
 namespace {
 
 HTTPClient::URL make_telemetry_endpoint(HTTPClient::URL url) {
@@ -174,26 +208,8 @@ Telemetry::Telemetry(FinalizedConfiguration config,
       host_info_(get_host_info()) {
   // Callback for successful telemetry HTTP requests, to examine HTTP
   // status.
-  telemetry_on_response_ = [logger = logger_](
-                               int response_status,
-                               const DictReader& /*response_headers*/,
-                               std::string response_body) {
-    if (response_status < 200 || response_status >= 300) {
-      logger->log_error([&](auto& stream) {
-        stream << "Unexpected telemetry response status " << response_status
-               << " with body (if any, starts on next line):\n"
-               << response_body;
-      });
-    }
-  };
-
-  // Callback for unsuccessful telemetry HTTP requests.
-  telemetry_on_error_ = [logger = logger_](Error error) {
-    logger->log_error(error.with_prefix(
-        "Error occurred during HTTP request for telemetry: "));
-  };
-
   send_telemetry("app-started", app_started());
+  http_client_->drain(clock_().tick + 2s);
   schedule_tasks();
 }
 
@@ -216,20 +232,23 @@ Telemetry::~Telemetry() {
     // The app-closing message is bundled with a message containing the
     // final metric values.
     send_telemetry("app-closing", app_closing());
-    http_client_->drain(clock_().tick + 1s);
+    http_client_->drain(clock_().tick + 2s);
   }
 }
 
 Telemetry::Telemetry(Telemetry&& rhs)
     : config_(std::move(rhs.config_)),
       logger_(std::move(rhs.logger_)),
-      telemetry_on_response_(std::move(rhs.telemetry_on_response_)),
-      telemetry_on_error_(std::move(rhs.telemetry_on_error_)),
       telemetry_endpoint_(std::move(rhs.telemetry_endpoint_)),
       tracer_signature_(std::move(rhs.tracer_signature_)),
       http_client_(rhs.http_client_),
       clock_(std::move(rhs.clock_)),
       scheduler_(std::move(rhs.scheduler_)),
+      counters_(std::move(rhs.counters_)),
+      counters_snapshot_(std::move(rhs.counters_snapshot_)),
+      rates_(std::move(rhs.rates_)),
+      rates_snapshot_(std::move(rhs.rates_snapshot_)),
+      distributions_(std::move(rhs.distributions_)),
       seq_id_(rhs.seq_id_),
       config_seq_ids_(rhs.config_seq_ids_),
       host_info_(rhs.host_info_) {
@@ -242,13 +261,17 @@ Telemetry& Telemetry::operator=(Telemetry&& rhs) {
     cancel_tasks(rhs.tasks_);
     std::swap(config_, rhs.config_);
     std::swap(logger_, rhs.logger_);
-    std::swap(telemetry_on_response_, rhs.telemetry_on_response_);
-    std::swap(telemetry_on_error_, rhs.telemetry_on_error_);
     std::swap(telemetry_endpoint_, rhs.telemetry_endpoint_);
     std::swap(http_client_, rhs.http_client_);
     std::swap(tracer_signature_, rhs.tracer_signature_);
     std::swap(http_client_, rhs.http_client_);
     std::swap(clock_, rhs.clock_);
+    std::swap(scheduler_, rhs.scheduler_);
+    std::swap(counters_, rhs.counters_);
+    std::swap(counters_snapshot_, rhs.counters_snapshot_);
+    std::swap(rates_, rhs.rates_);
+    std::swap(rates_snapshot_, rhs.rates_snapshot_);
+    std::swap(distributions_, rhs.distributions_);
     std::swap(seq_id_, rhs.seq_id_);
     std::swap(config_seq_ids_, rhs.config_seq_ids_);
     std::swap(host_info_, rhs.host_info_);
@@ -259,16 +282,19 @@ Telemetry& Telemetry::operator=(Telemetry&& rhs) {
 
 void Telemetry::log_error(std::string message) {
   if (!config_.report_logs) return;
+  increment_counter(internal_metrics::logs_created, {"level:error"});
   log(std::move(message), LogLevel::ERROR);
 }
 
 void Telemetry::log_error(std::string message, std::string stacktrace) {
   if (!config_.report_logs) return;
+  increment_counter(internal_metrics::logs_created, {"level:error"});
   log(std::move(message), LogLevel::ERROR, stacktrace);
 }
 
 void Telemetry::log_warning(std::string message) {
   if (!config_.report_logs) return;
+  increment_counter(internal_metrics::logs_created, {"level:warning"});
   log(std::move(message), LogLevel::WARNING);
 }
 
@@ -293,10 +319,55 @@ void Telemetry::send_telemetry(StringView request_type, std::string payload) {
     }
   };
 
-  auto post_result = http_client_->post(
-      telemetry_endpoint_, set_telemetry_headers, std::move(payload),
-      telemetry_on_response_, telemetry_on_error_, clock_().tick + 5s);
+  auto telemetry_on_response = [this, logger = logger_](
+                                   int response_status,
+                                   const DictReader& /*response_headers*/,
+                                   std::string response_body) {
+    if (response_status >= 500) {
+      increment_counter(internal_metrics::responses,
+                        {"status_code:5xx", "endpoint:agent"});
+    } else if (response_status >= 400) {
+      increment_counter(internal_metrics::responses,
+                        {"status_code:4xx", "endpoint:agent"});
+    } else if (response_status >= 300) {
+      increment_counter(internal_metrics::responses,
+                        {"status_code:3xx", "endpoint:agent"});
+    } else if (response_status >= 200) {
+      increment_counter(internal_metrics::responses,
+                        {"status_code:2xx", "endpoint:agent"});
+    } else if (response_status >= 100) {
+      increment_counter(internal_metrics::responses,
+                        {"status_code:1xx", "endpoint:agent"});
+    }
+
+    if (response_status < 200 || response_status >= 300) {
+      logger->log_error([&](auto& stream) {
+        stream << "Unexpected telemetry response status " << response_status
+               << " with body (if any, starts on next line):\n"
+               << response_body;
+      });
+    }
+  };
+
+  // Callback for unsuccessful telemetry HTTP requests.
+  auto telemetry_on_error = [this, logger = logger_](Error error) {
+    increment_counter(internal_metrics::errors,
+                      {"type:network", "endpoint:agent"});
+    logger->log_error(error.with_prefix(
+        "Error occurred during HTTP request for telemetry: "));
+  };
+
+  increment_counter(internal_metrics::requests, {"endpoint:agent"});
+  add_datapoint(internal_metrics::bytes_sent, {"endpoint:agent"},
+                payload.size());
+
+  auto post_result =
+      http_client_->post(telemetry_endpoint_, set_telemetry_headers,
+                         std::move(payload), std::move(telemetry_on_response),
+                         std::move(telemetry_on_error), clock_().tick + 5s);
   if (auto* error = post_result.if_error()) {
+    increment_counter(internal_metrics::errors,
+                      {"type:network", "endpoint:agent"});
     logger_->log_error(
         error->with_prefix("Unexpected error submitting telemetry event: "));
   }