WIP. Working API, limited metrics, missing tags for metrics

Author: cgilmour

CMakeLists.txt

@@ -110,6 +110,7 @@ target_sources(dd_trace_cpp-objects PRIVATE
     src/datadog/id_generator.cpp
     src/datadog/limiter.cpp
     src/datadog/logger.cpp
+    src/datadog/metrics.cpp
     src/datadog/msgpack.cpp
     src/datadog/null_collector.cpp
     src/datadog/parse_util.cpp
@@ -127,6 +128,7 @@ target_sources(dd_trace_cpp-objects PRIVATE
     src/datadog/tags.cpp
     src/datadog/threaded_event_scheduler.cpp
     src/datadog/tracer_config.cpp
+    src/datadog/tracer_telemetry.cpp
     src/datadog/tracer.cpp
     src/datadog/trace_id.cpp
     src/datadog/trace_sampler_config.cpp
@@ -165,6 +167,7 @@ target_sources(dd_trace_cpp-objects PUBLIC
   src/datadog/json.hpp
   src/datadog/limiter.h
   src/datadog/logger.h
+  src/datadog/metrics.h
   src/datadog/msgpack.h
   src/datadog/null_collector.h
   src/datadog/optional.h
@@ -189,6 +192,7 @@ target_sources(dd_trace_cpp-objects PUBLIC
   src/datadog/tags.h
   src/datadog/threaded_event_scheduler.h
   src/datadog/tracer_config.h
+  src/datadog/tracer_telemetry.h
   src/datadog/tracer.h
   src/datadog/trace_id.h
   src/datadog/trace_sampler_config.h

src/datadog/datadog_agent.cpp

@@ -22,13 +22,20 @@ namespace tracing {
 namespace {
 
 const StringView traces_api_path = "/v0.4/traces";
+const StringView telemetry_v2_path = "/telemetry/proxy/api/v2/apmtelemetry";
 
 HTTPClient::URL traces_endpoint(const HTTPClient::URL& agent_url) {
   auto traces_url = agent_url;
   append(traces_url.path, traces_api_path);
   return traces_url;
 }
 
+HTTPClient::URL telemetry_endpoint(const HTTPClient::URL& agent_url) {
+  auto telemetry_v2_url = agent_url;
+  append(telemetry_v2_url.path, telemetry_v2_path);
+  return telemetry_v2_url;
+}
+
 Expected<void> msgpack_encode(
     std::string& destination,
     const std::vector<DatadogAgent::TraceChunk>& trace_chunks) {
@@ -124,18 +131,30 @@ std::variant<CollectorResponse, std::string> parse_agent_traces_response(
 
 }  // namespace
 
-DatadogAgent::DatadogAgent(const FinalizedDatadogAgentConfig& config,
-                           const Clock& clock,
-                           const std::shared_ptr<Logger>& logger)
-    : clock_(clock),
+DatadogAgent::DatadogAgent(
+    const FinalizedDatadogAgentConfig& config,
+    const std::shared_ptr<TracerTelemetry>& tracer_telemetry,
+    const Clock& clock, const std::shared_ptr<Logger>& logger)
+    : tracer_telemetry_(tracer_telemetry),
+      clock_(clock),
       logger_(logger),
       traces_endpoint_(traces_endpoint(config.url)),
+      telemetry_endpoint_(telemetry_endpoint(config.url)),
       http_client_(config.http_client),
       event_scheduler_(config.event_scheduler),
       cancel_scheduled_flush_(event_scheduler_->schedule_recurring_event(
           config.flush_interval, [this]() { flush(); })),
+      cancel_heartbeat_timer_(event_scheduler_->schedule_recurring_event(
+            std::chrono::seconds(10), [this, n=0]() mutable {
+            n++;
+            tracer_telemetry_->captureMetrics();
+            if (n%6 == 0) {
+            sendHeartbeatAndTelemetry();
+            }
+            })),
       flush_interval_(config.flush_interval) {
   assert(logger_);
+  sendAppStarted();
 }
 
 DatadogAgent::~DatadogAgent() {
@@ -154,7 +173,6 @@ Expected<void> DatadogAgent::send(
 }
 
 nlohmann::json DatadogAgent::config_json() const {
-  const auto& url = traces_endpoint_;  // brevity
   const auto flush_interval_milliseconds =
       std::chrono::duration_cast<std::chrono::milliseconds>(flush_interval_)
           .count();
@@ -163,7 +181,8 @@ nlohmann::json DatadogAgent::config_json() const {
   return nlohmann::json::object({
     {"type", "datadog::tracing::DatadogAgent"},
     {"config", nlohmann::json::object({
-      {"url", (url.scheme + "://" + url.authority + url.path)},
+      {"traces_url", (traces_endpoint_.scheme + "://" + traces_endpoint_.authority + traces_endpoint_.path)},
+      {"telemetry_url", (telemetry_endpoint_.scheme + "://" + telemetry_endpoint_.authority + telemetry_endpoint_.path)},
       {"flush_interval_milliseconds", flush_interval_milliseconds},
       {"http_client", http_client_->config_json()},
       {"event_scheduler", event_scheduler_->config_json()},
@@ -251,10 +270,11 @@ void DatadogAgent::flush() {
   // request or retrieving the response.  It's invoked
   // asynchronously.
   auto on_error = [logger = logger_](Error error) {
-    logger->log_error(
-        error.with_prefix("Error occurred during HTTP request: "));
+    logger->log_error(error.with_prefix(
+        "Error occurred during HTTP request for submitting traces: "));
   };
 
+  tracer_telemetry_->trace_api_requests().inc();
   auto post_result = http_client_->post(
       traces_endpoint_, std::move(set_request_headers), std::move(body),
       std::move(on_response), std::move(on_error));
@@ -263,5 +283,87 @@ void DatadogAgent::flush() {
   }
 }
 
+void DatadogAgent::sendAppStarted() {
+  auto payload = tracer_telemetry_->appStarted();
+  auto set_request_headers = [&](DictWriter& headers) {
+    headers.set("Content-Type", "application/json");
+  };
+
+  // Callback for a successful HTTP request, to examine HTTP status.
+  auto 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 (starts on next line):\n"
+          << response_body;
+          });
+      return;
+    } else {
+      logger->log_error([&](auto& stream) {
+          stream << "Successful telemetry submission with response status " << response_status
+          << " and body (starts on next line):\n"
+          << response_body;
+          });
+    }
+
+  };
+
+  // Callback for unsuccessful HTTP request.
+  auto on_error = [logger = logger_](Error error) {
+    logger->log_error(error.with_prefix(
+        "Error occurred during HTTP request for telemetry: "));
+  };
+
+  auto post_result = http_client_->post(
+      telemetry_endpoint_, std::move(set_request_headers), std::move(payload),
+      std::move(on_response), std::move(on_error));
+  if (auto* error = post_result.if_error()) {
+    logger_->log_error(*error);
+  }
+}
+
+void DatadogAgent::sendHeartbeatAndTelemetry() {
+  auto payload = tracer_telemetry_->heartbeatAndTelemetry();
+  auto set_request_headers = [&](DictWriter& headers) {
+    headers.set("Content-Type", "application/json");
+  };
+
+  // Callback for a successful HTTP request, to examine HTTP status.
+  auto 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 (starts on next line):\n"
+          << response_body;
+          });
+      return;
+    } else {
+      logger->log_error([&](auto& stream) {
+          stream << "Successful telemetry submission with response status " << response_status
+          << " and body (starts on next line):\n"
+          << response_body;
+          });
+    }
+
+  };
+
+  // Callback for unsuccessful HTTP request.
+  auto on_error = [logger = logger_](Error error) {
+    logger->log_error(error.with_prefix(
+        "Error occurred during HTTP request for telemetry: "));
+  };
+
+  auto post_result = http_client_->post(
+      telemetry_endpoint_, std::move(set_request_headers), std::move(payload),
+      std::move(on_response), std::move(on_error));
+  if (auto* error = post_result.if_error()) {
+    logger_->log_error(*error);
+  }
+}
+
 }  // namespace tracing
 }  // namespace datadog

src/datadog/datadog_agent.h

@@ -14,6 +14,7 @@
 #include "collector.h"
 #include "event_scheduler.h"
 #include "http_client.h"
+#include "tracer_telemetry.h"
 
 namespace datadog {
 namespace tracing {
@@ -32,25 +33,31 @@ class DatadogAgent : public Collector {
 
  private:
   std::mutex mutex_;
+  std::shared_ptr<TracerTelemetry> tracer_telemetry_;
   Clock clock_;
   std::shared_ptr<Logger> logger_;
   std::vector<TraceChunk> trace_chunks_;
   HTTPClient::URL traces_endpoint_;
+  HTTPClient::URL telemetry_endpoint_;
   std::shared_ptr<HTTPClient> http_client_;
   std::shared_ptr<EventScheduler> event_scheduler_;
   EventScheduler::Cancel cancel_scheduled_flush_;
+  EventScheduler::Cancel cancel_heartbeat_timer_;
   std::chrono::steady_clock::duration flush_interval_;
 
   void flush();
 
  public:
-  DatadogAgent(const FinalizedDatadogAgentConfig&, const Clock& clock,
+  DatadogAgent(const FinalizedDatadogAgentConfig&,
+               const std::shared_ptr<TracerTelemetry>&, const Clock& clock,
                const std::shared_ptr<Logger>&);
   ~DatadogAgent();
 
   Expected<void> send(
       std::vector<std::unique_ptr<SpanData>>&& spans,
       const std::shared_ptr<TraceSampler>& response_handler) override;
+  void sendAppStarted();
+  void sendHeartbeatAndTelemetry();
 
   nlohmann::json config_json() const override;
 };

src/datadog/metrics.cpp

@@ -0,0 +1,32 @@
+#include "metrics.h"
+
+#include "json.hpp"
+
+namespace datadog {
+namespace tracing {
+
+Metric::Metric(std::string name, std::string type, bool common) : name_(name), type_(type), common_(common) {}
+std::string Metric::name() { return name_; }
+std::string Metric::type() { return type_; }
+bool Metric::common() { return common_; }
+uint64_t Metric::value() { return value_; }
+
+CounterMetric::CounterMetric(std::string name, bool common) : Metric(name, "count", common) {}
+void CounterMetric::inc() { add(1); }
+void CounterMetric::add(uint64_t amount) { value_ += amount; }
+
+GaugeMetric::GaugeMetric(std::string name, bool common) : Metric(name, "gauge", common) {}
+void GaugeMetric::set(uint64_t value) { value_ = value; }
+void GaugeMetric::inc() { add(1); }
+void GaugeMetric::add(uint64_t amount) { value_ += amount; }
+void GaugeMetric::dec() { sub(1); }
+void GaugeMetric::sub(uint64_t amount) {
+  if (amount > value_) {
+    value_ = 0;
+  } else {
+    value_ -= amount;
+  }
+}
+
+}  // namespace tracing
+}  // namespace datadog

src/datadog/metrics.h

@@ -0,0 +1,43 @@
+#pragma once
+
+#include <atomic>
+
+#include "json_fwd.hpp"
+#include "string_view.h"
+
+namespace datadog {
+namespace tracing {
+
+class Metric {
+  std::string name_;
+  std::string type_;
+  bool common_;
+  protected:
+  std::atomic<uint64_t> value_ = 0;
+  Metric(std::string name, std::string type, bool common);
+ public:
+  std::string name();
+  std::string type();
+  bool common();
+  uint64_t value();
+};
+
+class CounterMetric : public Metric {
+ public:
+  CounterMetric(std::string name, bool common);
+  void inc();
+  void add(uint64_t amount);
+};
+
+class GaugeMetric : public Metric {
+ public:
+  GaugeMetric(std::string name, bool common);
+  void set(uint64_t value);
+  void inc();
+  void add(uint64_t amount);
+  void dec();
+  void sub(uint64_t amount);
+};
+
+}  // namespace tracing
+}  // namespace datadog

src/datadog/trace_segment.cpp

@@ -83,6 +83,7 @@ void inject_trace_tags(
 TraceSegment::TraceSegment(
     const std::shared_ptr<Logger>& logger,
     const std::shared_ptr<Collector>& collector,
+    const std::shared_ptr<TracerTelemetry>& tracer_telemetry,
     const std::shared_ptr<TraceSampler>& trace_sampler,
     const std::shared_ptr<SpanSampler>& span_sampler,
     const std::shared_ptr<const SpanDefaults>& defaults,
@@ -96,6 +97,7 @@ TraceSegment::TraceSegment(
     std::unique_ptr<SpanData> local_root)
     : logger_(logger),
       collector_(collector),
+      tracer_telemetry_(tracer_telemetry),
       trace_sampler_(trace_sampler),
       span_sampler_(span_sampler),
       defaults_(defaults),
@@ -111,10 +113,14 @@ TraceSegment::TraceSegment(
           std::move(additional_datadog_w3c_tracestate)) {
   assert(logger_);
   assert(collector_);
+  assert(tracer_telemetry_);
   assert(trace_sampler_);
   assert(span_sampler_);
   assert(defaults_);
 
+  tracer_telemetry_->traces_started().inc();
+  tracer_telemetry_->active_traces().inc();
+
   register_span(std::move(local_root));
 }
 
@@ -135,12 +141,18 @@ Optional<SamplingDecision> TraceSegment::sampling_decision() const {
 Logger& TraceSegment::logger() const { return *logger_; }
 
 void TraceSegment::register_span(std::unique_ptr<SpanData> span) {
+  tracer_telemetry_->spans_started().inc();
+  tracer_telemetry_->active_spans().inc();
+
   std::lock_guard<std::mutex> lock(mutex_);
   assert(spans_.empty() || num_finished_spans_ < spans_.size());
   spans_.emplace_back(std::move(span));
 }
 
 void TraceSegment::span_finished() {
+  tracer_telemetry_->spans_finished().inc();
+  tracer_telemetry_->active_spans().dec();
+
   {
     std::lock_guard<std::mutex> lock(mutex_);
     ++num_finished_spans_;
@@ -220,6 +232,9 @@ void TraceSegment::span_finished() {
     logger_->log_error(
         error->with_prefix("Error sending spans to collector: "));
   }
+
+  tracer_telemetry_->traces_finished().inc();
+  tracer_telemetry_->active_traces().dec();
 }
 
 void TraceSegment::override_sampling_priority(int priority) {