enhancement(metrics): support for metrics v3 protocol

.gitlab-ci.yml

@@ -44,8 +44,7 @@ variables:
 
   # Converged Datadog Agent-specific variables, which control how we build the converged Datadog Agent image that we
   # publicly publish.
-  PUBLIC_DSD_VERSION: "7.77.2"
-  PUBLIC_DD_AGENT_VERSION: "7.77.2-jmx"
+  PUBLIC_DD_AGENT_IMAGE: "registry.datadoghq.com/agent:7.77.2-jmx"
 
   # Base images to copy Agent Data Plane into, depending on whether the image is meant for our internal environment or
   # public registries.

.gitlab/e2e.yml

@@ -82,7 +82,7 @@ build-bundled-agent-adp-image:
       --platform linux/amd64
       --file ./docker/Dockerfile.datadog-agent
       --tag ${SALUKI_IMAGE_REPO_BASE}/bundled-agent-adp:${CI_COMMIT_SHA}
-      --build-arg DD_AGENT_VERSION=${PUBLIC_DD_AGENT_VERSION}
+      --build-arg DD_AGENT_IMAGE=${PUBLIC_DD_AGENT_IMAGE}
       --build-arg ADP_IMAGE=${ADP_FULL_IMAGE_TAG}
       --build-arg BUILD_PROFILE=release
       --label git.repository=${CI_PROJECT_NAME}

Cargo.toml

@@ -158,6 +158,7 @@ tokio-rustls = { version = "0.26", default-features = false, features = [
   "aws_lc_rs",
 ] }
 anyhow = { version = "1", default-features = false }
+anymap3 = { version = "1", default-features = false, features = ["std"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 bytesize = { version = "2", default-features = false, features = ["serde"] }
 criterion = { version = "0.8", features = ["html_reports"] }

LICENSE-3rdparty.csv

@@ -9,6 +9,7 @@ android_system_properties,https://github.com/nical/android_system_properties,MIT
 anes,https://github.com/zrzka/anes-rs,MIT OR Apache-2.0,Robert Vojta <rvojta@me.com>
 anstyle,https://github.com/rust-cli/anstyle,MIT OR Apache-2.0,The anstyle Authors
 anyhow,https://github.com/dtolnay/anyhow,MIT OR Apache-2.0,David Tolnay <dtolnay@gmail.com>
+anymap3,https://github.com/reivilibre/anymap3,BlueOak-1.0.0 OR MIT OR Apache-2.0,"Olivier 'reivilibre' (fork maintainer) <contact@librepush.net>, Chris Morgan (original author) <rust@chrismorgan.info>"
 arc-swap,https://github.com/vorner/arc-swap,MIT OR Apache-2.0,Michal 'vorner' Vaner <vorner@vorner.cz>
 argh,https://github.com/google/argh,BSD-3-Clause,"Taylor Cramer <cramertj@google.com>, Benjamin Brittain <bwb@google.com>, Erick Tryzelaar <etryzelaar@google.com>"
 arrayvec,https://github.com/bluss/arrayvec,MIT OR Apache-2.0,bluss

Makefile

@@ -48,7 +48,7 @@ export LADING_VERSION ?= sha-d608ffbce8f8c77b147d6750b3bb6d6948af239a
 
 # Version of source repositories (Git tag) for vendored Protocol Buffers definitions.
 export PROTOBUF_SRC_REPO_DD_AGENT ?= 7.73.x
-export PROTOBUF_SRC_REPO_AGENT_PAYLOAD ?= v5.0.164
+export PROTOBUF_SRC_REPO_AGENT_PAYLOAD ?= v5.0.180
 export PROTOBUF_SRC_REPO_CONTAINERD ?= v2.2.0
 export PROTOBUF_SRC_REPO_SKETCHES_GO ?= v1.4.7
 
@@ -534,14 +534,20 @@ test-all: test test-property test-docs test-miri test-loom
 
 .PHONY: test-correctness
 test-correctness: ## Runs the complete correctness suite
-test-correctness: test-correctness-dsd-plain test-correctness-dsd-origin-detection test-correctness-otlp-metrics test-correctness-otlp-traces test-correctness-otlp-traces-ottl-filtering test-correctness-otlp-traces-ottl-transform
+test-correctness: test-correctness-dsd-plain test-correctness-dsd-plain-v3 test-correctness-dsd-origin-detection test-correctness-otlp-metrics test-correctness-otlp-traces test-correctness-otlp-traces-ottl-filtering test-correctness-otlp-traces-ottl-transform
 
 .PHONY: test-correctness-dsd-plain
 test-correctness-dsd-plain: build-ground-truth
 test-correctness-dsd-plain: ## Runs the 'dsd-plain' correctness test case
 	@echo "[*] Running 'dsd-plain' correctness test case..."
 	@target/release/ground-truth $(shell pwd)/test/correctness/dsd-plain/config.yaml
 
+.PHONY: test-correctness-dsd-plain-v3
+test-correctness-dsd-plain-v3: build-ground-truth
+test-correctness-dsd-plain-v3: ## Runs the 'dsd-plain-v3' correctness test case
+	@echo "[*] Running 'dsd-plain-v3' correctness test case..."
+	@target/release/ground-truth $(shell pwd)/test/correctness/dsd-plain-v3/config.yaml
+
 .PHONY: test-correctness-dsd-origin-detection
 test-correctness-dsd-origin-detection: build-ground-truth
 test-correctness-dsd-origin-detection: ## Runs the 'dsd-origin-detection' correctness test case

bin/correctness/datadog-intake/src/app/metrics/handlers.rs

@@ -1,4 +1,5 @@
 use axum::{body::Bytes, extract::State, http::StatusCode, Json};
+use datadog_protos::metrics::v3::Payload as V3Payload;
 use datadog_protos::metrics::{MetricPayload, SketchPayload};
 use protobuf::Message as _;
 use stele::Metric;
@@ -56,3 +57,49 @@ pub async fn handle_sketch_beta(State(state): State<MetricsState>, body: Bytes)
         }
     }
 }
