omni_dev/datadog/
events_api.rs

1//! Datadog Events API wrapper.
2//!
3//! Exposes a thin façade over [`DatadogClient`] for the read-only events
4//! stream (`GET /api/v2/events`).
5//!
6//! Datadog v2 events use cursor pagination via `meta.page.after`.
7//! [`EventsApi::list`] issues a single request optionally seeded with an
8//! `after` cursor token; [`EventsApi::list_all`] auto-paginates up to a
9//! caller-supplied limit (or [`HARD_CAP`] when the limit is `0`),
10//! mirroring [`MonitorsApi::list`].
11//!
12//! [`MonitorsApi::list`]: crate::datadog::monitors_api::MonitorsApi::list
13
14use anyhow::Result;
15use url::Url;
16
17use crate::datadog::client::DatadogClient;
18use crate::datadog::types::EventsResponse;
19
20/// Per-page upper bound enforced by Datadog's v2 events API.
21pub const MAX_PAGE_LIMIT: usize = 1000;
22
23/// Per-call upper bound on the number of events returned by
24/// [`EventsApi::list_all`], even when the caller passes `limit = 0`.
25pub const HARD_CAP: usize = 10_000;
26
27/// Filters accepted by `GET /api/v2/events`.
28///
29/// Each field is optional: the URL builder appends a query parameter
30/// only when the field is `Some(_)`. `from` / `to` are Unix epoch
31/// **seconds** — the `EventsApi::list` method converts them to RFC 3339
32/// before sending, matching the Datadog v2 API expectations.
33#[derive(Debug, Default, Clone)]
34pub struct EventsListFilter {
35    /// Datadog events query string (e.g. `service:api`).
36    pub query: Option<String>,
37    /// Comma-separated list of source names (e.g. `aws,kubernetes`).
38    pub sources: Option<String>,
39    /// Comma-separated list of `key:value` tags.
40    pub tags: Option<String>,
41}
42
43/// Events API façade.
44#[derive(Debug)]
45pub struct EventsApi<'a> {
46    client: &'a DatadogClient,
47}
48
49impl<'a> EventsApi<'a> {
50    /// Wraps an existing [`DatadogClient`] for events operations.
51    #[must_use]
52    pub fn new(client: &'a DatadogClient) -> Self {
53        Self { client }
54    }
55
56    /// Lists events matching `filter` between `from` and `to` (RFC 3339
57    /// strings) capped at `limit` per page.
58    ///
59    /// Single-page only. When `after` is `Some`, Datadog resumes
60    /// pagination at that cursor token (`page[cursor]` in the query
61    /// string). The next-page token is preserved on `meta.page.after`
62    /// of the response so callers (or [`EventsApi::list_all`]) can
63    /// iterate.
64    pub async fn list(
65        &self,
66        filter: &EventsListFilter,
67        from: &str,
68        to: &str,
69        limit: usize,
70        after: Option<&str>,
71    ) -> Result<EventsResponse> {
72        if limit > MAX_PAGE_LIMIT {
73            return Err(anyhow::anyhow!(
74                "`limit` must be <= {MAX_PAGE_LIMIT} (Datadog v2 events per-page cap; use `EventsApi::list_all` to auto-paginate across pages)"
75            ));
76        }
77        let url = build_list_url(self.client.base_url(), filter, from, to, limit, after)?;
78        self.client
79            .get_parsed(url.as_str(), "Failed to parse /api/v2/events response")
80            .await
81    }
82
83    /// Lists events, auto-paginating via cursor as needed.
84    ///
85    /// `limit == 0` means "fetch every match up to [`HARD_CAP`]". Any
86    /// non-zero `limit` is upper-bounded by [`HARD_CAP`] to keep a single
87    /// invocation from issuing more than 10k items' worth of requests.
88    /// Per-request page size is clamped to [`MAX_PAGE_LIMIT`].
89    ///
90    /// Termination follows cursor-pagination semantics: the loop stops
91    /// when the response omits `meta.page.after` (Datadog signals "no
92    /// more pages" only via the absent cursor — a short page on its own
93    /// is *not* a terminator) or when `cap` items have been collected.
94    ///
95    /// The returned envelope keeps the `meta` and `links` blocks from the
96    /// *last* successful page so the response's cursor reflects the
97    /// iterator's final position (typically `None` when the API is
98    /// exhausted).
99    pub async fn list_all(
100        &self,
101        filter: &EventsListFilter,
102        from: &str,
103        to: &str,
104        limit: usize,
105    ) -> Result<EventsResponse> {
106        let cap = effective_cap(limit);
107        let mut acc: Option<EventsResponse> = None;
108        let mut cursor: Option<String> = None;
109        loop {
110            let collected = acc.as_ref().map_or(0, |r| r.data.len());
111            let remaining = cap - collected;
112            let page_size = remaining.min(MAX_PAGE_LIMIT);
113            let page = self
114                .list(filter, from, to, page_size, cursor.as_deref())
115                .await?;
116            let next_cursor = page
117                .meta
118                .as_ref()
119                .and_then(|m| m.page.as_ref())
120                .and_then(|p| p.after.clone());
121            match acc.as_mut() {
122                Some(existing) => {
123                    existing.data.extend(page.data);
124                    existing.meta = page.meta;
125                    existing.links = page.links;
126                }
127                None => acc = Some(page),
128            }
129            let collected = acc.as_ref().map_or(0, |r| r.data.len());
130            if collected >= cap || next_cursor.is_none() {
131                break;
132            }
133            cursor = next_cursor;
134        }
135        let mut result = acc.unwrap_or_default();
136        result.data.truncate(cap);
137        Ok(result)
138    }
139}
140
141/// Clamps a caller-supplied limit to [`HARD_CAP`], treating `0` as
142/// "fetch as many as the cap allows".
143fn effective_cap(limit: usize) -> usize {
144    if limit == 0 {
145        HARD_CAP
146    } else {
147        limit.min(HARD_CAP)
148    }
149}
150
151/// Builds `{base_url}/api/v2/events?filter[query]=…&filter[from]=…&filter[to]=…&page[limit]=N&page[cursor]=…`.
152fn build_list_url(
153    base_url: &str,
154    filter: &EventsListFilter,
155    from: &str,
156    to: &str,
157    limit: usize,
158    after: Option<&str>,
159) -> Result<Url> {
160    let mut url = DatadogClient::api_url(base_url, "/api/v2/events")?;
161    {
162        let mut q = url.query_pairs_mut();
163        if let Some(query) = filter.query.as_deref() {
164            q.append_pair("filter[query]", query);
165        }
166        if let Some(sources) = filter.sources.as_deref() {
167            q.append_pair("filter[sources]", sources);
168        }
169        if let Some(tags) = filter.tags.as_deref() {
170            q.append_pair("filter[tags]", tags);
171        }
172        q.append_pair("filter[from]", from);
173        q.append_pair("filter[to]", to);
174        q.append_pair("page[limit]", &limit.to_string());
175        if let Some(cursor) = after {
176            q.append_pair("page[cursor]", cursor);
177        }
178    }
179    Ok(url)
180}
181
182#[cfg(test)]
183#[allow(clippy::unwrap_used, clippy::expect_used)]
184mod tests {
185    use super::*;
186
187    // ── effective_cap ──────────────────────────────────────────────
188
189    #[test]
190    fn effective_cap_zero_means_hard_cap() {
191        assert_eq!(effective_cap(0), HARD_CAP);
192    }
193
194    #[test]
195    fn effective_cap_clamps_to_hard_cap() {
196        assert_eq!(effective_cap(HARD_CAP + 5), HARD_CAP);
197    }
198
199    #[test]
200    fn effective_cap_passes_through_small_limits() {
201        assert_eq!(effective_cap(42), 42);
202    }
203
204    // ── URL builder ────────────────────────────────────────────────
205
206    #[test]
207    fn build_list_url_appends_only_provided_filters() {
208        let filter = EventsListFilter {
209            query: Some("service:api".into()),
210            sources: None,
211            tags: None,
212        };
213        let url = build_list_url(
214            "https://api.datadoghq.com",
215            &filter,
216            "2026-04-22T09:00:00Z",
217            "2026-04-22T10:00:00Z",
218            50,
219            None,
220        )
221        .unwrap();
222        let qs = url.query().unwrap();
223        assert!(qs.contains("filter%5Bquery%5D=service%3Aapi"));
224        assert!(qs.contains("filter%5Bfrom%5D=2026-04-22T09%3A00%3A00Z"));
225        assert!(qs.contains("filter%5Bto%5D=2026-04-22T10%3A00%3A00Z"));
226        assert!(qs.contains("page%5Blimit%5D=50"));
227        assert!(!qs.contains("filter%5Bsources%5D"));
228        assert!(!qs.contains("filter%5Btags%5D"));
229        assert!(!qs.contains("page%5Bcursor%5D"));
230    }
231
232    #[test]
233    fn build_list_url_encodes_sources_and_tags() {
234        let filter = EventsListFilter {
235            query: None,
236            sources: Some("aws,kubernetes".into()),
237            tags: Some("env:prod,team:sre".into()),
238        };
239        let url = build_list_url(
240            "https://api.datadoghq.com",
241            &filter,
242            "2026-04-22T09:00:00Z",
243            "2026-04-22T10:00:00Z",
244            10,
245            None,
246        )
247        .unwrap();
248        let qs = url.query().unwrap();
249        assert!(qs.contains("filter%5Bsources%5D=aws%2Ckubernetes"));
250        assert!(qs.contains("filter%5Btags%5D=env%3Aprod%2Cteam%3Asre"));
251    }
252
253    #[test]
254    fn build_list_url_appends_cursor_when_provided() {
255        let url = build_list_url(
256            "https://api.datadoghq.com",
257            &EventsListFilter::default(),
258            "2026-04-22T09:00:00Z",
259            "2026-04-22T10:00:00Z",
260            10,
261            Some("tok-2"),
262        )
263        .unwrap();
264        let qs = url.query().unwrap();
265        assert!(qs.contains("page%5Bcursor%5D=tok-2"));
266    }
267
268    #[test]
269    fn build_list_url_rejects_invalid_base() {
270        let err = build_list_url(
271            "not a url",
272            &EventsListFilter::default(),
273            "2026-04-22T09:00:00Z",
274            "2026-04-22T10:00:00Z",
275            10,
276            None,
277        )
278        .unwrap_err();
279        assert!(err.to_string().contains("Invalid Datadog base URL"));
280    }
281
282    // ── fixtures ───────────────────────────────────────────────────
283
284    fn event_json(id: &str) -> serde_json::Value {
285        serde_json::json!({
286            "id": id,
287            "type": "event",
288            "attributes": {
289                "timestamp": "2026-04-22T10:00:00.000Z",
290                "title": "Deploy",
291                "source": "github",
292                "tags": ["env:prod"]
293            }
294        })
295    }
296
297    fn page_body(ids: &[&str], next_cursor: Option<&str>) -> serde_json::Value {
298        let data: Vec<serde_json::Value> = ids.iter().map(|id| event_json(id)).collect();
299        let meta = match next_cursor {
300            Some(c) => serde_json::json!({ "page": { "after": c }, "status": "done" }),
301            None => serde_json::json!({ "page": {}, "status": "done" }),
302        };
303        serde_json::json!({ "data": data, "meta": meta })
304    }
305
306    fn sample_body() -> serde_json::Value {
307        serde_json::json!({
308            "data": [event_json("EV1")],
309            "meta": {"page": {"after": "next"}, "status": "done"}
310        })
311    }
312
313    // ── happy path ─────────────────────────────────────────────────
314
315    #[tokio::test]
316    async fn list_sends_filters_and_returns_parsed_response() {
317        let server = wiremock::MockServer::start().await;
318        wiremock::Mock::given(wiremock::matchers::method("GET"))
319            .and(wiremock::matchers::path("/api/v2/events"))
320            .and(wiremock::matchers::query_param(
321                "filter[query]",
322                "service:api",
323            ))
324            .and(wiremock::matchers::query_param(
325                "filter[from]",
326                "2026-04-22T09:00:00Z",
327            ))
328            .and(wiremock::matchers::query_param(
329                "filter[to]",
330                "2026-04-22T10:00:00Z",
331            ))
332            .and(wiremock::matchers::query_param("page[limit]", "10"))
333            .and(wiremock::matchers::header("DD-API-KEY", "api"))
334            .and(wiremock::matchers::header("DD-APPLICATION-KEY", "app"))
335            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_body()))
336            .expect(1)
337            .mount(&server)
338            .await;
339
340        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
341        let result = EventsApi::new(&client)
342            .list(
343                &EventsListFilter {
344                    query: Some("service:api".into()),
345                    sources: None,
346                    tags: None,
347                },
348                "2026-04-22T09:00:00Z",
349                "2026-04-22T10:00:00Z",
350                10,
351                None,
352            )
353            .await
354            .unwrap();
355        assert_eq!(result.data.len(), 1);
356        assert_eq!(result.data[0].id, "EV1");
357    }
358
359    #[tokio::test]
360    async fn list_includes_cursor_in_query_when_after_is_some() {
361        let server = wiremock::MockServer::start().await;
362        wiremock::Mock::given(wiremock::matchers::method("GET"))
363            .and(wiremock::matchers::path("/api/v2/events"))
364            .and(wiremock::matchers::query_param("page[cursor]", "tok-2"))
365            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_body()))
366            .expect(1)
367            .mount(&server)
368            .await;
369
370        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
371        EventsApi::new(&client)
372            .list(
373                &EventsListFilter::default(),
374                "2026-04-22T09:00:00Z",
375                "2026-04-22T10:00:00Z",
376                10,
377                Some("tok-2"),
378            )
379            .await
380            .unwrap();
381    }
382
383    // ── client-side / API errors ───────────────────────────────────
384
385    #[tokio::test]
386    async fn list_rejects_limit_above_max_page_limit_client_side() {
387        let client = DatadogClient::new("http://127.0.0.1:1", "api", "app").unwrap();
388        let err = EventsApi::new(&client)
389            .list(
390                &EventsListFilter::default(),
391                "2026-04-22T09:00:00Z",
392                "2026-04-22T10:00:00Z",
393                MAX_PAGE_LIMIT + 1,
394                None,
395            )
396            .await
397            .unwrap_err();
398        assert!(err.to_string().contains("limit"));
399        assert!(err.to_string().contains(&MAX_PAGE_LIMIT.to_string()));
400        assert!(err.to_string().contains("list_all"));
401    }
402
403    #[tokio::test]
404    async fn list_propagates_api_errors() {
405        let server = wiremock::MockServer::start().await;
406        wiremock::Mock::given(wiremock::matchers::method("GET"))
407            .and(wiremock::matchers::path("/api/v2/events"))
408            .respond_with(
409                wiremock::ResponseTemplate::new(403).set_body_string(r#"{"errors":["nope"]}"#),
410            )
411            .mount(&server)
412            .await;
413
414        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
415        let err = EventsApi::new(&client)
416            .list(
417                &EventsListFilter::default(),
418                "2026-04-22T09:00:00Z",
419                "2026-04-22T10:00:00Z",
420                10,
421                None,
422            )
423            .await
424            .unwrap_err();
425        let msg = err.to_string();
426        assert!(msg.contains("403"));
427        assert!(msg.contains("nope"));
428    }
429
430    #[tokio::test]
431    async fn list_propagates_invalid_base_url_error() {
432        let client = DatadogClient::new("not a url", "api", "app").unwrap();
433        let err = EventsApi::new(&client)
434            .list(
435                &EventsListFilter::default(),
436                "2026-04-22T09:00:00Z",
437                "2026-04-22T10:00:00Z",
438                10,
439                None,
440            )
441            .await
442            .unwrap_err();
443        assert!(err.to_string().contains("Invalid Datadog base URL"));
444    }
445
446    #[tokio::test]
447    async fn list_propagates_network_errors() {
448        let client = DatadogClient::new("http://127.0.0.1:1", "api", "app").unwrap();
449        let err = EventsApi::new(&client)
450            .list(
451                &EventsListFilter::default(),
452                "2026-04-22T09:00:00Z",
453                "2026-04-22T10:00:00Z",
454                10,
455                None,
456            )
457            .await
458            .unwrap_err();
459        assert!(err.to_string().contains("Failed to send"));
460    }
461
462    #[tokio::test]
463    async fn list_errors_on_malformed_response() {
464        let server = wiremock::MockServer::start().await;
465        wiremock::Mock::given(wiremock::matchers::method("GET"))
466            .and(wiremock::matchers::path("/api/v2/events"))
467            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
468            .mount(&server)
469            .await;
470
471        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
472        let err = EventsApi::new(&client)
473            .list(
474                &EventsListFilter::default(),
475                "2026-04-22T09:00:00Z",
476                "2026-04-22T10:00:00Z",
477                10,
478                None,
479            )
480            .await
481            .unwrap_err();
482        assert!(err.to_string().contains("Failed to parse"));
483    }
484
485    // ── list_all ───────────────────────────────────────────────────
486
487    #[tokio::test]
488    async fn list_all_single_page_when_response_has_no_cursor() {
489        let server = wiremock::MockServer::start().await;
490        wiremock::Mock::given(wiremock::matchers::method("GET"))
491            .and(wiremock::matchers::path("/api/v2/events"))
492            .and(wiremock::matchers::query_param("page[limit]", "100"))
493            .respond_with(
494                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&["a", "b"], None)),
495            )
496            .expect(1)
497            .mount(&server)
498            .await;
499
500        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
501        let result = EventsApi::new(&client)
502            .list_all(
503                &EventsListFilter::default(),
504                "2026-04-22T09:00:00Z",
505                "2026-04-22T10:00:00Z",
506                100,
507            )
508            .await
509            .unwrap();
510        assert_eq!(result.data.len(), 2);
511    }
512
513    #[tokio::test]
514    async fn list_all_follows_cursor_until_no_more_pages() {
515        // Page 1 returns 2 events + cursor "c1"; page 2 returns 1 event +
516        // no cursor. With `limit == 0`, the loop should issue both
517        // requests and concatenate their data.
518        let server = wiremock::MockServer::start().await;
519        let limit_str = MAX_PAGE_LIMIT.to_string();
520        wiremock::Mock::given(wiremock::matchers::method("GET"))
521            .and(wiremock::matchers::path("/api/v2/events"))
522            .and(wiremock::matchers::query_param(
523                "page[limit]",
524                limit_str.as_str(),
525            ))
526            .and(wiremock::matchers::query_param_is_missing("page[cursor]"))
527            .respond_with(
528                wiremock::ResponseTemplate::new(200)
529                    .set_body_json(page_body(&["a", "b"], Some("c1"))),
530            )
531            .expect(1)
532            .mount(&server)
533            .await;
534        wiremock::Mock::given(wiremock::matchers::method("GET"))
535            .and(wiremock::matchers::path("/api/v2/events"))
536            .and(wiremock::matchers::query_param("page[cursor]", "c1"))
537            .respond_with(
538                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&["c"], None)),
539            )
540            .expect(1)
541            .mount(&server)
542            .await;
543
544        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
545        let result = EventsApi::new(&client)
546            .list_all(
547                &EventsListFilter::default(),
548                "2026-04-22T09:00:00Z",
549                "2026-04-22T10:00:00Z",
550                0,
551            )
552            .await
553            .unwrap();
554        let ids: Vec<&str> = result.data.iter().map(|e| e.id.as_str()).collect();
555        assert_eq!(ids, ["a", "b", "c"]);
556        assert!(result
557            .meta
558            .as_ref()
559            .and_then(|m| m.page.as_ref())
560            .and_then(|p| p.after.as_deref())
561            .is_none());
562    }
563
564    #[tokio::test]
565    async fn list_all_stops_at_explicit_limit_within_first_page() {
566        let server = wiremock::MockServer::start().await;
567        let ids = ["a", "b", "c"];
568        wiremock::Mock::given(wiremock::matchers::method("GET"))
569            .and(wiremock::matchers::path("/api/v2/events"))
570            .and(wiremock::matchers::query_param("page[limit]", "3"))
571            .respond_with(
572                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&ids, Some("c1"))),
573            )
574            .expect(1)
575            .mount(&server)
576            .await;
577
578        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
579        let result = EventsApi::new(&client)
580            .list_all(
581                &EventsListFilter::default(),
582                "2026-04-22T09:00:00Z",
583                "2026-04-22T10:00:00Z",
584                3,
585            )
586            .await
587            .unwrap();
588        assert_eq!(result.data.len(), 3);
589    }
590
591    #[tokio::test]
592    async fn list_all_truncates_to_hard_cap_when_unbounded() {
593        let server = wiremock::MockServer::start().await;
594        let full_page: Vec<serde_json::Value> = (0..MAX_PAGE_LIMIT)
595            .map(|i| event_json(&format!("e{i}")))
596            .collect();
597        let body = serde_json::json!({
598            "data": full_page,
599            "meta": { "page": { "after": "always-more" } }
600        });
601        wiremock::Mock::given(wiremock::matchers::method("GET"))
602            .and(wiremock::matchers::path("/api/v2/events"))
603            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body))
604            .mount(&server)
605            .await;
606
607        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
608        let result = EventsApi::new(&client)
609            .list_all(
610                &EventsListFilter::default(),
611                "2026-04-22T09:00:00Z",
612                "2026-04-22T10:00:00Z",
613                0,
614            )
615            .await
616            .unwrap();
617        assert_eq!(result.data.len(), HARD_CAP);
618    }
619
620    #[tokio::test]
621    async fn list_all_propagates_api_errors_on_first_page() {
622        let server = wiremock::MockServer::start().await;
623        wiremock::Mock::given(wiremock::matchers::method("GET"))
624            .and(wiremock::matchers::path("/api/v2/events"))
625            .respond_with(
626                wiremock::ResponseTemplate::new(403).set_body_string(r#"{"errors":["nope"]}"#),
627            )
628            .mount(&server)
629            .await;
630
631        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
632        let err = EventsApi::new(&client)
633            .list_all(
634                &EventsListFilter::default(),
635                "2026-04-22T09:00:00Z",
636                "2026-04-22T10:00:00Z",
637                0,
638            )
639            .await
640            .unwrap_err();
641        let msg = err.to_string();
642        assert!(msg.contains("403"));
643        assert!(msg.contains("nope"));
644    }
645
646    #[tokio::test]
647    async fn list_all_caps_explicit_limit_at_hard_cap() {
648        let server = wiremock::MockServer::start().await;
649        let full_page: Vec<serde_json::Value> = (0..MAX_PAGE_LIMIT)
650            .map(|i| event_json(&format!("e{i}")))
651            .collect();
652        let body = serde_json::json!({
653            "data": full_page,
654            "meta": { "page": { "after": "always-more" } }
655        });
656        wiremock::Mock::given(wiremock::matchers::method("GET"))
657            .and(wiremock::matchers::path("/api/v2/events"))
658            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body))
659            .mount(&server)
660            .await;
661
662        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
663        let result = EventsApi::new(&client)
664            .list_all(
665                &EventsListFilter::default(),
666                "2026-04-22T09:00:00Z",
667                "2026-04-22T10:00:00Z",
668                HARD_CAP + 50,
669            )
670            .await
671            .unwrap();
672        assert_eq!(result.data.len(), HARD_CAP);
673    }
674}
omni_dev/datadog/events_api.rs

omni_dev/datadog/
events_api.rs
