chore(antithesis): Intro Intake V3

This commit introduces a v3 intake into antithesis harness to support
symmetric difference with ADP-on -- v2 by default -- and ADP-off -- v3
by default.

Author: blt

Cargo.lock

@@ -19,13 +19,14 @@ anyhow = { workspace = true, features = ["std"] }
 axum = { workspace = true, features = ["http1", "json", "tokio", "tracing"] }
 clap = { workspace = true, features = ["derive", "env", "error-context", "help", "std", "usage"] }
 datadog-protos = { workspace = true }
+harness = { path = "../harness" }
 headers = { workspace = true }
 mime = { workspace = true }
 protobuf = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-stele = { workspace = true }
 tokio = { workspace = true, features = [
+  "io-util",
   "macros",
   "net",
   "rt",

test/antithesis/intake/Cargo.toml

@@ -6,10 +6,11 @@ use std::collections::{btree_map::Entry, BTreeMap, BTreeSet};
 use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use datadog_protos::metrics::{metric_payload::MetricSeries, MetricPayload, SketchPayload};
+use datadog_protos::metrics::metric_payload::{MetricSeries, MetricType, Resource};
+use datadog_protos::metrics::{MetricPayload, SketchPayload};
 use serde::{Deserialize, Serialize};
-use stele::{Metric, MetricValue};
-use tracing::warn;
+
+use crate::lenient_decode::V3Series;
 
 const SELF_TELEMETRY_PREFIX: &str = "datadog.";
 
@@ -46,16 +47,23 @@ pub(crate) enum MetricKind {
     Rate,
     Gauge,
     Sketch,
+    /// A metric type outside the known set — an out-of-range v3 type nibble. Production keeps such a
+    /// series and forwards its type verbatim, so the intake keeps it too rather than dropping it and
+    /// masking a producer bug.
+    Other,
 }
 
 impl MetricKind {
-    fn of(metric: &Metric) -> Option<Self> {
-        metric.values().first().map(|(_, value)| match value {
-            MetricValue::Count { .. } => Self::Count,
-            MetricValue::Rate { .. } => Self::Rate,
-            MetricValue::Gauge { .. } => Self::Gauge,
-            MetricValue::Sketch { .. } => Self::Sketch,
-        })
+    /// Derives the kind from the v2 wire type field. The accessor defaults any out-of-range type to
+    /// `UNSPECIFIED`, which maps to `Other`, keeping the series and forwarding an unknown type rather
+    /// than dropping it and masking a producer bug, as the v3 path does for an unknown type nibble.
+    fn of(type_: MetricType) -> Self {
+        match type_ {
+            MetricType::COUNT => Self::Count,
+            MetricType::RATE => Self::Rate,
+            MetricType::GAUGE => Self::Gauge,
+            MetricType::UNSPECIFIED => Self::Other,
+        }
     }
 }
 
@@ -67,6 +75,11 @@ impl EpochSeconds {
         Self(secs)
     }
 
+    /// The whole seconds since the Unix epoch.
+    pub(crate) fn secs(self) -> i64 {
+        self.0
+    }
+
     /// The intake's current wall-clock time, or `None` if the clock predates
     /// the epoch or overflows.
     pub(crate) fn now() -> Option<Self> {
@@ -75,6 +88,27 @@ impl EpochSeconds {
     }
 }
 
+/// One point value as the native decoder reads it off the wire, kind-agnostic. The intake keeps no
+/// curve, only whether a series carries a point that survives the backend's per-point drops, so a
+/// series left with none emits no context.
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) enum BucketValue {
+    /// A count, rate, or gauge scalar.
+    Scalar(f64),
+    /// A `DDSketch` point: the summary the Agent emits plus its log-grid bins.
+    Sketch(SketchValue),
+}
+
+/// A `DDSketch` point: the summary the Agent emits plus the log-grid bins as `(key, count)`, key-sorted.
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) struct SketchValue {
+    pub(crate) count: i64,
+    pub(crate) sum: f64,
+    pub(crate) min: f64,
+    pub(crate) max: f64,
+    pub(crate) bins: Vec<(i32, u32)>,
+}
+
 /// A metric context: name, tagset, and type.
 #[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
 pub(crate) struct Context {
@@ -145,12 +179,17 @@ impl State {
     }
 
     pub(crate) fn record_series_v2(&self, target: Target, payload: MetricPayload, now: EpochSeconds) -> usize {
-        let contexts = observe_series(target, payload);
+        let contexts = observe_series(payload, now.secs());
         self.with_lanes(|lanes| lanes.record(target, &contexts, now))
     }
 
     pub(crate) fn record_sketches(&self, target: Target, payload: SketchPayload, now: EpochSeconds) -> usize {
-        let contexts = observe_sketches(target, payload);
+        let contexts = observe_sketches(payload);
+        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
+    }
+
+    pub(crate) fn record_series_v3(&self, target: Target, series: Vec<V3Series>, now: EpochSeconds) -> usize {
+        let contexts = observe_series_v3(series, now.secs());
         self.with_lanes(|lanes| lanes.record(target, &contexts, now))
     }
 
@@ -163,66 +202,156 @@ impl State {
     }
 }
 
-/// Longest metric name propjoe stores, in bytes (`model.MaxMetricLen`).
-const MAX_METRIC_NAME_LEN: usize = 350;
-/// Most tags propjoe keeps on a series (`model.MaxTagThresh`).
-const MAX_TAG_COUNT: usize = 100;
-/// Most resources propjoe keeps on a series (`model.MaxResourceThresh`).
-const MAX_RESOURCE_COUNT: usize = 500;
+/// Longest metric name the intake keeps, in bytes.
+pub(crate) const MAX_METRIC_NAME_LEN: usize = 350;
+/// Most tags the intake keeps on a series. The backend's tag limit is per-org (`tagLimitProvider`),
+/// defaulting to `model.MaxTagThresh`=100; the rig hardcodes the default, so an org with a non-default
+/// limit would diverge. This is a knowingly-deferred config-parity approximation, sound while the
+/// differential only exercises default-org limits.
+pub(crate) const MAX_TAG_COUNT: usize = 100;
+/// Most resources the intake keeps on a series. Per-org in the backend (`resourceLimitProvider`),
+/// defaulting to `model.MaxResourceThresh`=500; the rig hardcodes the default, same deferral as
+/// `MAX_TAG_COUNT`.
+pub(crate) const MAX_RESOURCE_COUNT: usize = 500;
+/// Longest `host` resource name the intake keeps on a series, in bytes.
+pub(crate) const MAX_HOST_NAME_LEN: usize = 255;
+/// How far past the intake's receipt clock a scalar point may sit before it is dropped, in seconds.
+/// Matches the backend's `payload.MaxSecondsInFuture` (intake/payload/normalizer.go:32, ten minutes).
+const MAX_SECONDS_IN_FUTURE: i64 = 600;
+
+/// Whether a scalar point is kept, mirroring the backend's per-point drops (v2
+/// api_series_v2_handler_helpers.go:264-275, v3 validatePoint api_series_v3_handler.go:549-557): a NaN
+/// value is dropped and a timestamp more than `MAX_SECONDS_IN_FUTURE` past the receipt clock is dropped.
+/// Past timestamps are kept, since late points are accepted downstream. Sketch points carry no scalar
+/// value and are not filtered here, matching the scalar-only scope of the backend's point checks.
+fn scalar_point_kept(value: &BucketValue, bucket_start: u64, now_secs: i64) -> bool {
+    match value {
+        BucketValue::Scalar(v) => {
+            !v.is_nan() && i128::from(bucket_start) <= i128::from(now_secs) + i128::from(MAX_SECONDS_IN_FUTURE)
+        }
+        BucketValue::Sketch(_) => true,
+    }
+}
 
-/// Whether propjoe's v2 ingest keeps this series. It drops any series with an invalid metric
-/// name (`ValidateMetricName`: empty, over `MaxMetricLen` bytes, or no ASCII-alphabetic byte),
-/// more than `MaxTagThresh` tags, or more than `MaxResourceThresh` resources. Matching keeps
-/// our captured context set equal to what production would store.
+/// Whether the intake keeps this metric name: non-empty, at most the max name length in bytes, and
+/// carrying at least one ASCII-alphabetic byte. Shared by the v2 and v3 drop rules.
+pub(crate) fn metric_name_kept(name: &str) -> bool {
+    !name.is_empty() && name.len() <= MAX_METRIC_NAME_LEN && name.bytes().any(|b| b.is_ascii_alphabetic())
+}
+
+/// Whether the intake's v2 ingest keeps this series. It drops any series with an invalid metric name
+/// (empty, over the max name length, or no ASCII-alphabetic byte), more than the max tag count, more
+/// than the max resource count, or a `host` resource whose name exceeds the max host length. Matching
+/// keeps our captured context set equal to what production would store, and keeps the two lanes' drop
+/// rules identical to the v3 path.
 pub(crate) fn series_kept_by_intake(series: &MetricSeries) -> bool {
-    let name = series.metric.as_str();
-    let name_ok =
-        !name.is_empty() && name.len() <= MAX_METRIC_NAME_LEN && name.bytes().any(|b| b.is_ascii_alphabetic());
-    name_ok && series.tags.len() <= MAX_TAG_COUNT && series.resources.len() <= MAX_RESOURCE_COUNT
+    let host_ok = series
+        .resources
+        .iter()
+        .find(|r| r.type_() == "host")
+        .is_none_or(|host| host.name().len() <= MAX_HOST_NAME_LEN);
+    metric_name_kept(series.metric.as_str())
+        && series.tags.len() <= MAX_TAG_COUNT
+        && series.resources.len() <= MAX_RESOURCE_COUNT
+        && host_ok
+}
+
+/// The tagset a series carries, its wire tags plus its `host` resource folded into a `host:<name>` tag,
+/// matching the fold the v3 lane applies.
+fn tagset_with_host(tags: &[String], host: Option<&str>) -> BTreeSet<String> {
+    let mut tagset: BTreeSet<String> = tags.iter().cloned().collect();
+    if let Some(host) = host {
+        if !host.is_empty() {
+            tagset.insert(format!("host:{host}"));
+        }
+    }
+    tagset
 }
 
-/// Decodes a `/api/v2/series` payload into contexts with stele's `Metric::try_from_series_v2`.
-fn observe_series(target: Target, payload: MetricPayload) -> Vec<Context> {
+/// Reads a `/api/v2/series` `MetricPayload` straight off the wire into contexts, no stele in the path.
+/// It applies the same `series_kept_by_intake` drop rules and the same `host` resource fold as the v3
+/// lane, and derives the kind from the wire type field. A series whose every point is dropped by the
+/// backend's per-point NaN and too-far-future checks (keyed on `now_secs`, the intake's receipt clock)
+/// emits no context, matching the backend's all-points-dropped series drop.
+fn observe_series(payload: MetricPayload, now_secs: i64) -> Vec<Context> {
     let mut contexts = Vec::new();
     for series in payload.series {
         if !series_kept_by_intake(&series) {
             continue;
         }
-        let mut single = MetricPayload::new();
-        single.series.push(series);
-        match Metric::try_from_series_v2(single) {
-            Ok(metrics) => contexts.extend(metrics.iter().filter_map(context_of)),
-            Err(error) => {
-                warn!(target = target.as_str(), %error, "skipped a series that did not convert to a stele metric");
-            }
+        let host = series
+            .resources
+            .iter()
+            .find(|r| r.type_() == "host")
+            .map(Resource::name);
+        let has_point = series.points.iter().any(|point| {
+            u64::try_from(point.timestamp)
+                .is_ok_and(|ts| scalar_point_kept(&BucketValue::Scalar(point.value), ts, now_secs))
+        });
+        if !has_point {
+            continue;
         }
+        contexts.push(Context {
+            name: series.metric.clone(),
+            tagset: tagset_with_host(&series.tags, host),
+            kind: MetricKind::of(series.type_()),
+        });
     }
     contexts
 }
 
-/// Decodes a sketch payload into contexts.
-fn observe_sketches(target: Target, payload: SketchPayload) -> Vec<Context> {
+/// Reads an `/api/beta/sketches` `SketchPayload` straight off the wire into contexts, no stele in the
+/// path. Each kept sketch is one `Sketch`-kind context; its `host` folds into a `host:<name>` tag as
+/// the v2 series path folds its host resource. A sketch left with no point whose timestamp fits a
+/// `u64` bucket-start emits no context.
+///
+/// The backend's `NormalizeDistributionReq` (intake/payload/normalizer.go:459-503) drops a distribution
+/// whose host exceeds the host-length cap, whose tag count exceeds the tag cap, or whose metric name is
+/// invalid, keeping the rest. This applies the same per-sketch keep predicate. The backend additionally
+/// REWRITES kept metric names (`NormMetricNameParse`) and tags (`NormalizeTags`); that normalization is
+/// a separate fidelity gap the sketch, v2, and v3 lanes all share and is not modeled here.
+fn observe_sketches(payload: SketchPayload) -> Vec<Context> {
     let mut contexts = Vec::new();
     for sketch in payload.sketches {
-        let mut single = SketchPayload::new();
-        single.sketches.push(sketch);
-        match Metric::try_from_sketch(single) {
-            Ok(metrics) => contexts.extend(metrics.iter().filter_map(context_of)),
-            Err(error) => {
-                warn!(target = target.as_str(), %error, "skipped a sketch that did not convert to a stele metric");
-            }
+        // Per-distribution keep rules, matching the backend. Resource count has no sketch analogue.
+        if !metric_name_kept(sketch.metric())
+            || sketch.tags.len() > MAX_TAG_COUNT
+            || sketch.host().len() > MAX_HOST_NAME_LEN
+        {
+            continue;
         }
+        let has_point = sketch.dogsketches.iter().any(|d| u64::try_from(d.ts).is_ok())
+            || sketch.distributions.iter().any(|d| u64::try_from(d.ts).is_ok());
+        if !has_point {
+            continue;
+        }
+        contexts.push(Context {
+            name: sketch.metric.clone(),
+            tagset: tagset_with_host(&sketch.tags, Some(sketch.host())),
+            kind: MetricKind::Sketch,
+        });
     }
     contexts
 }
 
-fn context_of(metric: &Metric) -> Option<Context> {
-    let kind = MetricKind::of(metric)?;
-    Some(Context {
-        name: metric.context().name().to_string(),
-        tagset: metric.context().tags().iter().cloned().collect(),
-        kind,
-    })
+/// Maps the natively decoded v3 series into contexts. The native decoder in `lenient_decode` already
+/// applied the two-tier failure model, the production intake's per-series validation, and the `host`
+/// resource fold; this applies the backend's per-point NaN and too-far-future drops (validatePoint),
+/// keyed on `now_secs`, and emits a context for each series left with at least one surviving point.
+fn observe_series_v3(series: Vec<V3Series>, now_secs: i64) -> Vec<Context> {
+    series
+        .into_iter()
+        .filter(|s| {
+            s.points
+                .iter()
+                .any(|(ts, value)| scalar_point_kept(value, *ts, now_secs))
+        })
+        .map(|s| Context {
+            name: s.name,
+            tagset: s.tags.into_iter().collect(),
+            kind: s.kind,
+        })
+        .collect()
 }
 
 #[cfg(test)]