feat(telemetry): implement app-extended-heartbeat event

include/datadog/environment.h

@@ -69,6 +69,7 @@ namespace environment {
   MACRO(DD_VERSION, STRING, "")                                                \
   MACRO(DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED, BOOLEAN, true)            \
   MACRO(DD_TELEMETRY_HEARTBEAT_INTERVAL, DECIMAL, 10)                          \
+  MACRO(DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL, INT, 86400)                  \
   MACRO(DD_TELEMETRY_METRICS_ENABLED, BOOLEAN, true)                           \
   MACRO(DD_TELEMETRY_METRICS_INTERVAL_SECONDS, DECIMAL, 60)                    \
   MACRO(DD_TELEMETRY_DEBUG, BOOLEAN, false)                                    \

include/datadog/telemetry/configuration.h

@@ -29,6 +29,10 @@ struct Configuration {
   // Interval at which the heartbeat payload will be sent.
   // Can be overriden by `DD_TELEMETRY_HEARTBEAT_INTERVAL` environment variable.
   tracing::Optional<double> heartbeat_interval_seconds;
+  // Interval at which the extended heartbeat payload will be sent.
+  // Can be overriden by `DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL` environment
+  // variable. Default: 86400 seconds (24 hours).
+  tracing::Optional<int> extended_heartbeat_interval_seconds;
   // `integration_name` is the name of the product integrating this library.
   // Example: "nginx", "envoy" or "istio".
   tracing::Optional<std::string> integration_name;
@@ -52,6 +56,7 @@ struct FinalizedConfiguration {
   bool report_logs;
   std::chrono::steady_clock::duration metrics_interval;
   std::chrono::steady_clock::duration heartbeat_interval;
+  std::chrono::steady_clock::duration extended_heartbeat_interval;
   std::string integration_name;
   std::string integration_version;
   std::vector<Product> products;

src/datadog/telemetry/configuration.cpp

@@ -48,6 +48,16 @@ tracing::Expected<Configuration> load_telemetry_env_config() {
     env_cfg.heartbeat_interval_seconds = *maybe_value;
   }
 
+  if (auto extended_heartbeat_interval_seconds =
+          lookup(environment::DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL)) {
+    auto maybe_value = parse_double(*extended_heartbeat_interval_seconds);
+    if (auto error = maybe_value.if_error()) {
+      return *error;
+    }
+    env_cfg.extended_heartbeat_interval_seconds =
+        static_cast<int>(*maybe_value);
+  }
+
   return env_cfg;
 }
 
@@ -112,6 +122,19 @@ tracing::Expected<FinalizedConfiguration> finalize_config(
       std::chrono::duration_cast<std::chrono::milliseconds>(
           std::chrono::duration<double>(heartbeat_interval.second));
 
+  // extended_heartbeat_interval_seconds
+  auto extended_heartbeat_interval =
+      pick(env_config->extended_heartbeat_interval_seconds,
+           user_config.extended_heartbeat_interval_seconds, 86400);
+  if (extended_heartbeat_interval.second <= 0) {
+    return Error{
+        Error::Code::OUT_OF_RANGE_INTEGER,
+        "Telemetry extended heartbeat interval must be a positive value"};
+  }
+  result.extended_heartbeat_interval =
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          std::chrono::seconds(extended_heartbeat_interval.second));
+
   // integration_name
   std::tie(origin, result.integration_name) =
       pick(env_config->integration_name, user_config.integration_name,

src/datadog/telemetry/telemetry_impl.cpp

@@ -227,6 +227,11 @@ void Telemetry::schedule_tasks() {
     tasks_.emplace_back(scheduler_->schedule_recurring_event(
         config_.metrics_interval, [this]() mutable { capture_metrics(); }));
   }
+
+  tasks_.emplace_back(scheduler_->schedule_recurring_event(
+      config_.extended_heartbeat_interval, [this]() {
+        send_payload("app-extended-heartbeat", extended_heartbeat_payload());
+      }));
 }
 
 Telemetry::~Telemetry() {
@@ -678,6 +683,28 @@ std::string Telemetry::app_started_payload() {
   return batch.dump();
 }
 
+std::string Telemetry::extended_heartbeat_payload() {
+  auto configuration_json = nlohmann::json::array();
+
+  for (const auto& product : config_.products) {
+    for (const auto& [_, config_metadatas] : product.configurations) {
+      for (const auto& config_metadata : config_metadatas) {
+        configuration_json.emplace_back(
+            generate_configuration_field(config_metadata));
+      }
+    }
+  }
+
+  auto extended_hb_msg = nlohmann::json{
+      {"request_type", "app-extended-heartbeat"},
+      {"payload", nlohmann::json{{"configuration", configuration_json}}},
+  };
+
+  auto batch = generate_telemetry_body("message-batch");
+  batch["payload"] = nlohmann::json::array({std::move(extended_hb_msg)});
+  return batch.dump();
+}
+
 nlohmann::json Telemetry::generate_telemetry_body(std::string request_type) {
   std::time_t tracer_time = std::chrono::duration_cast<std::chrono::seconds>(
                                 clock_().wall.time_since_epoch())

src/datadog/telemetry/telemetry_impl.h

@@ -152,6 +152,9 @@ class Telemetry final {
   // Constructs a messsage-batch containing `app-heartbeat`, and if metrics
   // have been modified, a `generate-metrics` message.
   std::string heartbeat_and_telemetry();
+  // Constructs a message-batch containing `app-extended-heartbeat` with the
+  // full configuration payload.
+  std::string extended_heartbeat_payload();
   // Constructs a message-batch containing `app-closing`, and if metrics have
   // been modified, a `generate-metrics` message.
   std::string app_closing_payload();

supported-configurations.json

@@ -119,6 +119,13 @@
         "type": "BOOLEAN"
       }
     ],
+    "DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL": [
+      {
+        "default": "86400",
+        "implementation": "A",
+        "type": "INT"
+      }
+    ],
     "DD_TELEMETRY_HEARTBEAT_INTERVAL": [
       {
         "default": "10",

test/telemetry/test_configuration.cpp

@@ -21,6 +21,7 @@ TELEMETRY_CONFIGURATION_TEST("defaults") {
   CHECK(cfg->report_metrics == true);
   CHECK(cfg->metrics_interval == 60s);
   CHECK(cfg->heartbeat_interval == 10s);
+  CHECK(cfg->extended_heartbeat_interval == 86400s);
   CHECK(cfg->install_id.has_value() == false);
   CHECK(cfg->install_type.has_value() == false);
   CHECK(cfg->install_time.has_value() == false);

test/telemetry/test_telemetry.cpp

@@ -46,19 +46,27 @@ struct FakeEventScheduler : public EventScheduler {
   size_t count_tasks = 0;
   std::function<void()> heartbeat_callback = nullptr;
   std::function<void()> metrics_callback = nullptr;
+  std::function<void()> extended_heartbeat_callback = nullptr;
   Optional<std::chrono::steady_clock::duration> heartbeat_interval;
   Optional<std::chrono::steady_clock::duration> metrics_interval;
+  Optional<std::chrono::steady_clock::duration> extended_heartbeat_interval;
   bool cancelled = false;
 
   // NOTE: White box testing. This is a limitation of the event scheduler API.
+  // Tasks are registered in order: heartbeat (0), metrics (1, if enabled),
+  // extended heartbeat (last).
   Cancel schedule_recurring_event(std::chrono::steady_clock::duration interval,
                                   std::function<void()> callback) override {
     if (count_tasks == 0) {
       heartbeat_callback = callback;
       heartbeat_interval = interval;
-    } else if (count_tasks == 1) {
+    } else if (interval <= std::chrono::minutes(1)) {
+      // Metrics interval is <= 60s; extended heartbeat is much larger.
       metrics_callback = callback;
       metrics_interval = interval;
+    } else {
+      extended_heartbeat_callback = callback;
+      extended_heartbeat_interval = interval;
     }
     count_tasks++;
     return [this]() { cancelled = true; };
@@ -74,6 +82,11 @@ struct FakeEventScheduler : public EventScheduler {
     metrics_callback();
   }
 
+  void trigger_extended_heartbeat() {
+    assert(extended_heartbeat_callback != nullptr);
+    extended_heartbeat_callback();
+  }
+
   std::string config() const override {
     return nlohmann::json::object({{"type", "FakeEventScheduler"}}).dump();
   }
@@ -391,6 +404,49 @@ TELEMETRY_IMPLEMENTATION_TEST("Tracer telemetry API") {
     REQUIRE(find_payload(message_batch["payload"], "app-heartbeat"));
   }
 
+  SECTION("generates an extended heartbeat with configuration") {
+    client->clear();
+
+    Product product;
+    product.name = Product::Name::tracing;
+    product.enabled = true;
+    product.version = tracer_version;
+    product.configurations =
+        std::unordered_map<ConfigName, std::vector<ConfigMetadata>>{
+            {ConfigName::SERVICE_NAME,
+             {ConfigMetadata(ConfigName::SERVICE_NAME, "my-service",
+                             ConfigMetadata::Origin::CODE)}},
+        };
+
+    Configuration cfg;
+    cfg.products.emplace_back(std::move(product));
+
+    auto scheduler2 = std::make_shared<FakeEventScheduler>();
+    Telemetry telemetry2{*finalize_config(cfg),
+                         tracer_signature,
+                         logger,
+                         client,
+                         scheduler2,
+                         *url};
+
+    client->clear();
+    scheduler2->trigger_extended_heartbeat();
+
+    auto message_batch = nlohmann::json::parse(client->request_body);
+    REQUIRE(is_valid_telemetry_payload(message_batch));
+
+    auto ext_hb =
+        find_payload(message_batch["payload"], "app-extended-heartbeat");
+    REQUIRE(ext_hb.has_value());
+
+    auto& config_payload = (*ext_hb)["payload"]["configuration"];
+    REQUIRE(config_payload.is_array());
+    REQUIRE(config_payload.size() == 1);
+    CHECK(config_payload[0]["name"] == "service");
+    CHECK(config_payload[0]["value"] == "my-service");
+    CHECK(config_payload[0]["origin"] == "code");
+  }
+
   SECTION("metrics reporting") {
     SECTION("counters are correctly serialized in generate-metrics payload") {
       client->clear();