enhancement(metrics): support for metrics v3 protocol

.vale/styles/config/vocabularies/technical/accept.txt

@@ -229,3 +229,5 @@ mpmc
 dhat
 profiler
 launchd
+varint
+serializer

Cargo.toml

@@ -62,6 +62,7 @@ stele = { path = "bin/correctness/stele" }
 stringtheory = { path = "lib/stringtheory" }
 ottl = { path = "lib/ottl" }
 otlp-protos = { path = "lib/protos/otlp" }
+anymap3 = { version = "1", default-features = false, features = ["std"] }
 async-trait = { version = "0.1", default-features = false }
 atty = { version = "0.2", default-features = false }
 axum = { version = "0.8", default-features = false }

LICENSE-3rdparty.csv

@@ -8,6 +8,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>"
 argh_derive,https://github.com/google/argh,BSD-3-Clause,"Taylor Cramer <cramertj@google.com>, Benjamin Brittain <bwb@google.com>, Erick Tryzelaar <etryzelaar@google.com>"

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

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

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

@@ -14,6 +14,9 @@ pub fn build_metrics_router() -> Router {
         .route("/metrics/dump", get(handle_metrics_dump))
         .route("/api/v1/series", post(handle_series_v1))
         .route("/api/v2/series", post(handle_series_v2))
+        .route("/api/intake/metrics/v3/series", post(handle_metrics_v3))
+        .route("/api/intake/metrics/v3beta/series", post(handle_metrics_v3))
         .route("/api/beta/sketches", post(handle_sketch_beta))
+        .route("/api/intake/metrics/v3/sketches", post(handle_metrics_v3))
         .with_state(MetricsState::new())
 }

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

@@ -1,6 +1,6 @@
 use std::sync::{Arc, Mutex};
 
-use datadog_protos::metrics::{MetricPayload, SketchPayload};
+use datadog_protos::metrics::{v3::Payload as V3Payload, MetricPayload, SketchPayload};
 use saluki_error::GenericError;
 use stele::Metric;
 
@@ -43,6 +43,16 @@ impl MetricsState {
         Ok(())
     }
 
+    /// Merges the given metrics 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(())
+    }
+
     /// Merges the given series v1 payload into the current metrics state.
     pub fn merge_series_v1_payload(&self, bytes: &[u8]) -> Result<(), GenericError> {
         let metrics = Metric::try_from_series_v1(bytes)?;

bin/correctness/stele/Cargo.toml

@@ -20,4 +20,5 @@ saluki-error = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { workspace = true }
+simdutf8 = { workspace = true }
 stringtheory = { workspace = true }