elasticctl_api/
alerts_ops.rs

1mod recheck;
2pub use recheck::{AlertFacet, observe_alerts};
3mod observed;
4pub use observed::{plan_status_by_ids_observed, plan_tags_observed};
5
6use crate::alerts::{self, AlertHit, AlertStatus, Conflicts, SignalsOutcome};
7use crate::{profiles, selection};
8use elasticctl_core::{Error, ErrorKind, Result, Transport};
9use serde_json::{Value, json};
10
11#[derive(Debug, Clone, Default)]
12pub struct AlertFilter {
13    pub status: Option<AlertStatus>,
14    pub severity: Option<String>,
15    pub rule: Option<String>,
16    pub tag: Option<String>,
17    pub assignee: Option<String>,
18    pub since: Option<String>,
19    pub search: Option<String>,
20}
21
22pub fn since_clause(since: &str) -> Value {
23    let bytes = since.as_bytes();
24    let is_duration = bytes.len() >= 2
25        && bytes[..bytes.len() - 1].iter().all(u8::is_ascii_digit)
26        && matches!(bytes[bytes.len() - 1], b's' | b'm' | b'h' | b'd' | b'w');
27    let gte = if is_duration {
28        format!("now-{since}")
29    } else {
30        since.to_string()
31    };
32    json!({"range": {"@timestamp": {"gte": gte}}})
33}
34
35pub fn default_sort() -> Value {
36    json!([
37        {"@timestamp": {"order": "desc"}},
38        {"kibana.alert.uuid": {"order": "asc"}}
39    ])
40}
41
42fn escape_wildcard(value: &str) -> String {
43    value
44        .replace('\\', "\\\\")
45        .replace('*', "\\*")
46        .replace('?', "\\?")
47}
48
49pub async fn build_query(t: &Transport, f: &AlertFilter) -> Result<Value> {
50    let mut filter = Vec::new();
51    if let Some(status) = f.status {
52        filter.push(json!({"term": {"kibana.alert.workflow_status": status.as_str()}}));
53    }
54    if let Some(severity) = &f.severity {
55        filter.push(json!({"term": {"kibana.alert.severity": severity}}));
56    }
57    if let Some(rule) = &f.rule {
58        let rule_id = selection::to_rule_id(t, rule).await?;
59        filter.push(json!({"term": {"kibana.alert.rule.rule_id": rule_id}}));
60    }
61    if let Some(tag) = &f.tag {
62        filter.push(json!({"term": {"kibana.alert.workflow_tags": tag}}));
63    }
64    if let Some(assignee) = &f.assignee {
65        let uid = profiles::resolve_assignee(t, assignee).await?;
66        filter.push(json!({"term": {"kibana.alert.workflow_assignee_ids": uid}}));
67    }
68    if let Some(since) = &f.since {
69        filter.push(since_clause(since));
70    }
71    if let Some(text) = &f.search {
72        let pattern = format!("*{}*", escape_wildcard(text));
73        filter.push(json!({"bool": {"minimum_should_match": 1, "should": [
74            {"wildcard": {"kibana.alert.rule.name": {"value": pattern, "case_insensitive": true}}},
75            {"wildcard": {"kibana.alert.reason": {"value": pattern, "case_insensitive": true}}}
76        ]}}));
77    }
78    if filter.is_empty() {
79        Ok(json!({"match_all": {}}))
80    } else {
81        Ok(json!({"bool": {"filter": filter}}))
82    }
83}
84
85#[derive(Debug, Clone, PartialEq)]
86pub struct AlertList {
87    pub hits: Vec<AlertHit>,
88    pub total: Option<u64>,
89    pub truncated: bool,
90}
91
92pub async fn list(t: &Transport, f: &AlertFilter, limit: usize) -> Result<AlertList> {
93    let query = build_query(t, f).await?;
94    let body = json!({
95        "query": query,
96        "sort": default_sort(),
97        "size": limit.saturating_add(1),
98        "track_total_hits": true,
99    });
100    let mut page = alerts::search(t, &body).await?;
101    let truncated = page.hits.len() > limit;
102    page.hits.truncate(limit);
103    Ok(AlertList {
104        hits: page.hits,
105        total: page.total,
106        truncated,
107    })
108}
109
110pub async fn export(t: &Transport, f: &AlertFilter, limit: Option<usize>) -> Result<Vec<AlertHit>> {
111    let query = build_query(t, f).await?;
112    alerts::search_all(t, &query, &default_sort(), limit).await
113}
114
115pub async fn get_one(t: &Transport, alert_id: &str) -> Result<AlertHit> {
116    let body = json!({"query": {"ids": {"values": [alert_id]}}, "size": 1});
117    let page = alerts::search(t, &body).await?;
118    page.hits.into_iter().next().ok_or_else(|| {
119        Error::new(
120            ErrorKind::NotFound,
121            format!("No alert with id '{alert_id}'"),
122        )
123    })
124}
125
126#[derive(Debug, Clone, PartialEq, Eq)]
127pub struct ResolvedAlert {
128    pub id: String,
129    pub rule_name: String,
130    pub status: String,
131}
132
133pub const RESOLVE_SOURCE_FIELDS: &[&str] =
134    &["kibana.alert.rule.name", "kibana.alert.workflow_status"];
135
136pub(crate) fn source_str<'a>(source: &'a Value, key: &str) -> Option<&'a str> {
137    if let Some(v) = source.get(key).and_then(Value::as_str) {
138        return Some(v);
139    }
140    let pointer = format!("/{}", key.replace('.', "/"));
141    source.pointer(&pointer).and_then(Value::as_str)
142}
143
144fn workflow_status(source: &Value) -> &str {
145    source_str(source, "kibana.alert.workflow_status").unwrap_or("unknown")
146}
147
148fn resolve_body(unique: &[String]) -> Value {
149    json!({
150        "query": {"ids": {"values": unique}},
151        "size": unique.len(),
152        "_source": RESOLVE_SOURCE_FIELDS,
153    })
154}
155
156async fn resolve_ids(t: &Transport, ids: &[String]) -> Result<Vec<ResolvedAlert>> {
157    let mut unique: Vec<String> = Vec::with_capacity(ids.len());
158    for id in ids {
159        if !unique.contains(id) {
160            unique.push(id.clone());
161        }
162    }
163    let body = resolve_body(&unique);
164    let page = alerts::search(t, &body).await?;
165    let mut resolved = Vec::with_capacity(unique.len());
166    for id in &unique {
167        let hit = page.hits.iter().find(|h| &h.id == id).ok_or_else(|| {
168            let missing: Vec<&str> = unique
169                .iter()
170                .filter(|i| !page.hits.iter().any(|h| &h.id == *i))
171                .map(String::as_str)
172                .collect();
173            Error::new(
174                ErrorKind::NotFound,
175                format!("No alert with id: {}", missing.join(", ")),
176            )
177        })?;
178        resolved.push(ResolvedAlert {
179            id: id.clone(),
180            rule_name: source_str(&hit.source, "kibana.alert.rule.name")
181                .unwrap_or("(unnamed rule)")
182                .to_string(),
183            status: workflow_status(&hit.source).to_string(),
184        });
185    }
186    Ok(resolved)
187}
188
189fn alert_noun(n: usize) -> &'static str {
190    if n == 1 { "alert" } else { "alerts" }
191}
192
193fn thousands(n: u64) -> String {
194    let digits = n.to_string();
195    let mut out = String::with_capacity(digits.len() + digits.len() / 3);
196    for (i, c) in digits.chars().enumerate() {
197        if i > 0 && (digits.len() - i).is_multiple_of(3) {
198            out.push(',');
199        }
200        out.push(c);
201    }
202    out
203}
204
205#[derive(Debug, Clone, PartialEq)]
206pub struct StatusPlan {
207    pub status: AlertStatus,
208    pub reason: Option<String>,
209    pub targets: Vec<String>,
210    pub preview_action: String,
211    pub preview_details: Vec<String>,
212}
213
214pub async fn plan_status_by_ids(
215    t: &Transport,
216    ids: &[String],
217    status: AlertStatus,
218    reason: Option<String>,
219) -> Result<StatusPlan> {
220    Ok(plan_status_by_ids_observed(t, ids, status, reason).await?.0)
221}
222
223#[derive(Debug, Clone, PartialEq, serde::Serialize)]
224pub struct StatusReport {
225    pub applied: bool,
226    pub status: String,
227    pub total: u64,
228    pub updated: u64,
229    pub version_conflicts: u64,
230    pub noops: u64,
231    pub failed: u64,
232    pub failures: Vec<Value>,
233}
234
235fn failed_count(outcome: &SignalsOutcome, conflicts: Conflicts) -> Result<u64> {
236    (outcome.failures.len() as u64)
237        .checked_add(match conflicts {
238            Conflicts::Abort => outcome.version_conflicts,
239            Conflicts::Proceed => 0,
240        })
241        .ok_or_else(|| {
242            Error::new(
243                ErrorKind::Http,
244                "decoding alert response: counters are unreadable (failure count overflows u64)",
245            )
246        })
247}
248
249fn status_report(
250    status: AlertStatus,
251    outcome: SignalsOutcome,
252    conflicts: Conflicts,
253) -> Result<StatusReport> {
254    Ok(StatusReport {
255        applied: true,
256        status: status.as_str().to_string(),
257        total: outcome.total,
258        updated: outcome.updated,
259        version_conflicts: outcome.version_conflicts,
260        noops: outcome.noops,
261        failed: failed_count(&outcome, conflicts)?,
262        failures: outcome.failures,
263    })
264}
265
266pub async fn apply_status_by_ids(t: &Transport, plan: &StatusPlan) -> Result<StatusReport> {
267    require_targets(&plan.targets)?;
268    let outcome =
269        alerts::status_by_ids(t, &plan.targets, plan.status, plan.reason.as_deref()).await?;
270    status_report(plan.status, outcome, Conflicts::Abort)
271}
272
273pub const QUERY_SAMPLE_SIZE: usize = 10;
274
275#[derive(Debug, Clone, PartialEq)]
276pub struct QueryStatusPlan {
277    pub status: AlertStatus,
278    pub reason: Option<String>,
279    pub conflicts: Conflicts,
280    pub query: Value,
281    pub matched: u64,
282    pub preview_action: String,
283    pub preview_details: Vec<String>,
284}
285
286fn require_scoped_query(query: &Value) -> Result<()> {
287    let obj = query
288        .as_object()
289        .ok_or_else(|| Error::new(ErrorKind::Error, "--query must be a JSON object"))?;
290    if obj.is_empty() {
291        return Err(Error::new(
292            ErrorKind::Error,
293            "an empty --query would mutate every alert; say what it matches, \
294             e.g. an explicit {\"match_all\":{}}",
295        ));
296    }
297    Ok(())
298}
299
300pub async fn plan_status_by_query(
301    t: &Transport,
302    query: Value,
303    status: AlertStatus,
304    conflicts: Conflicts,
305    reason: Option<String>,
306) -> Result<QueryStatusPlan> {
307    require_scoped_query(&query)?;
308    let body = json!({
309        "query": &query,
310        "size": QUERY_SAMPLE_SIZE,
311        "track_total_hits": true,
312        "sort": default_sort(),
313        "_source": ["kibana.alert.rule.name", "kibana.alert.severity", "@timestamp"],
314    });
315    let page = alerts::search(t, &body).await?;
316    let matched = page.total.ok_or_else(|| {
317        Error::new(
318            ErrorKind::Http,
319            "decoding alerts response field `hits.total.value`",
320        )
321    })?;
322    let mut preview_details = vec![format!(
323        "matched now: {}   showing {} of {}",
324        thousands(matched),
325        page.hits.len(),
326        thousands(matched)
327    )];
328    for hit in &page.hits {
329        preview_details.push(format!(
330            "{}  {}  {}  {}",
331            hit.id,
332            source_str(&hit.source, "kibana.alert.rule.name").unwrap_or("(unnamed rule)"),
333            source_str(&hit.source, "kibana.alert.severity").unwrap_or("-"),
334            source_str(&hit.source, "@timestamp").unwrap_or("-"),
335        ));
336    }
337    preview_details.push("The set is resolved again at apply time; this count is advisory.".into());
338    Ok(QueryStatusPlan {
339        preview_action: format!("{} alerts matching query", status.verb()),
340        status,
341        reason,
342        conflicts,
343        query,
344        matched,
345        preview_details,
346    })
347}
348
349pub async fn apply_status_by_query(t: &Transport, plan: &QueryStatusPlan) -> Result<StatusReport> {
350    require_scoped_query(&plan.query)?;
351    let outcome = alerts::status_by_query(
352        t,
353        &plan.query,
354        plan.status,
355        plan.conflicts,
356        plan.reason.as_deref(),
357    )
358    .await?;
359    status_report(plan.status, outcome, plan.conflicts)
360}
361
362#[derive(Debug, Clone, PartialEq, serde::Serialize)]
363pub struct EditReport {
364    pub applied: bool,
365    pub total: u64,
366    pub updated: u64,
367    pub version_conflicts: u64,
368    pub noops: u64,
369    pub failed: u64,
370    pub failures: Vec<Value>,
371}
372
373fn edit_report(outcome: SignalsOutcome) -> Result<EditReport> {
374    Ok(EditReport {
375        applied: true,
376        total: outcome.total,
377        updated: outcome.updated,
378        version_conflicts: outcome.version_conflicts,
379        noops: outcome.noops,
380        failed: failed_count(&outcome, Conflicts::Abort)?,
381        failures: outcome.failures,
382    })
383}
384
385fn require_targets(ids: &[String]) -> Result<()> {
386    if ids.is_empty() {
387        return Err(Error::new(ErrorKind::Error, "pass at least one alert id"));
388    }
389    Ok(())
390}
391
392fn require_edit(add: &[String], remove: &[String]) -> Result<()> {
393    if add.is_empty() && remove.is_empty() {
394        return Err(Error::new(ErrorKind::Error, "pass --add and/or --remove"));
395    }
396    Ok(())
397}
398
399fn require_disjoint(add: &[String], remove: &[String], what: &str) -> Result<()> {
400    let overlap: Vec<&str> = add
401        .iter()
402        .filter(|a| remove.contains(a))
403        .map(String::as_str)
404        .collect();
405    if overlap.is_empty() {
406        Ok(())
407    } else {
408        Err(Error::new(
409            ErrorKind::Conflict,
410            format!("{what} both added and removed: {}", overlap.join(", ")),
411        ))
412    }
413}
414
415fn edit_summary(add: &[String], remove: &[String]) -> String {
416    let mut parts: Vec<String> = add.iter().map(|a| format!("+{a}")).collect();
417    parts.extend(remove.iter().map(|r| format!("-{r}")));
418    parts.join(" ")
419}
420
421#[derive(Debug, Clone, PartialEq)]
422pub struct TagsPlan {
423    pub targets: Vec<String>,
424    pub add: Vec<String>,
425    pub remove: Vec<String>,
426    pub preview_action: String,
427    pub preview_details: Vec<String>,
428}
429
430pub async fn plan_tags(
431    t: &Transport,
432    ids: &[String],
433    add: Vec<String>,
434    remove: Vec<String>,
435) -> Result<TagsPlan> {
436    Ok(plan_tags_observed(t, ids, add, remove).await?.0)
437}
438
439pub async fn apply_tags(t: &Transport, plan: &TagsPlan) -> Result<EditReport> {
440    require_targets(&plan.targets)?;
441    require_edit(&plan.add, &plan.remove)?;
442    require_disjoint(&plan.add, &plan.remove, "tags")?;
443    edit_report(alerts::set_tags(t, &plan.targets, &plan.add, &plan.remove).await?)
444}
445
446#[derive(Debug, Clone, PartialEq)]
447pub struct AssignPlan {
448    pub targets: Vec<String>,
449    pub add: Vec<String>,
450    pub remove: Vec<String>,
451    pub preview_action: String,
452    pub preview_details: Vec<String>,
453}
454
455pub async fn plan_assign(
456    t: &Transport,
457    ids: &[String],
458    add_users: &[String],
459    remove_users: &[String],
460) -> Result<AssignPlan> {
461    require_targets(ids)?;
462    require_edit(add_users, remove_users)?;
463    let mut add = Vec::with_capacity(add_users.len());
464    let mut remove = Vec::with_capacity(remove_users.len());
465    let mut mapping = Vec::new();
466    for user in add_users {
467        let uid = profiles::resolve_assignee(t, user).await?;
468        mapping.push(format!("add {user} -> {uid}"));
469        add.push(uid);
470    }
471    for user in remove_users {
472        let uid = profiles::resolve_assignee(t, user).await?;
473        mapping.push(format!("remove {user} -> {uid}"));
474        remove.push(uid);
475    }
476    require_disjoint(&add, &remove, "assignees")?;
477    let resolved = resolve_ids(t, ids).await?;
478    let summary = edit_summary(&add, &remove);
479    let mut preview_details = mapping;
480    preview_details.extend(
481        resolved
482            .iter()
483            .map(|r| format!("{}  {}  {}", r.id, r.rule_name, summary)),
484    );
485    Ok(AssignPlan {
486        preview_action: format!("Assign {} {}", resolved.len(), alert_noun(resolved.len())),
487        targets: resolved.into_iter().map(|r| r.id).collect(),
488        add,
489        remove,
490        preview_details,
491    })
492}
493
494pub async fn apply_assign(t: &Transport, plan: &AssignPlan) -> Result<EditReport> {
495    require_targets(&plan.targets)?;
496    require_edit(&plan.add, &plan.remove)?;
497    require_disjoint(&plan.add, &plan.remove, "assignees")?;
498    edit_report(alerts::set_assignees(t, &plan.targets, &plan.add, &plan.remove).await?)
499}
elasticctl_api/alerts_ops.rs

elasticctl_api/
alerts_ops.rs