+
+pub async fn handle_series_v3(State(state): State<MetricsState>, body: Bytes) -> StatusCode {
+    info!("Received v3 series payload.");
+
+    let payload = match V3Payload::parse_from_bytes(&body[..]) {
+        Ok(payload) => payload,
+        Err(e) => {
+            error!(error = %e, "Failed to parse v3 series payload.");
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+
+    match state.merge_v3_payload(payload) {
+        Ok(()) => {
+            info!("Processed v3 series payload.");
+            StatusCode::ACCEPTED
+        }
+        Err(e) => {
+            error!(error = %e, "Failed to merge v3 series payload.");
+            StatusCode::BAD_REQUEST
+        }
+    }
+}
+
+pub async fn handle_sketch_v3(State(state): State<MetricsState>, body: Bytes) -> StatusCode {
+    info!("Received v3 sketch payload.");
+
+    let payload = match V3Payload::parse_from_bytes(&body[..]) {
+        Ok(payload) => payload,
+        Err(e) => {
+            error!(error = %e, "Failed to parse v3 sketch payload.");
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+
+    match state.merge_v3_payload(payload) {
+        Ok(()) => {
+            info!("Processed v3 sketch payload.");
+            StatusCode::ACCEPTED
+        }
+        Err(e) => {
+            error!(error = %e, "Failed to merge v3 sketch payload.");
+            StatusCode::BAD_REQUEST
+        }
+    }
+}

bin/correctness/datadog-intake/src/app/metrics/mod.rs

@@ -14,5 +14,7 @@ pub fn build_metrics_router() -> Router {
         .route("/metrics/dump", get(handle_metrics_dump))
         .route("/api/v2/series", post(handle_series_v2))
         .route("/api/beta/sketches", post(handle_sketch_beta))
+        .route("/api/intake/metrics/v3/series", post(handle_series_v3))
+        .route("/api/intake/metrics/v3/sketches", post(handle_sketch_v3))
         .with_state(MetricsState::new())
 }

bin/correctness/datadog-intake/src/app/metrics/state.rs

@@ -1,5 +1,6 @@
 use std::sync::{Arc, Mutex};
 
+use datadog_protos::metrics::v3::Payload as V3Payload;
 use datadog_protos::metrics::{MetricPayload, SketchPayload};
 use saluki_error::GenericError;
 use stele::Metric;
@@ -42,4 +43,14 @@ impl MetricsState {
 
         Ok(())
     }
+
+    /// Merges the given v3 payload into the current metrics state.
+    pub fn merge_v3_payload(&self, payload: V3Payload) -> Result<(), GenericError> {
+        let metrics = Metric::try_from_v3(payload)?;
+
+        let mut data = self.metrics.lock().unwrap();
+        data.extend(metrics);
+
+        Ok(())
+    }
 }

bin/correctness/ground-truth/src/analysis/metrics/types.rs

@@ -108,15 +108,15 @@ impl NormalizedMetric {
     ///
     /// If the raw values are empty, or if the values cannot be normalized, an error is returned.
     pub fn try_from_values(
-        context: MetricContext, mut raw_values: Vec<(u64, MetricValue)>,
+        context: NormalizedMetricContext, mut raw_values: Vec<(u64, MetricValue)>,
     ) -> Result<Self, GenericError> {
         // We need to first sort the raw values by timestamp, to ensure we have proper ordering semantics.
         raw_values.sort_by(|a, b| a.0.cmp(&b.0));
         let value = try_normalize_values(&raw_values)
             .with_error_context(|| format!("Failed to normalize values for metric '{}'", context.name()))?;
 
         Ok(Self {
-            context: NormalizedMetricContext::from_stele_context(context),
+            context,
             raw_values,
             value,
         })
@@ -183,9 +183,9 @@ impl NormalizedMetrics {
         let mut aggregated_context_values = BTreeMap::new();
 
         for metric in metrics {
-            let context = metric.context();
+            let context = metric.context().clone();
             let metric_type = metric_value_type(metric.values());
-            let key = (context.clone(), metric_type);
+            let key = (NormalizedMetricContext::from_stele_context(context), metric_type);
             let context_values = aggregated_context_values.entry(key).or_insert_with(Vec::new);
             context_values.extend_from_slice(metric.values());
         }