elasticctl_api/
alerts_ops.rs

1//! Alert orchestration: filter construction, list/get, and the triage
2//! mutation plans behind the CLI guard.
3
4mod recheck;
5pub use recheck::{AlertFacet, observe_alerts};
6mod observed;
7pub use observed::{plan_status_by_ids_observed, plan_tags_observed};
8
9use crate::alerts::{self, AlertHit, AlertStatus, Conflicts, SignalsOutcome};
10use crate::{profiles, selection};
11use elasticctl_core::{Error, ErrorKind, Result, Transport};
12use serde_json::{Value, json};
13
14/// The `alerts list` filter set. Every field composes into one boolean query
15/// over the measured `kibana.alert.*` fields (triage spec section 4).
16#[derive(Debug, Clone, Default)]
17pub struct AlertFilter {
18    pub status: Option<AlertStatus>,
19    pub severity: Option<String>,
20    /// A rule name or `rule_id`, resolved through the standard name-or-id
21    /// resolution before filtering on `kibana.alert.rule.rule_id`.
22    pub rule: Option<String>,
23    pub tag: Option<String>,
24    /// A username or `uid:<profile_uid>`, resolved to a profile uid.
25    pub assignee: Option<String>,
26    /// A duration (`90m`, `24h`, `7d`) or an ISO timestamp.
27    pub since: Option<String>,
28    /// Substring match on the rule name and reason text.
29    pub search: Option<String>,
30}
31
32/// `--since` as a range clause: `<digits><s|m|h|d|w>` becomes `now-<dur>`;
33/// anything else passes through verbatim for the server to validate as a
34/// timestamp or date-math expression.
35pub fn since_clause(since: &str) -> Value {
36    let bytes = since.as_bytes();
37    let is_duration = bytes.len() >= 2
38        && bytes[..bytes.len() - 1].iter().all(u8::is_ascii_digit)
39        && matches!(bytes[bytes.len() - 1], b's' | b'm' | b'h' | b'd' | b'w');
40    let gte = if is_duration {
41        format!("now-{since}")
42    } else {
43        since.to_string()
44    };
45    json!({"range": {"@timestamp": {"gte": gte}}})
46}
47
48/// Newest first, with `kibana.alert.uuid` as the total-order tiebreaker
49/// `search_after` needs.
50pub fn default_sort() -> Value {
51    json!([
52        {"@timestamp": {"order": "desc"}},
53        {"kibana.alert.uuid": {"order": "asc"}}
54    ])
55}
56
57/// Escape `\`, `*`, and `?` so `--search` text cannot inject its own
58/// Elasticsearch `wildcard` metacharacters into the substring match. Escape
59/// backslashes first, or an inserted `\*`/`\?` escape would itself be
60/// re-escaped. Mirrors `rules::kql_escape_wildcard`'s ordering, though this
61/// value sits in a plain Query DSL string, not a quoted KQL literal, so no
62/// quote-escaping applies here.
63fn escape_wildcard(value: &str) -> String {
64    value
65        .replace('\\', "\\\\")
66        .replace('*', "\\*")
67        .replace('?', "\\?")
68}
69
70/// Compose the filter into one boolean query, resolving `rule` and
71/// `assignee` first. An empty filter is an explicit `match_all`.
72pub async fn build_query(t: &Transport, f: &AlertFilter) -> Result<Value> {
73    let mut filter = Vec::new();
74    if let Some(status) = f.status {
75        filter.push(json!({"term": {"kibana.alert.workflow_status": status.as_str()}}));
76    }
77    if let Some(severity) = &f.severity {
78        filter.push(json!({"term": {"kibana.alert.severity": severity}}));
79    }
80    if let Some(rule) = &f.rule {
81        let rule_id = selection::to_rule_id(t, rule).await?;
82        filter.push(json!({"term": {"kibana.alert.rule.rule_id": rule_id}}));
83    }
84    if let Some(tag) = &f.tag {
85        filter.push(json!({"term": {"kibana.alert.workflow_tags": tag}}));
86    }
87    if let Some(assignee) = &f.assignee {
88        let uid = profiles::resolve_assignee(t, assignee).await?;
89        filter.push(json!({"term": {"kibana.alert.workflow_assignee_ids": uid}}));
90    }
91    if let Some(since) = &f.since {
92        filter.push(since_clause(since));
93    }
94    if let Some(text) = &f.search {
95        let pattern = format!("*{}*", escape_wildcard(text));
96        filter.push(json!({"bool": {"minimum_should_match": 1, "should": [
97            {"wildcard": {"kibana.alert.rule.name": {"value": pattern, "case_insensitive": true}}},
98            {"wildcard": {"kibana.alert.reason": {"value": pattern, "case_insensitive": true}}}
99        ]}}));
100    }
101    if filter.is_empty() {
102        Ok(json!({"match_all": {}}))
103    } else {
104        Ok(json!({"bool": {"filter": filter}}))
105    }
106}
107
108#[derive(Debug, Clone, PartialEq)]
109pub struct AlertList {
110    pub hits: Vec<AlertHit>,
111    pub total: Option<u64>,
112    pub truncated: bool,
113}
114
115/// One bounded peek: `limit + 1` rows so truncation is observable without a
116/// second request.
117pub async fn list(t: &Transport, f: &AlertFilter, limit: usize) -> Result<AlertList> {
118    let query = build_query(t, f).await?;
119    let body = json!({
120        "query": query,
121        "sort": default_sort(),
122        "size": limit.saturating_add(1),
123        "track_total_hits": true,
124    });
125    let mut page = alerts::search(t, &body).await?;
126    let truncated = page.hits.len() > limit;
127    page.hits.truncate(limit);
128    Ok(AlertList {
129        hits: page.hits,
130        total: page.total,
131        truncated,
132    })
133}
134
135/// The `--out` path: page the filtered set fully, or stop at `limit` rows
136/// when the caller passes one (matching `search dsl --out --limit`).
137pub async fn export(t: &Transport, f: &AlertFilter, limit: Option<usize>) -> Result<Vec<AlertHit>> {
138    let query = build_query(t, f).await?;
139    alerts::search_all(t, &query, &default_sort(), limit).await
140}
141
142/// `alerts get`: an `_id`-filtered search returning one document.
143pub async fn get_one(t: &Transport, alert_id: &str) -> Result<AlertHit> {
144    let body = json!({"query": {"ids": {"values": [alert_id]}}, "size": 1});
145    let page = alerts::search(t, &body).await?;
146    page.hits.into_iter().next().ok_or_else(|| {
147        Error::new(
148            ErrorKind::NotFound,
149            format!("No alert with id '{alert_id}'"),
150        )
151    })
152}
153
154/// One explicitly named alert, resolved before a preview.
155#[derive(Debug, Clone, PartialEq, Eq)]
156pub struct ResolvedAlert {
157    pub id: String,
158    pub rule_name: String,
159    pub status: String,
160}
161
162/// `_source` fields `resolve_ids` requests: only what a mutation preview
163/// renders (rule name and current workflow status), not the whole document.
164/// `pub` so the fixture recorder can send the identical production body
165/// instead of a hand-rolled approximation (triage spec section 10).
166pub const RESOLVE_SOURCE_FIELDS: &[&str] =
167    &["kibana.alert.rule.name", "kibana.alert.workflow_status"];
168
169/// Alert documents store dotted field names as flat `_source` keys; older
170/// pipelines may nest them. Read both shapes. `pub(crate)` so `cases_ops`
171/// shares this instead of carrying its own copy (finding 11).
172pub(crate) fn source_str<'a>(source: &'a Value, key: &str) -> Option<&'a str> {
173    if let Some(v) = source.get(key).and_then(Value::as_str) {
174        return Some(v);
175    }
176    let pointer = format!("/{}", key.replace('.', "/"));
177    source.pointer(&pointer).and_then(Value::as_str)
178}
179
180fn workflow_status(source: &Value) -> &str {
181    source_str(source, "kibana.alert.workflow_status").unwrap_or("unknown")
182}
183
184/// Build the search body for explicit alert IDs.
185fn resolve_body(unique: &[String]) -> Value {
186    json!({
187        "query": {"ids": {"values": unique}},
188        "size": unique.len(),
189        "_source": RESOLVE_SOURCE_FIELDS,
190    })
191}
192
193/// Resolve explicit ids to alerts. Fail-closed: every id must resolve or the
194/// command refuses to proceed on the partial set (main spec section 6.3).
195/// Duplicate ids are collapsed, preserving first-seen order.
196async fn resolve_ids(t: &Transport, ids: &[String]) -> Result<Vec<ResolvedAlert>> {
197    let mut unique: Vec<String> = Vec::with_capacity(ids.len());
198    for id in ids {
199        if !unique.contains(id) {
200            unique.push(id.clone());
201        }
202    }
203    let body = resolve_body(&unique);
204    let page = alerts::search(t, &body).await?;
205    let mut resolved = Vec::with_capacity(unique.len());
206    for id in &unique {
207        let hit = page.hits.iter().find(|h| &h.id == id).ok_or_else(|| {
208            let missing: Vec<&str> = unique
209                .iter()
210                .filter(|i| !page.hits.iter().any(|h| &h.id == *i))
211                .map(String::as_str)
212                .collect();
213            Error::new(
214                ErrorKind::NotFound,
215                format!("No alert with id: {}", missing.join(", ")),
216            )
217        })?;
218        resolved.push(ResolvedAlert {
219            id: id.clone(),
220            rule_name: source_str(&hit.source, "kibana.alert.rule.name")
221                .unwrap_or("(unnamed rule)")
222                .to_string(),
223            status: workflow_status(&hit.source).to_string(),
224        });
225    }
226    Ok(resolved)
227}
228
229fn alert_noun(n: usize) -> &'static str {
230    if n == 1 { "alert" } else { "alerts" }
231}
232
233/// `1214` → `1,214`, matching the preview format in triage spec section 6.
234fn thousands(n: u64) -> String {
235    let digits = n.to_string();
236    let mut out = String::with_capacity(digits.len() + digits.len() / 3);
237    for (i, c) in digits.chars().enumerate() {
238        if i > 0 && (digits.len() - i).is_multiple_of(3) {
239            out.push(',');
240        }
241        out.push(c);
242    }
243    out
244}
245
246#[derive(Debug, Clone, PartialEq)]
247pub struct StatusPlan {
248    pub status: AlertStatus,
249    pub reason: Option<String>,
250    pub targets: Vec<String>,
251    pub preview_action: String,
252    pub preview_details: Vec<String>,
253}
254
255pub async fn plan_status_by_ids(
256    t: &Transport,
257    ids: &[String],
258    status: AlertStatus,
259    reason: Option<String>,
260) -> Result<StatusPlan> {
261    Ok(plan_status_by_ids_observed(t, ids, status, reason).await?.0)
262}
263
264/// The mutation report the CLI renders. `failed` is elasticctl's judgment —
265/// route failures plus, under `--conflicts abort`, version conflicts — and
266/// drives the non-zero exit; the verbatim counters render beside it.
267#[derive(Debug, Clone, PartialEq, serde::Serialize)]
268pub struct StatusReport {
269    pub applied: bool,
270    pub status: String,
271    pub total: u64,
272    pub updated: u64,
273    pub version_conflicts: u64,
274    pub noops: u64,
275    pub failed: u64,
276    pub failures: Vec<Value>,
277}
278
279fn failed_count(outcome: &SignalsOutcome, conflicts: Conflicts) -> Result<u64> {
280    (outcome.failures.len() as u64)
281        .checked_add(match conflicts {
282            Conflicts::Abort => outcome.version_conflicts,
283            Conflicts::Proceed => 0,
284        })
285        .ok_or_else(|| {
286            Error::new(
287                ErrorKind::Http,
288                "decoding alert response: counters are unreadable (failure count overflows u64)",
289            )
290        })
291}
292
293fn status_report(
294    status: AlertStatus,
295    outcome: SignalsOutcome,
296    conflicts: Conflicts,
297) -> Result<StatusReport> {
298    Ok(StatusReport {
299        applied: true,
300        status: status.as_str().to_string(),
301        total: outcome.total,
302        updated: outcome.updated,
303        version_conflicts: outcome.version_conflicts,
304        noops: outcome.noops,
305        failed: failed_count(&outcome, conflicts)?,
306        failures: outcome.failures,
307    })
308}
309
310pub async fn apply_status_by_ids(t: &Transport, plan: &StatusPlan) -> Result<StatusReport> {
311    require_targets(&plan.targets)?;
312    let outcome =
313        alerts::status_by_ids(t, &plan.targets, plan.status, plan.reason.as_deref()).await?;
314    status_report(plan.status, outcome, Conflicts::Abort)
315}
316
317pub const QUERY_SAMPLE_SIZE: usize = 10;
318
319#[derive(Debug, Clone, PartialEq)]
320pub struct QueryStatusPlan {
321    pub status: AlertStatus,
322    pub reason: Option<String>,
323    pub conflicts: Conflicts,
324    pub query: Value,
325    pub matched: u64,
326    pub preview_action: String,
327    pub preview_details: Vec<String>,
328}
329
330/// Refuse an empty `{}` query: it would mutate every alert. Called from both
331/// `plan_status_by_query` and `apply_status_by_query` — `QueryStatusPlan`'s
332/// fields and `apply_status_by_query` are both public, so an `-api` consumer
333/// that builds a plan directly (skipping `plan_status_by_query`, as the
334/// planned MCP server might) must not be able to bypass the check.
335fn require_scoped_query(query: &Value) -> Result<()> {
336    let obj = query
337        .as_object()
338        .ok_or_else(|| Error::new(ErrorKind::Error, "--query must be a JSON object"))?;
339    if obj.is_empty() {
340        return Err(Error::new(
341            ErrorKind::Error,
342            "an empty --query would mutate every alert; say what it matches, \
343             e.g. an explicit {\"match_all\":{}}",
344        ));
345    }
346    Ok(())
347}
348
349/// Resolve the operator's query to a count and a sample so the implicit set
350/// is visible before it is mutated (triage spec section 6).
351pub async fn plan_status_by_query(
352    t: &Transport,
353    query: Value,
354    status: AlertStatus,
355    conflicts: Conflicts,
356    reason: Option<String>,
357) -> Result<QueryStatusPlan> {
358    require_scoped_query(&query)?;
359    let body = json!({
360        "query": &query,
361        "size": QUERY_SAMPLE_SIZE,
362        "track_total_hits": true,
363        "sort": default_sort(),
364        "_source": ["kibana.alert.rule.name", "kibana.alert.severity", "@timestamp"],
365    });
366    let page = alerts::search(t, &body).await?;
367    let matched = page.total.ok_or_else(|| {
368        Error::new(
369            ErrorKind::Http,
370            "decoding alerts response field `hits.total.value`",
371        )
372    })?;
373    let mut preview_details = vec![format!(
374        "matched now: {}   showing {} of {}",
375        thousands(matched),
376        page.hits.len(),
377        thousands(matched)
378    )];
379    for hit in &page.hits {
380        preview_details.push(format!(
381            "{}  {}  {}  {}",
382            hit.id,
383            source_str(&hit.source, "kibana.alert.rule.name").unwrap_or("(unnamed rule)"),
384            source_str(&hit.source, "kibana.alert.severity").unwrap_or("-"),
385            source_str(&hit.source, "@timestamp").unwrap_or("-"),
386        ));
387    }
388    preview_details.push("The set is resolved again at apply time; this count is advisory.".into());
389    Ok(QueryStatusPlan {
390        preview_action: format!("{} alerts matching query", status.verb()),
391        status,
392        reason,
393        conflicts,
394        query,
395        matched,
396        preview_details,
397    })
398}
399
400pub async fn apply_status_by_query(t: &Transport, plan: &QueryStatusPlan) -> Result<StatusReport> {
401    require_scoped_query(&plan.query)?;
402    let outcome = alerts::status_by_query(
403        t,
404        &plan.query,
405        plan.status,
406        plan.conflicts,
407        plan.reason.as_deref(),
408    )
409    .await?;
410    status_report(plan.status, outcome, plan.conflicts)
411}
412
413/// The tags/assignees report: the same counters without a target status.
414#[derive(Debug, Clone, PartialEq, serde::Serialize)]
415pub struct EditReport {
416    pub applied: bool,
417    pub total: u64,
418    pub updated: u64,
419    pub version_conflicts: u64,
420    pub noops: u64,
421    pub failed: u64,
422    pub failures: Vec<Value>,
423}
424
425fn edit_report(outcome: SignalsOutcome) -> Result<EditReport> {
426    Ok(EditReport {
427        applied: true,
428        total: outcome.total,
429        updated: outcome.updated,
430        version_conflicts: outcome.version_conflicts,
431        noops: outcome.noops,
432        failed: failed_count(&outcome, Conflicts::Abort)?,
433        failures: outcome.failures,
434    })
435}
436
437/// Refuse an empty target list before any resolution or request. The CLI's
438/// `required = true` covers the `alerts tag`/`assign` invocations, but
439/// `plan_tags` and `plan_assign` are public, so an `-api` consumer (the
440/// planned MCP server) calling them with `&[]` must not be able to reach a
441/// `{"ids": []}` POST that reports `applied: true, total: 0`.
442fn require_targets(ids: &[String]) -> Result<()> {
443    if ids.is_empty() {
444        return Err(Error::new(ErrorKind::Error, "pass at least one alert id"));
445    }
446    Ok(())
447}
448
449fn require_edit(add: &[String], remove: &[String]) -> Result<()> {
450    if add.is_empty() && remove.is_empty() {
451        return Err(Error::new(ErrorKind::Error, "pass --add and/or --remove"));
452    }
453    Ok(())
454}
455
456fn require_disjoint(add: &[String], remove: &[String], what: &str) -> Result<()> {
457    let overlap: Vec<&str> = add
458        .iter()
459        .filter(|a| remove.contains(a))
460        .map(String::as_str)
461        .collect();
462    if overlap.is_empty() {
463        Ok(())
464    } else {
465        Err(Error::new(
466            ErrorKind::Conflict,
467            format!("{what} both added and removed: {}", overlap.join(", ")),
468        ))
469    }
470}
471
472fn edit_summary(add: &[String], remove: &[String]) -> String {
473    let mut parts: Vec<String> = add.iter().map(|a| format!("+{a}")).collect();
474    parts.extend(remove.iter().map(|r| format!("-{r}")));
475    parts.join(" ")
476}
477
478#[derive(Debug, Clone, PartialEq)]
479pub struct TagsPlan {
480    pub targets: Vec<String>,
481    pub add: Vec<String>,
482    pub remove: Vec<String>,
483    pub preview_action: String,
484    pub preview_details: Vec<String>,
485}
486
487pub async fn plan_tags(
488    t: &Transport,
489    ids: &[String],
490    add: Vec<String>,
491    remove: Vec<String>,
492) -> Result<TagsPlan> {
493    Ok(plan_tags_observed(t, ids, add, remove).await?.0)
494}
495
496pub async fn apply_tags(t: &Transport, plan: &TagsPlan) -> Result<EditReport> {
497    require_targets(&plan.targets)?;
498    require_edit(&plan.add, &plan.remove)?;
499    require_disjoint(&plan.add, &plan.remove, "tags")?;
500    edit_report(alerts::set_tags(t, &plan.targets, &plan.add, &plan.remove).await?)
501}
502
503#[derive(Debug, Clone, PartialEq)]
504pub struct AssignPlan {
505    pub targets: Vec<String>,
506    /// Resolved profile uids.
507    pub add: Vec<String>,
508    pub remove: Vec<String>,
509    pub preview_action: String,
510    pub preview_details: Vec<String>,
511}
512
513pub async fn plan_assign(
514    t: &Transport,
515    ids: &[String],
516    add_users: &[String],
517    remove_users: &[String],
518) -> Result<AssignPlan> {
519    require_targets(ids)?;
520    require_edit(add_users, remove_users)?;
521    let mut add = Vec::with_capacity(add_users.len());
522    let mut remove = Vec::with_capacity(remove_users.len());
523    let mut mapping = Vec::new();
524    for user in add_users {
525        let uid = profiles::resolve_assignee(t, user).await?;
526        mapping.push(format!("add {user} -> {uid}"));
527        add.push(uid);
528    }
529    for user in remove_users {
530        let uid = profiles::resolve_assignee(t, user).await?;
531        mapping.push(format!("remove {user} -> {uid}"));
532        remove.push(uid);
533    }
534    require_disjoint(&add, &remove, "assignees")?;
535    let resolved = resolve_ids(t, ids).await?;
536    let summary = edit_summary(&add, &remove);
537    let mut preview_details = mapping;
538    preview_details.extend(
539        resolved
540            .iter()
541            .map(|r| format!("{}  {}  {}", r.id, r.rule_name, summary)),
542    );
543    Ok(AssignPlan {
544        preview_action: format!("Assign {} {}", resolved.len(), alert_noun(resolved.len())),
545        targets: resolved.into_iter().map(|r| r.id).collect(),
546        add,
547        remove,
548        preview_details,
549    })
550}
551
552pub async fn apply_assign(t: &Transport, plan: &AssignPlan) -> Result<EditReport> {
553    require_targets(&plan.targets)?;
554    require_edit(&plan.add, &plan.remove)?;
555    require_disjoint(&plan.add, &plan.remove, "assignees")?;
556    edit_report(alerts::set_assignees(t, &plan.targets, &plan.add, &plan.remove).await?)
557}
elasticctl_api/alerts_ops.rs

elasticctl_api/
alerts_ops.rs
