Datadog intake blackhole

This PR introduces support for Datadog intake v2 in a new blackhole,
the goal being to accumulate trace-agent metrics that are only
transmitted to dogstatsd and then out to the intake. The construction
method here is intended to make it easy to add support for new intake
versions.

Signed-off-by: Brian L. Troutwine <[email protected]>

Author: blt

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Added
 - Stable throttle now has a 'timeout' configuration parameter to model IO done
   with timeout.
+- Added a 'Datadog Intake' blackhole, supporting only the v2 metrics protobuf
+  protocol for now. This captures metrics submitted into capture files.
 
 ## [0.28.0]
 ## Added

Cargo.lock

@@ -21,7 +21,8 @@ metrics-exporter-prometheus = { version = "0.15", default-features = false, feat
   "http-listener",
   "uds-listener",
 ] }
-prost = "0.13"
+prost = { version = "0.13" }
+prost-build = { version = "0.13" }
 rand = { version = "0.9", default-features = false }
 rustc-hash = { version = "1.1" }
 serde = { version = "1.0", features = ["std", "derive"] }
@@ -81,9 +82,12 @@ missing_docs = "deny"
 warnings = "deny"
 
 [profile.release]
-lto = true        # Optimize our binary at link stage.
-codegen-units = 1 # Increases compile time but improves optimization alternatives.
-opt-level = 3     # Optimize with 'all' optimization flipped on. May produce larger binaries than 's' or 'z'.
+# lto = true        # Optimize our binary at link stage.
+# codegen-units = 1 # Increases compile time but improves optimization alternatives.
+# opt-level = 3     # Optimize with 'all' optimization flipped on. May produce larger binaries than 's' or 'z'.
+lto = false        # Optimize our binary at link stage.
+codegen-units = 8 # Increases compile time but improves optimization alternatives.
+opt-level = 's'     # Optimize with 'all' optimization flipped on. May produce larger binaries than 's' or 'z'.
 panic = "abort"
 debug = 2                       # keep symbols in release too
 split-debuginfo = "unpacked"

Cargo.toml

@@ -8,3 +8,4 @@ lint:
     - COMMENTS
   ignore:
     - integration
+    - lading/proto/agent_payload.proto

buf.yaml

@@ -14,9 +14,6 @@ keywords = ["random_test", "generator"]
 categories = ["development-tools::profiling"]
 description = "A tool for load testing daemons."
 
-[package.metadata.cargo-machete]
-ignored = ["prost"]
-
 [dependencies]
 lading-capture = { version = "0.2", path = "../lading_capture" }
 lading-payload = { version = "0.1", path = "../lading_payload" }
@@ -106,6 +103,9 @@ tempfile = "3.23"
 proptest = { workspace = true }
 tower-test = "0.4"
 
+[build-dependencies]
+prost-build = { workspace = true }
+
 [features]
 default = []
 logrotate_fs = ["fuser"]

lading/Cargo.toml

@@ -0,0 +1,13 @@
+//! Build script for `lading` crate.
+
+fn main() -> std::io::Result<()> {
+    println!("cargo:rerun-if-changed=proto/");
+
+    let includes = ["proto/"];
+    prost_build::Config::new()
+        .out_dir("src/proto/")
+        .protoc_arg("--experimental_allow_proto3_optional")
+        .compile_protos(&["proto/agent_payload.proto"], &includes)?;
+
+    Ok(())
+}

lading/build.rs

