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, MetricPayload, SketchPayload};
10use serde::{Deserialize, Serialize};
11use stele::{Metric, MetricValue};
12use tracing::warn;
13
14const SELF_TELEMETRY_PREFIX: &str = "datadog.";
15
16#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
17#[serde(rename_all = "snake_case")]
18pub(crate) enum Target {
19    Agent,
20    Adp,
21}
22
23impl Target {
24    #[must_use]
25    pub(crate) fn parse(value: &str) -> Option<Self> {
26        match value {
27            "agent" => Some(Self::Agent),
28            "adp" => Some(Self::Adp),
29            _ => None,
30        }
31    }
32
33    pub(crate) fn as_str(self) -> &'static str {
34        match self {
35            Self::Agent => "agent",
36            Self::Adp => "adp",
37        }
38    }
39}
40
41/// The flushed type of a metric, part of a context's identity.
42#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
43#[serde(rename_all = "snake_case")]
44pub(crate) enum MetricKind {
45    Count,
46    Rate,
47    Gauge,
48    Sketch,
49}
50
51impl MetricKind {
52    fn of(metric: &Metric) -> Option<Self> {
53        metric.values().first().map(|(_, value)| match value {
54            MetricValue::Count { .. } => Self::Count,
55            MetricValue::Rate { .. } => Self::Rate,
56            MetricValue::Gauge { .. } => Self::Gauge,
57            MetricValue::Sketch { .. } => Self::Sketch,
58        })
59    }
60}
61
62#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
63pub(crate) struct EpochSeconds(i64);
64
65impl EpochSeconds {
66    pub(crate) const fn from_epoch_secs(secs: i64) -> Self {
67        Self(secs)
68    }
69
70    /// The intake's current wall-clock time, or `None` if the clock predates
71    /// the epoch or overflows.
72    pub(crate) fn now() -> Option<Self> {
73        let secs = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
74        i64::try_from(secs).ok().map(Self)
75    }
76}
77
78/// A metric context: name, tagset, and type.
79#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
80pub(crate) struct Context {
81    pub(crate) name: String,
82    pub(crate) tagset: BTreeSet<String>,
83    pub(crate) kind: MetricKind,
84}
85
86/// A context and the time it first arrived on its lane.
87#[derive(Clone, Debug, Deserialize, Serialize)]
88pub(crate) struct ContextAt {
89    #[serde(flatten)]
90    pub(crate) context: Context,
91    pub(crate) first_seen: EpochSeconds,
92}
93
94/// One lane's contexts and the intake's current time.
95#[derive(Clone, Debug, Deserialize, Serialize)]
96pub(crate) struct LaneView {
97    pub(crate) now: EpochSeconds,
98    pub(crate) contexts: Vec<ContextAt>,
99}
100
101#[derive(Debug, Default)]
102struct Lanes {
103    seen: BTreeMap<(Target, Context), EpochSeconds>,
104}
105
106impl Lanes {
107    fn record(&mut self, target: Target, contexts: &[Context], now: EpochSeconds) -> usize {
108        let mut added = 0;
109        for context in contexts {
110            if context.name.starts_with(SELF_TELEMETRY_PREFIX) {
111                continue;
112            }
113            if let Entry::Vacant(slot) = self.seen.entry((target, context.clone())) {
114                slot.insert(now);
115                added += 1;
116            }
117        }
118        added
119    }
120
121    fn contexts(&self, target: Target) -> Vec<ContextAt> {
122        self.seen
123            .iter()
124            .filter(|((lane, _), _)| *lane == target)
125            .map(|((_, context), &first_seen)| ContextAt {
126                context: context.clone(),
127                first_seen,
128            })
129            .collect()
130    }
131}
132
133/// Shared handle to the lanes mechanism. Written to by HTTP handlers, read from
134/// by the check programs via control routes.
135#[derive(Clone, Debug, Default)]
136pub struct State {
137    lanes: Arc<Mutex<Lanes>>,
138}
139
140impl State {
141    /// Creates an empty recorder.
142    #[must_use]
143    pub fn new() -> Self {
144        Self::default()
145    }
146
147    pub(crate) fn record_series_v2(&self, target: Target, payload: MetricPayload, now: EpochSeconds) -> usize {
148        let contexts = observe_series(target, payload);
149        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
150    }
151
152    pub(crate) fn record_sketches(&self, target: Target, payload: SketchPayload, now: EpochSeconds) -> usize {
153        let contexts = observe_sketches(target, payload);
154        self.with_lanes(|lanes| lanes.record(target, &contexts, now))
155    }
156
157    pub(crate) fn contexts(&self, target: Target) -> Vec<ContextAt> {
158        self.with_lanes(|lanes| lanes.contexts(target))
159    }
160
161    fn with_lanes<T>(&self, f: impl FnOnce(&mut Lanes) -> T) -> T {
162        f(&mut self.lanes.lock().expect("capture lock poisoned"))
163    }
164}
165
166/// Longest metric name propjoe stores, in bytes (`model.MaxMetricLen`).
167const MAX_METRIC_NAME_LEN: usize = 350;
168/// Most tags propjoe keeps on a series (`model.MaxTagThresh`).
169const MAX_TAG_COUNT: usize = 100;
170/// Most resources propjoe keeps on a series (`model.MaxResourceThresh`).
171const MAX_RESOURCE_COUNT: usize = 500;
172
173/// Whether propjoe's v2 ingest keeps this series. It drops any series with an invalid metric
174/// name (`ValidateMetricName`: empty, over `MaxMetricLen` bytes, or no ASCII-alphabetic byte),
175/// more than `MaxTagThresh` tags, or more than `MaxResourceThresh` resources. Matching keeps
176/// our captured context set equal to what production would store.
177pub(crate) fn series_kept_by_intake(series: &MetricSeries) -> bool {
178    let name = series.metric.as_str();
179    let name_ok =
180        !name.is_empty() && name.len() <= MAX_METRIC_NAME_LEN && name.bytes().any(|b| b.is_ascii_alphabetic());
181    name_ok && series.tags.len() <= MAX_TAG_COUNT && series.resources.len() <= MAX_RESOURCE_COUNT
182}
183
184/// Decodes a `/api/v2/series` payload into contexts with stele's `Metric::try_from_series_v2`.
185fn observe_series(target: Target, payload: MetricPayload) -> Vec<Context> {
186    let mut contexts = Vec::new();
187    for series in payload.series {
188        if !series_kept_by_intake(&series) {
189            continue;
190        }
191        let mut single = MetricPayload::new();
192        single.series.push(series);
193        match Metric::try_from_series_v2(single) {
194            Ok(metrics) => contexts.extend(metrics.iter().filter_map(context_of)),
195            Err(error) => {
196                warn!(target = target.as_str(), %error, "skipped a series that did not convert to a stele metric");
197            }
198        }
199    }
200    contexts
201}
202
203/// Decodes a sketch payload into contexts.
204fn observe_sketches(target: Target, payload: SketchPayload) -> Vec<Context> {
205    let mut contexts = Vec::new();
206    for sketch in payload.sketches {
207        let mut single = SketchPayload::new();
208        single.sketches.push(sketch);
209        match Metric::try_from_sketch(single) {
210            Ok(metrics) => contexts.extend(metrics.iter().filter_map(context_of)),
211            Err(error) => {
212                warn!(target = target.as_str(), %error, "skipped a sketch that did not convert to a stele metric");
213            }
214        }
215    }
216    contexts
217}
218
219fn context_of(metric: &Metric) -> Option<Context> {
220    let kind = MetricKind::of(metric)?;
221    Some(Context {
222        name: metric.context().name().to_string(),
223        tagset: metric.context().tags().iter().cloned().collect(),
224        kind,
225    })
226}
227
228#[cfg(test)]
229mod tests;
antithesis_intake/capture.rs

antithesis_intake/
capture.rs
