antithesis_scenario_differential/
contexts.rs

1//! Context comparison shared by the `eventually_` and `finally_` checks.
2//!
3//! Each check fetches the two lanes from the intake and builds a [`Difference`], the symmetric
4//! difference `D` of their context sets. `eventually_` fails an overdue member. `finally_` fails any
5//! residual after a drain.
6
7use std::collections::{BTreeMap, BTreeSet};
8use std::time::Duration;
9
10use reqwest::blocking::Client;
11use serde::{Deserialize, Serialize};
12
13/// A metric context: name, tagset, and flushed type. The whole triple is the identity; `tagset` is a
14/// set, so tag order never decides equality.
15#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
16pub struct Context {
17    name: String,
18    tagset: BTreeSet<String>,
19    kind: String,
20}
21
22/// A context and the time it first arrived on a lane, the flat wire shape the intake serves.
23#[derive(Clone, Debug, Deserialize)]
24struct Captured {
25    #[serde(flatten)]
26    context: Context,
27    first_seen: i64,
28}
29
30/// A lane's cumulative context set with each context's first-seen time. The README's `C_ADP`/`C_DA`.
31type Cumulative = BTreeMap<Context, i64>;
32
33/// The lane to read from the intake control API.
34#[derive(Clone, Copy, Debug, Serialize)]
35#[serde(rename_all = "snake_case")]
36pub enum Lane {
37    Agent,
38    Adp,
39}
40
41impl Lane {
42    fn as_str(self) -> &'static str {
43        match self {
44            Lane::Agent => "agent",
45            Lane::Adp => "adp",
46        }
47    }
48}
49
50/// One lane's contexts and the intake's current time. `now` ages `first_seen` against the same clock
51/// that stamped it.
52#[derive(Clone, Debug, Deserialize)]
53pub struct LaneView {
54    now: i64,
55    contexts: Vec<Captured>,
56}
57
58impl LaneView {
59    /// Fetches a lane's contexts and the intake's current time from the control API.
60    ///
61    /// # Errors
62    ///
63    /// Errors when the request fails, the intake returns an error status, or the body does not parse.
64    pub fn fetch(client: &Client, intake_addr: &str, lane: Lane) -> anyhow::Result<Self> {
65        let url = format!("http://{intake_addr}/antithesis/metrics/{}", lane.as_str());
66        client
67            .get(&url)
68            .send()
69            .map_err(|e| anyhow::anyhow!("GET {url}: {e}"))?
70            .error_for_status()
71            .map_err(|e| anyhow::anyhow!("GET {url} returned an error status: {e}"))?
72            .json()
73            .map_err(|e| anyhow::anyhow!("parse JSON response from {url}: {e}"))
74    }
75
76    fn cumulative(&self) -> Cumulative {
77        self.contexts
78            .iter()
79            .map(|c| (c.context.clone(), c.first_seen))
80            .collect()
81    }
82}
83
84/// A diverging context's lane and first-seen time, before it is aged against `now`.
85#[derive(Clone, Copy, Debug)]
86struct Member {
87    lane: Lane,
88    first_seen: i64,
89}
90
91/// A diverging context tagged with the lane that carries it and how long it has sat in `D`.
92///
93/// `agent` means ADP dropped or mangled a context the Datadog Agent emitted; `adp` means ADP emitted
94/// one the Agent did not. The Agent is normative, so either direction is an ADP defect.
95#[derive(Clone, Debug, Serialize)]
96pub struct Diverging {
97    pub lane: Lane,
98    #[serde(flatten)]
99    pub context: Context,
100    pub age_secs: i64,
101}
102
103/// The symmetric difference `D` of `C_ADP` and `C_DA`: contexts on one lane but not both, aged
104/// against the intake's clock.
105pub struct Difference {
106    members: BTreeMap<Context, Member>,
107    now: i64,
108}
109
110impl Difference {
111    /// Computes `D` from the two lanes' cumulative sets.
112    #[must_use]
113    pub fn between(agent: &LaneView, adp: &LaneView) -> Self {
114        let c_da = agent.cumulative();
115        let c_adp = adp.cumulative();
116        let mut members = BTreeMap::new();
117        for (context, &first_seen) in &c_da {
118            if !c_adp.contains_key(context) {
119                members.insert(
120                    context.clone(),
121                    Member {
122                        lane: Lane::Agent,
123                        first_seen,
124                    },
125                );
126            }
127        }
128        for (context, &first_seen) in &c_adp {
129            if !c_da.contains_key(context) {
130                members.insert(
131                    context.clone(),
132                    Member {
133                        lane: Lane::Adp,
134                        first_seen,
135                    },
136                );
137            }
138        }
139        Self {
140            members,
141            now: agent.now.max(adp.now),
142        }
143    }
144
145    /// The number of diverging contexts.
146    #[must_use]
147    pub fn len(&self) -> usize {
148        self.members.len()
149    }
150
151    /// Whether the lanes agree.
152    #[must_use]
153    pub fn is_empty(&self) -> bool {
154        self.members.is_empty()
155    }
156
157    /// How many members are `delayed`: sat in `D` longer than `budget` by the intake's clock.
158    #[must_use]
159    pub fn delayed(&self, budget: Duration) -> usize {
160        let budget = budget.as_secs() as i64;
161        self.members
162            .values()
163            .filter(|m| self.now - m.first_seen > budget)
164            .count()
165    }
166
167    /// The diverging contexts, each tagged with its lane and its age in `D` by the intake's clock.
168    #[must_use]
169    pub fn diverging(&self) -> Vec<Diverging> {
170        self.members
171            .iter()
172            .map(|(context, member)| Diverging {
173                lane: member.lane,
174                context: context.clone(),
175                age_secs: self.now - member.first_seen,
176            })
177            .collect()
178    }
179}
180
181#[cfg(test)]
182mod tests {
183    use serde_json::json;
184
185    use super::*;
186
187    fn captured(name: &str, kind: &str, tags: &[&str], first_seen: i64) -> Captured {
188        Captured {
189            context: Context {
190                name: name.to_string(),
191                tagset: tags.iter().map(|t| (*t).to_string()).collect(),
192                kind: kind.to_string(),
193            },
194            first_seen,
195        }
196    }
197
198    fn lane(now: i64, contexts: Vec<Captured>) -> LaneView {
199        LaneView { now, contexts }
200    }
201
202    #[test]
203    fn flushed_type_is_part_of_identity() {
204        // Same name and tags but different flushed type are distinct contexts.
205        let agent = lane(10, vec![captured("adp.requests", "count", &["host:h"], 10)]);
206        let adp = lane(10, vec![captured("adp.requests", "gauge", &["host:h"], 10)]);
207
208        assert_eq!(Difference::between(&agent, &adp).len(), 2);
209    }
210
211    #[test]
212    fn tag_order_does_not_decide_identity() {
213        // The same tag set in any order is one context, so the lanes agree.
214        let agent = lane(10, vec![captured("adp.requests", "count", &["b:2", "a:1"], 10)]);
215        let adp = lane(11, vec![captured("adp.requests", "count", &["a:1", "b:2"], 11)]);
216
217        assert!(Difference::between(&agent, &adp).is_empty());
218    }
219
220    #[test]
221    fn diverging_tags_each_member_with_its_lane() {
222        // A context only the Agent emitted is agent-side; one only ADP emitted is adp-side. Age is the
223        // intake clock minus first_seen.
224        let agent = lane(10, vec![captured("agent.only", "count", &["host:h"], 4)]);
225        let adp = lane(10, vec![captured("adp.only", "gauge", &["host:h"], 7)]);
226
227        let mut members = Difference::between(&agent, &adp).diverging();
228        members.sort_by(|a, b| a.context.name.cmp(&b.context.name));
229
230        assert!(matches!(members[0].lane, Lane::Adp));
231        assert_eq!(members[0].context.name, "adp.only");
232        assert_eq!(members[0].age_secs, 3);
233        assert!(matches!(members[1].lane, Lane::Agent));
234        assert_eq!(members[1].context.name, "agent.only");
235        assert_eq!(members[1].age_secs, 6);
236    }
237
238    #[test]
239    fn delayed_counts_members_past_the_budget() {
240        // A member is delayed once the intake clock has moved more than the budget past first_seen.
241        let present = vec![captured("adp.requests", "count", &["host:h"], 100)];
242        let fresh = Difference::between(&lane(140, present.clone()), &lane(140, vec![]));
243        let stale = Difference::between(&lane(200, present), &lane(200, vec![]));
244
245        assert_eq!(fresh.delayed(Duration::from_secs(60)), 0);
246        assert_eq!(stale.delayed(Duration::from_secs(60)), 1);
247    }
248
249    #[test]
250    fn deserializes_the_flat_wire_shape() -> Result<(), serde_json::Error> {
251        let view: LaneView = serde_json::from_value(json!({
252            "now": 2000,
253            "contexts": [{ "name": "adp.requests", "tagset": ["host:h"], "kind": "count", "first_seen": 1000 }],
254        }))?;
255
256        assert_eq!(view.now, 2000);
257        assert_eq!(view.contexts.len(), 1);
258        Ok(())
259    }
260}
antithesis_scenario_differential/contexts.rs

antithesis_scenario_differential/
contexts.rs
