antithesis_intake/
capture.rs

1//! Differential metric context capture.
2//!
3//! See scenario README for details.
4
5use std::collections::{btree_map::Entry, BTreeMap, BTreeSet};
6use std::sync::{Arc, Mutex};
7use std::time::{SystemTime, UNIX_EPOCH};
8
9use datadog_protos::metrics::metric_payload::{MetricSeries, MetricType, Resource};
10use datadog_protos::metrics::{MetricPayload, SketchPayload};
11use serde::{Deserialize, Serialize};
12
13use crate::lenient_decode::V3Series;
14
15const SELF_TELEMETRY_PREFIX: &str = "datadog.";
16
17#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
18#[serde(rename_all = "snake_case")]
19pub(crate) enum Target {
20    Agent,
21    Adp,
22}
23
24impl Target {
25    #[must_use]
26    pub(crate) fn parse(value: &str) -> Option<Self> {
27        match value {
28            "agent" => Some(Self::Agent),
29            "adp" => Some(Self::Adp),
30            _ => None,
31        }
32    }
33
34    pub(crate) fn as_str(self) -> &'static str {
35        match self {
36            Self::Agent => "agent",
37            Self::Adp => "adp",
38        }
39    }
40}
41
42/// The flushed type of a metric, part of a context's identity.
43#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
44#[serde(rename_all = "snake_case")]
45pub(crate) enum MetricKind {
46    Count,
47    Rate,
48    Gauge,
49    Sketch,
50    /// A metric type outside the known set — an out-of-range v3 type nibble. Production keeps such a
51    /// series and forwards its type verbatim, so the intake keeps it too rather than dropping it and
52    /// masking a producer bug.
53    Other,
54}
55
56impl MetricKind {
57    /// Derives the kind from the v2 wire type field. The accessor defaults any out-of-range type to
58    /// `UNSPECIFIED`, which maps to `Other`, keeping the series and forwarding an unknown type rather
59    /// than dropping it and masking a producer bug, as the v3 path does for an unknown type nibble.
60    fn of(type_: MetricType) -> Self {
61        match type_ {
62            MetricType::COUNT => Self::Count,
63            MetricType::RATE => Self::Rate,
64            MetricType::GAUGE => Self::Gauge,
65            MetricType::UNSPECIFIED => Self::Other,
66        }
67    }
68}
69
70#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
71pub(crate) struct EpochSeconds(i64);
72
73impl EpochSeconds {
74    pub(crate) const fn from_epoch_secs(secs: i64) -> Self {
75        Self(secs)
76    }
77
78    /// The whole seconds since the Unix epoch.
79    pub(crate) fn secs(self) -> i64 {
80        self.0
81    }
82
83    /// The intake's current wall-clock time, or `None` if the clock predates
84    /// the epoch or overflows.
85    pub(crate) fn now() -> Option<Self> {
86        let secs = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
87        i64::try_from(secs).ok().map(Self)
88    }
89}
90
91/// One point value as the native decoder reads it off the wire, kind-agnostic. The intake keeps no
92/// curve, only whether a series carries a point that survives the backend's per-point drops, so a
93/// series left with none emits no context.
94#[derive(Clone, Debug, PartialEq)]
95pub(crate) enum BucketValue {
96    /// A count, rate, or gauge scalar.
97    Scalar(f64),
98    /// A `DDSketch` point: the summary the Agent emits plus its log-grid bins.
99    Sketch(SketchValue),
100}
101
102/// A `DDSketch` point: the summary the Agent emits plus the log-grid bins as `(key, count)`, key-sorted.
103#[derive(Clone, Debug, PartialEq)]
104pub(crate) struct SketchValue {
105    pub(crate) count: i64,
106    pub(crate) sum: f64,
107    pub(crate) min: f64,
108    pub(crate) max: f64,
109    pub(crate) bins: Vec<(i32, u32)>,
110}
111
112/// A metric context: name, tagset, and type.
113#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
114pub(crate) struct Context {
115    pub(crate) name: String,
116    pub(crate) tagset: BTreeSet<String>,
117    pub(crate) kind: MetricKind,
118}
119
120/// A context and the time it first arrived on its lane.
121#[derive(Clone, Debug, Deserialize, Serialize)]
122pub(crate) struct ContextAt {
123    #[serde(flatten)]
124    pub(crate) context: Context,
125    pub(crate) first_seen: EpochSeconds,
126}
127
128/// One lane's contexts and the intake's current time.
129#[derive(Clone, Debug, Deserialize, Serialize)]
130pub(crate) struct LaneView {
131    pub(crate) now: EpochSeconds,
132    pub(crate) contexts: Vec<ContextAt>,
133}
134
135#[derive(Debug, Default)]
136struct Lanes {
137    seen: BTreeMap<(Target, Context), EpochSeconds>,
138}
139
140impl Lanes {
141    fn record(&mut self, target: Target, contexts: &[Context], now: EpochSeconds) -> usize {
142        let mut added = 0;
143        for context in contexts {
144            if context.name.starts_with(SELF_TELEMETRY_PREFIX) {
145                continue;
146            }
147            if let Entry::Vacant(slot) = self.seen.entry((target, context.clone())) {
148                slot.insert(now);
149                added += 1;
150            }
151        }
152        added
153    }
154
155    fn contexts(&self, target: Target) -> Vec<ContextAt> {
156        self.seen
157            .iter()
158            .filter(|((lane, _), _)| *lane == target)
159            .map(|((_, context), &first_seen)| ContextAt {
160                context: context.clone(),
161                first_seen,
162            })
163            .collect()
164    }
165}
166
167/// Shared handle to the lanes mechanism. Written to by HTTP handlers, read from
168/// by the check programs via control routes.
169#[derive(Clone, Debug, Default)]
170pub struct State {
171    lanes: Arc<Mutex<Lanes>>,
172}
173
174impl State {
175    /// Creates an empty recorder.
176    #[must_use]
177    pub fn new() -> Self {
178        Self::default()
179    }
180
181    pub(crate) fn record_series_v2(&self, target: Target, payload: MetricPayload, now: EpochSeconds) -> usize {
182        let contexts = observe_series(payload, now.secs());
183        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
184    }
185
186    pub(crate) fn record_sketches(&self, target: Target, payload: SketchPayload, now: EpochSeconds) -> usize {
187        let contexts = observe_sketches(payload);
188        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
189    }
190
191    pub(crate) fn record_series_v3(&self, target: Target, series: Vec<V3Series>, now: EpochSeconds) -> usize {
192        let contexts = observe_series_v3(series, now.secs());
193        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
194    }
195
196    pub(crate) fn contexts(&self, target: Target) -> Vec<ContextAt> {
197        self.with_lanes(|lanes| lanes.contexts(target))
198    }
199
200    fn with_lanes<T>(&self, f: impl FnOnce(&mut Lanes) -> T) -> T {
201        f(&mut self.lanes.lock().expect("capture lock poisoned"))
202    }
203}
204
205/// Longest metric name the intake keeps, in bytes.
206pub(crate) const MAX_METRIC_NAME_LEN: usize = 350;
207/// Most tags the intake keeps on a series. The backend's tag limit is per-org (`tagLimitProvider`),
208/// defaulting to `model.MaxTagThresh`=100; the rig hardcodes the default, so an org with a non-default
209/// limit would diverge. This is a knowingly-deferred config-parity approximation, sound while the
210/// differential only exercises default-org limits.
211pub(crate) const MAX_TAG_COUNT: usize = 100;
212/// Most resources the intake keeps on a series. Per-org in the backend (`resourceLimitProvider`),
213/// defaulting to `model.MaxResourceThresh`=500; the rig hardcodes the default, same deferral as
214/// `MAX_TAG_COUNT`.
215pub(crate) const MAX_RESOURCE_COUNT: usize = 500;
216/// Longest `host` resource name the intake keeps on a series, in bytes.
217pub(crate) const MAX_HOST_NAME_LEN: usize = 255;
218/// How far past the intake's receipt clock a scalar point may sit before it is dropped, in seconds.
219/// Matches the backend's `payload.MaxSecondsInFuture` (intake/payload/normalizer.go:32, ten minutes).
220const MAX_SECONDS_IN_FUTURE: i64 = 600;
221
222/// Whether a scalar point is kept, mirroring the backend's per-point drops (v2
223/// api_series_v2_handler_helpers.go:264-275, v3 validatePoint api_series_v3_handler.go:549-557): a NaN
224/// value is dropped and a timestamp more than `MAX_SECONDS_IN_FUTURE` past the receipt clock is dropped.
225/// Past timestamps are kept, since late points are accepted downstream. Sketch points carry no scalar
226/// value and are not filtered here, matching the scalar-only scope of the backend's point checks.
227fn scalar_point_kept(value: &BucketValue, bucket_start: u64, now_secs: i64) -> bool {
228    match value {
229        BucketValue::Scalar(v) => {
230            !v.is_nan() && i128::from(bucket_start) <= i128::from(now_secs) + i128::from(MAX_SECONDS_IN_FUTURE)
231        }
232        BucketValue::Sketch(_) => true,
233    }
234}
235
236/// Whether the intake keeps this metric name: non-empty, at most the max name length in bytes, and
237/// carrying at least one ASCII-alphabetic byte. Shared by the v2 and v3 drop rules.
238pub(crate) fn metric_name_kept(name: &str) -> bool {
239    !name.is_empty() && name.len() <= MAX_METRIC_NAME_LEN && name.bytes().any(|b| b.is_ascii_alphabetic())
240}
241
242/// Whether the intake's v2 ingest keeps this series. It drops any series with an invalid metric name
243/// (empty, over the max name length, or no ASCII-alphabetic byte), more than the max tag count, more
244/// than the max resource count, or a `host` resource whose name exceeds the max host length. Matching
245/// keeps our captured context set equal to what production would store, and keeps the two lanes' drop
246/// rules identical to the v3 path.
247pub(crate) fn series_kept_by_intake(series: &MetricSeries) -> bool {
248    let host_ok = series
249        .resources
250        .iter()
251        .find(|r| r.type_() == "host")
252        .is_none_or(|host| host.name().len() <= MAX_HOST_NAME_LEN);
253    metric_name_kept(series.metric.as_str())
254        && series.tags.len() <= MAX_TAG_COUNT
255        && series.resources.len() <= MAX_RESOURCE_COUNT
256        && host_ok
257}
258
259/// The tagset a series carries, its wire tags plus its `host` resource folded into a `host:<name>` tag,
260/// matching the fold the v3 lane applies.
261fn tagset_with_host(tags: &[String], host: Option<&str>) -> BTreeSet<String> {
262    let mut tagset: BTreeSet<String> = tags.iter().cloned().collect();
263    if let Some(host) = host {
264        if !host.is_empty() {
265            tagset.insert(format!("host:{host}"));
266        }
267    }
268    tagset
269}
270
271/// Reads a `/api/v2/series` `MetricPayload` straight off the wire into contexts, no stele in the path.
272/// It applies the same `series_kept_by_intake` drop rules and the same `host` resource fold as the v3
273/// lane, and derives the kind from the wire type field. A series whose every point is dropped by the
274/// backend's per-point NaN and too-far-future checks (keyed on `now_secs`, the intake's receipt clock)
275/// emits no context, matching the backend's all-points-dropped series drop.
276fn observe_series(payload: MetricPayload, now_secs: i64) -> Vec<Context> {
277    let mut contexts = Vec::new();
278    for series in payload.series {
279        if !series_kept_by_intake(&series) {
280            continue;
281        }
282        let host = series
283            .resources
284            .iter()
285            .find(|r| r.type_() == "host")
286            .map(Resource::name);
287        let has_point = series.points.iter().any(|point| {
288            u64::try_from(point.timestamp)
289                .is_ok_and(|ts| scalar_point_kept(&BucketValue::Scalar(point.value), ts, now_secs))
290        });
291        if !has_point {
292            continue;
293        }
294        contexts.push(Context {
295            name: series.metric.clone(),
296            tagset: tagset_with_host(&series.tags, host),
297            kind: MetricKind::of(series.type_()),
298        });
299    }
300    contexts
301}
302
303/// Reads an `/api/beta/sketches` `SketchPayload` straight off the wire into contexts, no stele in the
304/// path. Each kept sketch is one `Sketch`-kind context; its `host` folds into a `host:<name>` tag as
305/// the v2 series path folds its host resource. A sketch left with no point whose timestamp fits a
306/// `u64` bucket-start emits no context.
307///
308/// The backend's `NormalizeDistributionReq` (intake/payload/normalizer.go:459-503) drops a distribution
309/// whose host exceeds the host-length cap, whose tag count exceeds the tag cap, or whose metric name is
310/// invalid, keeping the rest. This applies the same per-sketch keep predicate. The backend additionally
311/// REWRITES kept metric names (`NormMetricNameParse`) and tags (`NormalizeTags`); that normalization is
312/// a separate fidelity gap the sketch, v2, and v3 lanes all share and is not modeled here.
313fn observe_sketches(payload: SketchPayload) -> Vec<Context> {
314    let mut contexts = Vec::new();
315    for sketch in payload.sketches {
316        // Per-distribution keep rules, matching the backend. Resource count has no sketch analogue.
317        if !metric_name_kept(sketch.metric())
318            || sketch.tags.len() > MAX_TAG_COUNT
319            || sketch.host().len() > MAX_HOST_NAME_LEN
320        {
321            continue;
322        }
323        let has_point = sketch.dogsketches.iter().any(|d| u64::try_from(d.ts).is_ok())
324            || sketch.distributions.iter().any(|d| u64::try_from(d.ts).is_ok());
325        if !has_point {
326            continue;
327        }
328        contexts.push(Context {
329            name: sketch.metric.clone(),
330            tagset: tagset_with_host(&sketch.tags, Some(sketch.host())),
331            kind: MetricKind::Sketch,
332        });
333    }
334    contexts
335}
336
337/// Maps the natively decoded v3 series into contexts. The native decoder in `lenient_decode` already
338/// applied the two-tier failure model, the production intake's per-series validation, and the `host`
339/// resource fold; this applies the backend's per-point NaN and too-far-future drops (validatePoint),
340/// keyed on `now_secs`, and emits a context for each series left with at least one surviving point.
341fn observe_series_v3(series: Vec<V3Series>, now_secs: i64) -> Vec<Context> {
342    series
343        .into_iter()
344        .filter(|s| {
345            s.points
346                .iter()
347                .any(|(ts, value)| scalar_point_kept(value, *ts, now_secs))
348        })
349        .map(|s| Context {
350            name: s.name,
351            tagset: s.tags.into_iter().collect(),
352            kind: s.kind,
353        })
354        .collect()
355}
356
357#[cfg(test)]
358mod tests;
antithesis_intake/capture.rs

antithesis_intake/
capture.rs