@@ -0,0 +1,125 @@
+// Vendored from
+// https://github.com/DataDog/agent-payload/blob/master/proto/metrics/agent_payload.proto
+// SHA 9e27b06757db73581484382d8c0772b3e7f1e009.
+
+syntax = "proto3";
+
+package datadog.agentpayload;
+
+message CommonMetadata {
+  string agent_version = 1;
+  string timezone = 2;
+  double current_epoch = 3;
+  string internal_ip = 4;
+  string public_ip = 5;
+  string api_key = 6;
+}
+
+message Origin {
+  reserved 1,2,3;
+  uint32 origin_product = 4;
+  uint32 origin_category = 5;
+  uint32 origin_service = 6;
+}
+
+// Metadata is used in both the MetricSeries and Sketch messages defined below.
+message Metadata {
+  Origin origin = 1;
+}
+
+message MetricPayload {
+  enum MetricType {
+    UNSPECIFIED = 0;
+    COUNT = 1;
+    RATE = 2;
+    GAUGE = 3;
+  }
+
+  message MetricPoint {
+    // metric value
+    double value = 1;
+    // timestamp for this value in seconds since the UNIX epoch
+    int64 timestamp = 2;
+  }
+
+  message Resource {
+    string type = 1;
+    string name = 2;
+  }
+
+  message MetricSeries {
+    // Resources this series applies to; include at least
+    // { type="host", name=<hostname> }
+    repeated Resource resources = 1;
+    // metric name
+    string metric = 2;
+    // tags for this metric
+    repeated string tags = 3;
+    // data points for this metric
+    repeated MetricPoint points = 4;
+    // type of metric
+    MetricType type = 5;
+    // metric unit name
+    string unit = 6;
+    // source of this metric (check name, etc.)
+    string source_type_name = 7;
+    // interval, in seconds, between samples of this metric
+    int64 interval = 8;
+    // Metrics origin metadata
+    Metadata metadata = 9;
+  }
+  repeated MetricSeries series = 1;
+}
+
+message EventsPayload {
+  message Event {
+    string title = 1;
+    string text = 2;
+    int64 ts = 3;
+    string priority = 4;
+    string host = 5;
+    repeated string tags = 6;
+    string alert_type = 7;
+    string aggregation_key = 8;
+    string source_type_name = 9;
+  }
+  repeated Event events = 1;
+  CommonMetadata metadata = 2;
+}
+
+message SketchPayload {
+  message Sketch {
+    message Distribution {
+      int64 ts = 1;
+      int64 cnt = 2;
+      double min = 3;
+      double max = 4;
+      double avg = 5;
+      double sum = 6;
+      repeated double v = 7;
+      repeated uint32 g = 8;
+      repeated uint32 delta = 9;
+      repeated double buf = 10;
+    }
+    message Dogsketch {
+      int64 ts = 1;
+      int64 cnt = 2;
+      double min = 3;
+      double max = 4;
+      double avg = 5;
+      double sum = 6;
+      repeated sint32 k = 7;
+      repeated uint32 n = 8;
+    }
+    string metric = 1;
+    string host = 2;
+    repeated Distribution distributions = 3;
+    repeated string tags = 4;
+    reserved 5, 6;
+    reserved "distributionsK", "distributionsC";
+    repeated Dogsketch dogsketches = 7;
+    Metadata metadata = 8;
+  }
+  repeated Sketch sketches = 1;
+  CommonMetadata metadata = 2;
+}

lading/proto/agent_payload.proto

@@ -427,7 +427,7 @@ async fn inner_main(
                 capture_manager
                     .start()
                     .await
-                    .expect("failed to start capture manager");
+                    .expect("capture manager suffered unexpected failure");
             });
             capture_manager_handle = Some(handle);
         }

lading/src/bin/lading.rs

@@ -8,6 +8,7 @@
 use serde::{Deserialize, Serialize};
 
 mod common;
+pub mod datadog;
 pub mod http;
 pub mod otlp;
 pub mod splunk_hec;
@@ -23,6 +24,9 @@ pub enum Error {
     /// See [`crate::blackhole::tcp::Error`] for details.
     #[error(transparent)]
     Tcp(tcp::Error),
+    /// See [`crate::blackhole::datadog::Error`] for details.
+    #[error(transparent)]
+    Datadog(datadog::Error),
     /// See [`crate::blackhole::http::Error`] for details.
     #[error(transparent)]
     Http(http::Error),
@@ -75,6 +79,8 @@ pub struct General {
 pub enum Inner {
     /// See [`crate::blackhole::tcp::Config`] for details.
     Tcp(tcp::Config),
+    /// See [`crate::blackhole::datadog::Config`] for details.
+    Datadog(datadog::Config),
     /// See [`crate::blackhole::http::Config`] for details.
     Http(http::Config),
     /// See [`crate::blackhole::splunk_hec::Config`] for details.
@@ -99,6 +105,8 @@ pub enum Inner {
 pub enum Server {
     /// See [`crate::blackhole::tcp::Tcp`] for details.
     Tcp(tcp::Tcp),
+    /// See [`crate::blackhole::datadog::Datadog`] for details.
+    Datadog(datadog::Datadog),
     /// See [`crate::blackhole::http::Http`] for details.
     Http(http::Http),
     /// See [`crate::blackhole::splunk_hec::SplunkHec`] for details.
@@ -128,6 +136,9 @@ impl Server {
     pub fn new(config: Config, shutdown: lading_signal::Watcher) -> Result<Self, Error> {
         let server = match config.inner {
             Inner::Tcp(conf) => Self::Tcp(tcp::Tcp::new(config.general, &conf, shutdown)),
+            Inner::Datadog(conf) => {
+                Self::Datadog(datadog::Datadog::new(config.general, conf, shutdown))
+            }
             Inner::Http(conf) => {
                 Self::Http(http::Http::new(config.general, &conf, shutdown).map_err(Error::Http)?)
             }
@@ -163,6 +174,7 @@ impl Server {
     pub async fn run(self) -> Result<(), Error> {
         match self {
             Server::Tcp(inner) => inner.run().await.map_err(Error::Tcp),
+            Server::Datadog(inner) => inner.run().await.map_err(Error::Datadog),
             Server::Http(inner) => inner.run().await.map_err(Error::Http),
             Server::Udp(inner) => Box::pin(inner.run()).await.map_err(Error::Udp),
             Server::UnixStream(inner) => inner.run().await.map_err(Error::UnixStream),