elasticctl_api/compose/
rule_snapshot.rs

1use crate::{Rule, normalize, rules, selection};
2use elasticctl_core::{Error, ErrorKind, Result, Transport};
3use serde_json::Value;
4#[derive(Debug)]
5pub struct ExecutionMetrics {
6    pub total_search_duration_ms: Option<f64>,
7    pub total_indexing_duration_ms: Option<f64>,
8    pub total_enrichment_duration_ms: Option<f64>,
9    pub execution_gap_duration_s: Option<f64>,
10}
11#[derive(Debug)]
12pub struct LastExecution {
13    pub status: String,
14    pub date: String,
15    pub status_order: Option<i64>,
16    pub message: Option<String>,
17    pub message_truncated: bool,
18    pub metrics: ExecutionMetrics,
19}
20#[derive(Debug)]
21pub struct RuleSnapshot {
22    pub rule: Rule,
23    pub execution: Option<LastExecution>,
24}
25fn invalid() -> Error {
26    Error::new(ErrorKind::Http, "invalid rule execution response")
27}
28fn optional<T: serde::de::DeserializeOwned>(v: &Value, k: &str) -> Result<Option<T>> {
29    v.get(k)
30        .map(|v| serde_json::from_value(v.clone()).map_err(|_| invalid()))
31        .transpose()
32}
33fn snapshot(rule: Rule) -> Result<RuleSnapshot> {
34    // An explicit null summary means the same as an absent one: never run.
35    let execution = if let Some(summary) = rule
36        .as_map()
37        .get("execution_summary")
38        .filter(|summary| !summary.is_null())
39    {
40        let e = summary
41            .get("last_execution")
42            .filter(|v| v.is_object())
43            .ok_or_else(invalid)?;
44        let status = e
45            .get("status")
46            .and_then(Value::as_str)
47            .ok_or_else(invalid)?
48            .to_owned();
49        let date = e
50            .get("date")
51            .and_then(Value::as_str)
52            .ok_or_else(invalid)?
53            .to_owned();
54        let mut message: Option<String> = optional(e, "message")?;
55        let mut message_truncated = false;
56        if let Some(m) = &mut message
57            && m.len() > super::MESSAGE_BYTES
58        {
59            let mut end = super::MESSAGE_BYTES;
60            while !m.is_char_boundary(end) {
61                end -= 1;
62            }
63            m.truncate(end);
64            message_truncated = true;
65        }
66        let empty = serde_json::json!({});
67        let metrics = e.get("metrics").unwrap_or(&empty);
68        if !metrics.is_object() {
69            return Err(invalid());
70        }
71        Some(LastExecution {
72            status,
73            date,
74            status_order: optional(e, "status_order")?,
75            message,
76            message_truncated,
77            metrics: ExecutionMetrics {
78                total_search_duration_ms: optional(metrics, "total_search_duration_ms")?,
79                total_indexing_duration_ms: optional(metrics, "total_indexing_duration_ms")?,
80                total_enrichment_duration_ms: optional(metrics, "total_enrichment_duration_ms")?,
81                execution_gap_duration_s: optional(metrics, "execution_gap_duration_s")?,
82            },
83        })
84    } else {
85        None
86    };
87    Ok(RuleSnapshot {
88        rule: normalize::canonical(&rule),
89        execution,
90    })
91}
92pub async fn snapshot_by_id(t: &Transport, rule_id: &str) -> Result<RuleSnapshot> {
93    let rule = rules::get(t, rule_id).await?;
94    if rule.rule_id()? != rule_id {
95        return Err(Error::new(ErrorKind::NotFound, "rule identity mismatch"));
96    }
97    snapshot(rule)
98}
99pub async fn snapshot_by_selector(t: &Transport, selector: &str) -> Result<RuleSnapshot> {
100    let (_, rule) = selection::to_rule(t, selector).await?;
101    snapshot(rule)
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    fn recorded() -> Rule {
108        let value: Value = serde_json::from_str(include_str!(concat!(
109            env!("CARGO_MANIFEST_DIR"),
110            "/../../tests/fixtures/traditional-9.5.1/rules_get.json"
111        )))
112        .unwrap();
113        Rule::from_value(value["response"].clone()).unwrap()
114    }
115    #[test]
116    fn execution_decoder_and_character_boundary() {
117        let mut rule = recorded();
118        rule.as_map_mut().remove("execution_summary");
119        assert!(snapshot(rule.clone()).unwrap().execution.is_none());
120        rule.as_map_mut().insert("execution_summary".into(),serde_json::json!({"last_execution":{"status":"failed","date":"today","message":"界".repeat(1000),"metrics":{"unknown":"ignored","total_search_duration_ms":3}}}));
121        let execution = snapshot(rule.clone()).unwrap().execution.unwrap();
122        assert_eq!(execution.message.as_ref().unwrap().len(), 2046);
123        assert!(execution.message_truncated);
124        assert_eq!(execution.metrics.total_search_duration_ms, Some(3.0));
125        for field in ["status", "date"] {
126            let mut bad = rule.clone();
127            bad.as_map_mut()["execution_summary"]["last_execution"]
128                .as_object_mut()
129                .unwrap()
130                .remove(field);
131            assert_eq!(snapshot(bad).unwrap_err().kind, ErrorKind::Http);
132        }
133        for field in [
134            "total_search_duration_ms",
135            "total_indexing_duration_ms",
136            "total_enrichment_duration_ms",
137            "execution_gap_duration_s",
138        ] {
139            let mut bad = rule.clone();
140            bad.as_map_mut()["execution_summary"]["last_execution"]["metrics"][field] =
141                serde_json::json!("bad");
142            assert_eq!(snapshot(bad).unwrap_err().kind, ErrorKind::Http);
143        }
144    }
145}
elasticctl_api/compose/rule_snapshot.rs

elasticctl_api/compose/
rule_snapshot.rs
