elasticctl_api/
alerts.rs

1//! Detection alerts: the signals search, status, tags, and assignees routes.
2//!
3//! Alert identity is the document `_id`. Every mutation here takes explicit
4//! ids or an explicit query; there is no reconciliation path (triage spec
5//! section 2).
6
7use elasticctl_core::{Error, ErrorKind, Result, Transport};
8use serde_json::{Value, json};
9
10pub const SEARCH_PATH: &str = "/api/detection_engine/signals/search";
11pub const STATUS_PATH: &str = "/api/detection_engine/signals/status";
12pub const TAGS_PATH: &str = "/api/detection_engine/signals/tags";
13pub const ASSIGNEES_PATH: &str = "/api/detection_engine/signals/assignees";
14
15/// The modern status vocabulary. The route also accepts `in-progress`, the
16/// pre-8.0 name `acknowledged` replaced; elasticctl never sends it.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub enum AlertStatus {
19    Open,
20    Acknowledged,
21    Closed,
22}
23
24impl AlertStatus {
25    pub fn as_str(self) -> &'static str {
26        match self {
27            AlertStatus::Open => "open",
28            AlertStatus::Acknowledged => "acknowledged",
29            AlertStatus::Closed => "closed",
30        }
31    }
32
33    /// The verb a preview banner uses: `Open 2 alerts`, `Close 1 alert`.
34    pub fn verb(self) -> &'static str {
35        match self {
36            AlertStatus::Open => "Open",
37            AlertStatus::Acknowledged => "Acknowledge",
38            AlertStatus::Closed => "Close",
39        }
40    }
41
42    pub fn parse(s: &str) -> Result<AlertStatus> {
43        match s {
44            "open" => Ok(AlertStatus::Open),
45            "acknowledged" => Ok(AlertStatus::Acknowledged),
46            "closed" => Ok(AlertStatus::Closed),
47            other => Err(Error::new(
48                ErrorKind::Error,
49                format!("unknown alert status '{other}': expected open, acknowledged, or closed"),
50            )),
51        }
52    }
53}
54
55/// Version-conflict handling for query-scoped transitions. `Abort` is the
56/// server default: a document whose version moved between resolution and
57/// write stops the run rather than being silently skipped.
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
59pub enum Conflicts {
60    #[default]
61    Abort,
62    Proceed,
63}
64
65impl Conflicts {
66    pub fn as_str(self) -> &'static str {
67        match self {
68            Conflicts::Abort => "abort",
69            Conflicts::Proceed => "proceed",
70        }
71    }
72
73    pub fn parse(s: &str) -> Result<Conflicts> {
74        match s {
75            "abort" => Ok(Conflicts::Abort),
76            "proceed" => Ok(Conflicts::Proceed),
77            other => Err(Error::new(
78                ErrorKind::Error,
79                format!("unknown conflicts mode '{other}': expected abort or proceed"),
80            )),
81        }
82    }
83}
84
85#[derive(Debug, Clone, PartialEq)]
86pub struct AlertHit {
87    /// The document `_id` — the identity every mutation route takes.
88    pub id: String,
89    /// The backing index, from `_index`. The cases attach body needs it.
90    pub index: Option<String>,
91    pub source: Value,
92    pub sort: Option<Vec<Value>>,
93}
94
95#[derive(Debug, Clone, PartialEq)]
96pub struct AlertPage {
97    pub hits: Vec<AlertHit>,
98    pub total: Option<u64>,
99}
100
101/// Decode a signals-search response. Fail-closed: `hits.hits` must be an
102/// array and every hit must carry a string `_id` — an alert without identity
103/// cannot be acted on — and an object `_source`.
104pub fn decode_page(value: &Value) -> Result<AlertPage> {
105    let hits = value
106        .pointer("/hits/hits")
107        .and_then(Value::as_array)
108        .ok_or_else(|| {
109            Error::new(
110                ErrorKind::Http,
111                "decoding alerts response field `hits.hits`",
112            )
113        })?;
114    let mut out = Vec::with_capacity(hits.len());
115    for hit in hits {
116        let id = hit
117            .get("_id")
118            .and_then(Value::as_str)
119            .ok_or_else(|| Error::new(ErrorKind::Http, "decoding alert hit field `_id`"))?
120            .to_string();
121        let index = hit.get("_index").and_then(Value::as_str).map(str::to_owned);
122        let source = hit
123            .get("_source")
124            .filter(|s| s.is_object())
125            .cloned()
126            .ok_or_else(|| Error::new(ErrorKind::Http, "decoding alert hit field `_source`"))?;
127        let sort = hit.get("sort").and_then(Value::as_array).cloned();
128        out.push(AlertHit {
129            id,
130            index,
131            source,
132            sort,
133        });
134    }
135    let total = value.pointer("/hits/total/value").and_then(Value::as_u64);
136    Ok(AlertPage { hits: out, total })
137}
138
139/// Run one bounded signals search with the caller's body verbatim.
140pub async fn search(t: &Transport, body: &Value) -> Result<AlertPage> {
141    decode_page(&t.post(SEARCH_PATH, Some(body)).await?)
142}
143
144/// Page a query fully with `sort` + `search_after` through the same route.
145/// `sort` must be a total order (the caller ends it with a tiebreaker field).
146pub async fn search_all(
147    t: &Transport,
148    query: &Value,
149    sort: &Value,
150    limit: Option<usize>,
151) -> Result<Vec<AlertHit>> {
152    search_all_with_page_size(t, query, sort, limit, 1000).await
153}
154
155/// The paging loop with an explicit page size, exposed for tests.
156pub async fn search_all_with_page_size(
157    t: &Transport,
158    query: &Value,
159    sort: &Value,
160    limit: Option<usize>,
161    page_size: usize,
162) -> Result<Vec<AlertHit>> {
163    let mut all = Vec::new();
164    let mut search_after: Option<Vec<Value>> = None;
165    loop {
166        let mut body = json!({
167            "query": query,
168            "sort": sort,
169            "size": page_size,
170        });
171        if let Some(sa) = &search_after {
172            body["search_after"] = json!(sa);
173        }
174        let page = search(t, &body).await?;
175        let short_page = page.hits.len() < page_size;
176        let last_sort = page.hits.last().and_then(|h| h.sort.clone());
177        all.extend(page.hits);
178        if let Some(limit) = limit
179            && all.len() >= limit
180        {
181            all.truncate(limit);
182            return Ok(all);
183        }
184        if short_page {
185            return Ok(all);
186        }
187        match last_sort {
188            Some(sa) => search_after = Some(sa),
189            // A full page whose last hit has no sort values cannot advance;
190            // stop rather than loop forever.
191            None => return Ok(all),
192        }
193    }
194}
195
196/// The raw update-by-query envelope the status, tags, and assignees routes
197/// answer with (measured, triage spec section 10).
198#[derive(Debug, Clone, PartialEq, serde::Serialize)]
199pub struct SignalsOutcome {
200    pub total: u64,
201    pub updated: u64,
202    pub version_conflicts: u64,
203    pub noops: u64,
204    pub failures: Vec<Value>,
205}
206
207/// Fail-closed decode: all four counters and the `failures` array are
208/// required. A response missing one is an `http` error, never "nothing
209/// happened" (main spec section 6.3).
210pub fn decode_outcome(value: &Value) -> Result<SignalsOutcome> {
211    let counter = |name: &str| {
212        value.get(name).and_then(Value::as_u64).ok_or_else(|| {
213            Error::new(
214                ErrorKind::Http,
215                format!("decoding signals outcome field `{name}`"),
216            )
217        })
218    };
219    Ok(SignalsOutcome {
220        total: counter("total")?,
221        updated: counter("updated")?,
222        version_conflicts: counter("version_conflicts")?,
223        noops: counter("noops")?,
224        failures: value
225            .get("failures")
226            .and_then(Value::as_array)
227            .cloned()
228            .ok_or_else(|| {
229                Error::new(ErrorKind::Http, "decoding signals outcome field `failures`")
230            })?,
231    })
232}
233
234/// Transition explicit alerts. The route is idempotent: a no-op transition
235/// counts as processed.
236pub async fn status_by_ids(
237    t: &Transport,
238    ids: &[String],
239    status: AlertStatus,
240    reason: Option<&str>,
241) -> Result<SignalsOutcome> {
242    let mut body = json!({ "signal_ids": ids, "status": status.as_str() });
243    if let Some(r) = reason {
244        body["reason"] = json!(r);
245    }
246    decode_outcome(&t.post(STATUS_PATH, Some(&body)).await?)
247}
248
249/// Transition every alert a query matches. The server resolves the set and
250/// mutates it in one update-by-query — no client-side id round-trip.
251pub async fn status_by_query(
252    t: &Transport,
253    query: &Value,
254    status: AlertStatus,
255    conflicts: Conflicts,
256    reason: Option<&str>,
257) -> Result<SignalsOutcome> {
258    let mut body = json!({
259        "query": query,
260        "status": status.as_str(),
261        "conflicts": conflicts.as_str(),
262    });
263    if let Some(r) = reason {
264        body["reason"] = json!(r);
265    }
266    decode_outcome(&t.post(STATUS_PATH, Some(&body)).await?)
267}
268
269/// Add and remove workflow tags on explicit alerts in one request.
270pub async fn set_tags(
271    t: &Transport,
272    ids: &[String],
273    add: &[String],
274    remove: &[String],
275) -> Result<SignalsOutcome> {
276    let body = json!({
277        "ids": ids,
278        "tags": { "tags_to_add": add, "tags_to_remove": remove },
279    });
280    decode_outcome(&t.post(TAGS_PATH, Some(&body)).await?)
281}
282
283/// Add and remove assignee profile uids on explicit alerts in one request.
284/// The route rejects a uid present in both lists; callers pre-check.
285pub async fn set_assignees(
286    t: &Transport,
287    ids: &[String],
288    add: &[String],
289    remove: &[String],
290) -> Result<SignalsOutcome> {
291    let body = json!({
292        "ids": ids,
293        "assignees": { "add": add, "remove": remove },
294    });
295    decode_outcome(&t.post(ASSIGNEES_PATH, Some(&body)).await?)
296}
elasticctl_api/alerts.rs

elasticctl_api/
alerts.rs
