elasticctl_api/
alerts_ops.rs

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

elasticctl_api/
alerts_ops.rs
