omni_dev/datadog/
logs_api.rs

1//! Datadog Logs API wrapper.
2//!
3//! Exposes a thin façade over [`DatadogClient`] for the v2 logs search
4//! endpoint. Datadog v2 logs search uses **cursor pagination**
5//! (`meta.page.after`), not offset. [`LogsApi::search`] issues a single
6//! request optionally seeded with an `after` cursor token;
7//! [`LogsApi::search_all`] auto-paginates up to a caller-supplied limit
8//! (or [`HARD_CAP`] when the limit is `0`), mirroring [`MonitorsApi::list`].
9//!
10//! [`MonitorsApi::list`]: crate::datadog::monitors_api::MonitorsApi::list
11
12use anyhow::Result;
13use serde::Serialize;
14
15use crate::datadog::client::DatadogClient;
16use crate::datadog::types::{LogSearchResult, SortOrder};
17
18/// Maximum page size accepted by `POST /api/v2/logs/events/search`.
19///
20/// Datadog rejects page sizes above 1000; the API client surfaces a
21/// clearer error than the server's HTTP 400 by validating before the
22/// request is sent.
23pub const MAX_PAGE_LIMIT: usize = 1000;
24
25/// Per-call upper bound on the number of log events returned by
26/// [`LogsApi::search_all`], even when the caller passes `limit = 0`.
27pub const HARD_CAP: usize = 10_000;
28
29/// Logs API façade.
30#[derive(Debug)]
31pub struct LogsApi<'a> {
32    client: &'a DatadogClient,
33}
34
35impl<'a> LogsApi<'a> {
36    /// Wraps an existing [`DatadogClient`] for log operations.
37    #[must_use]
38    pub fn new(client: &'a DatadogClient) -> Self {
39        Self { client }
40    }
41
42    /// Searches log events.
43    ///
44    /// `from` and `to` are passed through to Datadog as strings. Datadog
45    /// accepts ISO 8601 timestamps, epoch milliseconds, and relative
46    /// shorthand like `now-15m` / `now`; callers are expected to convert
47    /// CLI-level inputs into a form Datadog understands before calling.
48    ///
49    /// Returns a single page only. When `after` is `Some`, Datadog
50    /// resumes pagination at that cursor token (`page.cursor` in the
51    /// request body). The next-page token is preserved on
52    /// `meta.page.after` of the response so the caller (or
53    /// [`LogsApi::search_all`]) can iterate.
54    ///
55    /// `limit` is rejected client-side when it exceeds [`MAX_PAGE_LIMIT`].
56    pub async fn search(
57        &self,
58        query: &str,
59        from: &str,
60        to: &str,
61        limit: usize,
62        sort: SortOrder,
63        after: Option<&str>,
64    ) -> Result<LogSearchResult> {
65        if limit > MAX_PAGE_LIMIT {
66            return Err(anyhow::anyhow!(
67                "`limit` must be <= {MAX_PAGE_LIMIT} (Datadog v2 logs search per-page cap; use `LogsApi::search_all` to auto-paginate across pages)"
68            ));
69        }
70        let body = SearchRequest {
71            filter: Filter { query, from, to },
72            page: Page {
73                limit,
74                cursor: after,
75            },
76            sort,
77        };
78        let url = format!("{}/api/v2/logs/events/search", self.client.base_url());
79        let response = self.client.post_json(&url, &body).await?;
80        self.client
81            .parse_response(
82                response,
83                "Failed to parse /api/v2/logs/events/search response",
84            )
85            .await
86    }
87
88    /// Searches log events, auto-paginating via cursor as needed.
89    ///
90    /// `limit == 0` means "fetch every match up to [`HARD_CAP`]". Any
91    /// non-zero `limit` is upper-bounded by [`HARD_CAP`] to keep a single
92    /// invocation from issuing more than 10k items' worth of requests.
93    /// Per-request page size is clamped to [`MAX_PAGE_LIMIT`].
94    ///
95    /// Termination follows cursor-pagination semantics: the loop stops
96    /// when the response omits `meta.page.after` (Datadog signals "no
97    /// more pages" only via the absent cursor — a short page on its own
98    /// is *not* a terminator) or when `cap` items have been collected.
99    ///
100    /// The returned envelope keeps the `meta` block from the *last*
101    /// successful page so the response's cursor reflects the iterator's
102    /// final position (typically `None` when the API is exhausted).
103    pub async fn search_all(
104        &self,
105        query: &str,
106        from: &str,
107        to: &str,
108        limit: usize,
109        sort: SortOrder,
110    ) -> Result<LogSearchResult> {
111        let cap = effective_cap(limit);
112        let mut acc: Option<LogSearchResult> = None;
113        let mut cursor: Option<String> = None;
114        loop {
115            let collected = acc.as_ref().map_or(0, |r| r.data.len());
116            let remaining = cap - collected;
117            let page_size = remaining.min(MAX_PAGE_LIMIT);
118            let page = self
119                .search(query, from, to, page_size, sort, cursor.as_deref())
120                .await?;
121            let next_cursor = page
122                .meta
123                .as_ref()
124                .and_then(|m| m.page.as_ref())
125                .and_then(|p| p.after.clone());
126            match acc.as_mut() {
127                Some(existing) => {
128                    existing.data.extend(page.data);
129                    existing.meta = page.meta;
130                }
131                None => acc = Some(page),
132            }
133            let collected = acc.as_ref().map_or(0, |r| r.data.len());
134            if collected >= cap || next_cursor.is_none() {
135                break;
136            }
137            cursor = next_cursor;
138        }
139        let mut result = acc.unwrap_or_default();
140        result.data.truncate(cap);
141        Ok(result)
142    }
143}
144
145/// Clamps a caller-supplied limit to [`HARD_CAP`], treating `0` as
146/// "fetch as many as the cap allows".
147fn effective_cap(limit: usize) -> usize {
148    if limit == 0 {
149        HARD_CAP
150    } else {
151        limit.min(HARD_CAP)
152    }
153}
154
155#[derive(Debug, Serialize)]
156struct SearchRequest<'a> {
157    filter: Filter<'a>,
158    page: Page<'a>,
159    sort: SortOrder,
160}
161
162#[derive(Debug, Serialize)]
163struct Filter<'a> {
164    query: &'a str,
165    from: &'a str,
166    to: &'a str,
167}
168
169#[derive(Debug, Serialize)]
170struct Page<'a> {
171    limit: usize,
172    #[serde(skip_serializing_if = "Option::is_none")]
173    cursor: Option<&'a str>,
174}
175
176#[cfg(test)]
177#[allow(clippy::unwrap_used, clippy::expect_used)]
178mod tests {
179    use super::*;
180
181    fn sample_search_body() -> serde_json::Value {
182        serde_json::json!({
183            "data": [
184                {
185                    "id": "AAAA",
186                    "type": "log",
187                    "attributes": {
188                        "timestamp": "2026-04-22T10:00:00.000Z",
189                        "service": "api",
190                        "status": "info",
191                        "message": "hello",
192                        "tags": ["env:prod"]
193                    }
194                }
195            ],
196            "meta": {
197                "page": { "after": "next-cursor" },
198                "status": "done",
199                "elapsed": 12
200            }
201        })
202    }
203
204    fn log_event_json(id: &str) -> serde_json::Value {
205        serde_json::json!({
206            "id": id,
207            "type": "log",
208            "attributes": {
209                "timestamp": "2026-04-22T10:00:00.000Z",
210                "service": "api",
211                "status": "info",
212                "message": id,
213                "tags": []
214            }
215        })
216    }
217
218    fn page_body(ids: &[&str], next_cursor: Option<&str>) -> serde_json::Value {
219        let data: Vec<serde_json::Value> = ids.iter().map(|id| log_event_json(id)).collect();
220        let meta = match next_cursor {
221            Some(c) => serde_json::json!({ "page": { "after": c }, "status": "done" }),
222            None => serde_json::json!({ "page": {}, "status": "done" }),
223        };
224        serde_json::json!({ "data": data, "meta": meta })
225    }
226
227    // ── effective_cap ──────────────────────────────────────────────
228
229    #[test]
230    fn effective_cap_zero_means_hard_cap() {
231        assert_eq!(effective_cap(0), HARD_CAP);
232    }
233
234    #[test]
235    fn effective_cap_clamps_to_hard_cap() {
236        assert_eq!(effective_cap(HARD_CAP + 5), HARD_CAP);
237    }
238
239    #[test]
240    fn effective_cap_passes_through_small_limits() {
241        assert_eq!(effective_cap(42), 42);
242    }
243
244    // ── search ─────────────────────────────────────────────────────
245
246    #[tokio::test]
247    async fn search_posts_exact_body_shape_and_parses_response() {
248        let server = wiremock::MockServer::start().await;
249        wiremock::Mock::given(wiremock::matchers::method("POST"))
250            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
251            .and(wiremock::matchers::header("DD-API-KEY", "api"))
252            .and(wiremock::matchers::header("DD-APPLICATION-KEY", "app"))
253            .and(wiremock::matchers::header(
254                "Content-Type",
255                "application/json",
256            ))
257            .and(wiremock::matchers::body_json(serde_json::json!({
258                "filter": {
259                    "query": "service:api status:error",
260                    "from": "now-15m",
261                    "to": "now"
262                },
263                "page": { "limit": 100 },
264                "sort": "-timestamp"
265            })))
266            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_search_body()))
267            .expect(1)
268            .mount(&server)
269            .await;
270
271        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
272        let result = LogsApi::new(&client)
273            .search(
274                "service:api status:error",
275                "now-15m",
276                "now",
277                100,
278                SortOrder::TimestampDesc,
279                None,
280            )
281            .await
282            .unwrap();
283        assert_eq!(result.data.len(), 1);
284        assert_eq!(result.data[0].id, "AAAA");
285        assert_eq!(
286            result
287                .meta
288                .as_ref()
289                .and_then(|m| m.page.as_ref())
290                .and_then(|p| p.after.as_deref()),
291            Some("next-cursor")
292        );
293    }
294
295    #[tokio::test]
296    async fn search_includes_cursor_in_body_when_after_is_some() {
297        let server = wiremock::MockServer::start().await;
298        wiremock::Mock::given(wiremock::matchers::method("POST"))
299            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
300            .and(wiremock::matchers::body_json(serde_json::json!({
301                "filter": { "query": "*", "from": "now-1h", "to": "now" },
302                "page": { "limit": 50, "cursor": "tok-2" },
303                "sort": "-timestamp"
304            })))
305            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_search_body()))
306            .expect(1)
307            .mount(&server)
308            .await;
309
310        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
311        LogsApi::new(&client)
312            .search(
313                "*",
314                "now-1h",
315                "now",
316                50,
317                SortOrder::TimestampDesc,
318                Some("tok-2"),
319            )
320            .await
321            .unwrap();
322    }
323
324    #[tokio::test]
325    async fn search_serializes_ascending_sort_without_minus_prefix() {
326        let server = wiremock::MockServer::start().await;
327        wiremock::Mock::given(wiremock::matchers::method("POST"))
328            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
329            .and(wiremock::matchers::body_json(serde_json::json!({
330                "filter": { "query": "*", "from": "now-1h", "to": "now" },
331                "page": { "limit": 50 },
332                "sort": "timestamp"
333            })))
334            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_search_body()))
335            .expect(1)
336            .mount(&server)
337            .await;
338
339        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
340        LogsApi::new(&client)
341            .search("*", "now-1h", "now", 50, SortOrder::TimestampAsc, None)
342            .await
343            .unwrap();
344    }
345
346    #[tokio::test]
347    async fn search_rejects_limit_above_max_page_limit_client_side() {
348        let client = DatadogClient::new("http://127.0.0.1:1", "api", "app").unwrap();
349        let err = LogsApi::new(&client)
350            .search(
351                "*",
352                "now-1h",
353                "now",
354                MAX_PAGE_LIMIT + 1,
355                SortOrder::TimestampDesc,
356                None,
357            )
358            .await
359            .unwrap_err();
360        let msg = err.to_string();
361        assert!(msg.contains("limit"));
362        assert!(msg.contains("1000"));
363        assert!(msg.contains("search_all"));
364    }
365
366    #[tokio::test]
367    async fn search_accepts_limit_at_max_page_limit_boundary() {
368        let server = wiremock::MockServer::start().await;
369        wiremock::Mock::given(wiremock::matchers::method("POST"))
370            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
371            .and(wiremock::matchers::body_json(serde_json::json!({
372                "filter": { "query": "*", "from": "now-1h", "to": "now" },
373                "page": { "limit": MAX_PAGE_LIMIT },
374                "sort": "-timestamp"
375            })))
376            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(sample_search_body()))
377            .expect(1)
378            .mount(&server)
379            .await;
380
381        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
382        LogsApi::new(&client)
383            .search(
384                "*",
385                "now-1h",
386                "now",
387                MAX_PAGE_LIMIT,
388                SortOrder::TimestampDesc,
389                None,
390            )
391            .await
392            .unwrap();
393    }
394
395    #[tokio::test]
396    async fn search_propagates_api_errors() {
397        let server = wiremock::MockServer::start().await;
398        wiremock::Mock::given(wiremock::matchers::method("POST"))
399            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
400            .respond_with(
401                wiremock::ResponseTemplate::new(400).set_body_string(r#"{"errors":["bad query"]}"#),
402            )
403            .mount(&server)
404            .await;
405
406        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
407        let err = LogsApi::new(&client)
408            .search("???", "now-1h", "now", 10, SortOrder::TimestampDesc, None)
409            .await
410            .unwrap_err();
411        let msg = err.to_string();
412        assert!(msg.contains("400"));
413        assert!(msg.contains("bad query"));
414    }
415
416    #[tokio::test]
417    async fn search_propagates_network_errors() {
418        let client = DatadogClient::new("http://127.0.0.1:1", "api", "app").unwrap();
419        let err = LogsApi::new(&client)
420            .search("*", "now-1h", "now", 10, SortOrder::TimestampDesc, None)
421            .await
422            .unwrap_err();
423        assert!(err.to_string().contains("Failed to send"));
424    }
425
426    #[tokio::test]
427    async fn search_errors_on_malformed_response() {
428        let server = wiremock::MockServer::start().await;
429        wiremock::Mock::given(wiremock::matchers::method("POST"))
430            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
431            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
432            .mount(&server)
433            .await;
434
435        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
436        let err = LogsApi::new(&client)
437            .search("*", "now-1h", "now", 10, SortOrder::TimestampDesc, None)
438            .await
439            .unwrap_err();
440        assert!(err.to_string().contains("Failed to parse"));
441    }
442
443    // ── search_all ─────────────────────────────────────────────────
444
445    #[tokio::test]
446    async fn search_all_single_page_when_response_has_no_cursor() {
447        let server = wiremock::MockServer::start().await;
448        wiremock::Mock::given(wiremock::matchers::method("POST"))
449            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
450            .and(wiremock::matchers::body_json(serde_json::json!({
451                "filter": { "query": "*", "from": "now-1h", "to": "now" },
452                "page": { "limit": 100 },
453                "sort": "-timestamp"
454            })))
455            .respond_with(
456                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&["a", "b"], None)),
457            )
458            .expect(1)
459            .mount(&server)
460            .await;
461
462        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
463        let result = LogsApi::new(&client)
464            .search_all("*", "now-1h", "now", 100, SortOrder::TimestampDesc)
465            .await
466            .unwrap();
467        assert_eq!(result.data.len(), 2);
468    }
469
470    #[tokio::test]
471    async fn search_all_follows_cursor_until_no_more_pages() {
472        // Page 1 returns 2 items + cursor "c1", page 2 returns 1 item + no
473        // cursor. With `limit == 0`, the loop should issue both requests
474        // and concatenate their data.
475        let server = wiremock::MockServer::start().await;
476        wiremock::Mock::given(wiremock::matchers::method("POST"))
477            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
478            .and(wiremock::matchers::body_json(serde_json::json!({
479                "filter": { "query": "*", "from": "now-1h", "to": "now" },
480                "page": { "limit": MAX_PAGE_LIMIT },
481                "sort": "-timestamp"
482            })))
483            .respond_with(
484                wiremock::ResponseTemplate::new(200)
485                    .set_body_json(page_body(&["a", "b"], Some("c1"))),
486            )
487            .expect(1)
488            .mount(&server)
489            .await;
490        wiremock::Mock::given(wiremock::matchers::method("POST"))
491            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
492            .and(wiremock::matchers::body_json(serde_json::json!({
493                "filter": { "query": "*", "from": "now-1h", "to": "now" },
494                "page": { "limit": MAX_PAGE_LIMIT, "cursor": "c1" },
495                "sort": "-timestamp"
496            })))
497            .respond_with(
498                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&["c"], None)),
499            )
500            .expect(1)
501            .mount(&server)
502            .await;
503
504        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
505        let result = LogsApi::new(&client)
506            .search_all("*", "now-1h", "now", 0, SortOrder::TimestampDesc)
507            .await
508            .unwrap();
509        let ids: Vec<&str> = result.data.iter().map(|e| e.id.as_str()).collect();
510        assert_eq!(ids, ["a", "b", "c"]);
511        // The final response had no cursor — meta.page.after is None.
512        assert!(result
513            .meta
514            .as_ref()
515            .and_then(|m| m.page.as_ref())
516            .and_then(|p| p.after.as_deref())
517            .is_none());
518    }
519
520    #[tokio::test]
521    async fn search_all_stops_at_explicit_limit_within_first_page() {
522        // limit=5 → page_size=5; the API returns exactly 5 (a "short
523        // page" by user's request) so we stop without a second call.
524        let server = wiremock::MockServer::start().await;
525        let ids = ["a", "b", "c", "d", "e"];
526        wiremock::Mock::given(wiremock::matchers::method("POST"))
527            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
528            .and(wiremock::matchers::body_json(serde_json::json!({
529                "filter": { "query": "*", "from": "now-1h", "to": "now" },
530                "page": { "limit": 5 },
531                "sort": "-timestamp"
532            })))
533            .respond_with(
534                wiremock::ResponseTemplate::new(200).set_body_json(page_body(&ids, Some("c1"))),
535            )
536            .expect(1)
537            .mount(&server)
538            .await;
539
540        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
541        let result = LogsApi::new(&client)
542            .search_all("*", "now-1h", "now", 5, SortOrder::TimestampDesc)
543            .await
544            .unwrap();
545        assert_eq!(result.data.len(), 5);
546    }
547
548    #[tokio::test]
549    async fn search_all_truncates_to_hard_cap_when_unbounded() {
550        // The mock returns a full MAX_PAGE_LIMIT page + cursor on every
551        // request, so the only stopping condition is HARD_CAP.
552        let server = wiremock::MockServer::start().await;
553        let full_page: Vec<serde_json::Value> = (0..MAX_PAGE_LIMIT)
554            .map(|i| log_event_json(&format!("e{i}")))
555            .collect();
556        let body = serde_json::json!({
557            "data": full_page,
558            "meta": { "page": { "after": "always-more" } }
559        });
560        wiremock::Mock::given(wiremock::matchers::method("POST"))
561            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
562            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body))
563            .mount(&server)
564            .await;
565
566        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
567        let result = LogsApi::new(&client)
568            .search_all("*", "now-1h", "now", 0, SortOrder::TimestampDesc)
569            .await
570            .unwrap();
571        assert_eq!(result.data.len(), HARD_CAP);
572    }
573
574    #[tokio::test]
575    async fn search_all_propagates_api_errors_on_first_page() {
576        let server = wiremock::MockServer::start().await;
577        wiremock::Mock::given(wiremock::matchers::method("POST"))
578            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
579            .respond_with(
580                wiremock::ResponseTemplate::new(403).set_body_string(r#"{"errors":["nope"]}"#),
581            )
582            .mount(&server)
583            .await;
584
585        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
586        let err = LogsApi::new(&client)
587            .search_all("*", "now-1h", "now", 0, SortOrder::TimestampDesc)
588            .await
589            .unwrap_err();
590        let msg = err.to_string();
591        assert!(msg.contains("403"));
592        assert!(msg.contains("nope"));
593    }
594
595    #[tokio::test]
596    async fn search_all_caps_explicit_limit_at_hard_cap() {
597        // Caller asked for HARD_CAP + 50; per-page mock returns full pages
598        // forever. We stop at HARD_CAP without honouring the surplus.
599        let server = wiremock::MockServer::start().await;
600        let full_page: Vec<serde_json::Value> = (0..MAX_PAGE_LIMIT)
601            .map(|i| log_event_json(&format!("e{i}")))
602            .collect();
603        let body = serde_json::json!({
604            "data": full_page,
605            "meta": { "page": { "after": "always-more" } }
606        });
607        wiremock::Mock::given(wiremock::matchers::method("POST"))
608            .and(wiremock::matchers::path("/api/v2/logs/events/search"))
609            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(body))
610            .mount(&server)
611            .await;
612
613        let client = DatadogClient::new(&server.uri(), "api", "app").unwrap();
614        let result = LogsApi::new(&client)
615            .search_all(
616                "*",
617                "now-1h",
618                "now",
619                HARD_CAP + 50,
620                SortOrder::TimestampDesc,
621            )
622            .await
623            .unwrap();
624        assert_eq!(result.data.len(), HARD_CAP);
625    }
626}
omni_dev/datadog/logs_api.rs

omni_dev/datadog/
logs_api.rs
