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    let execution = if let Some(summary) = rule
35        .as_map()
36        .get("execution_summary")
37        .filter(|summary| !summary.is_null())
38    {
39        let e = summary
40            .get("last_execution")
41            .filter(|v| v.is_object())
42            .ok_or_else(invalid)?;
43        let status = e
44            .get("status")
45            .and_then(Value::as_str)
46            .ok_or_else(invalid)?
47            .to_owned();
48        let date = e
49            .get("date")
50            .and_then(Value::as_str)
51            .ok_or_else(invalid)?
52            .to_owned();
53        let mut message: Option<String> = optional(e, "message")?;
54        let mut message_truncated = false;
55        if let Some(m) = &mut message
56            && m.len() > super::MESSAGE_BYTES
57        {
58            let mut end = super::MESSAGE_BYTES;
59            while !m.is_char_boundary(end) {
60                end -= 1;
61            }
62            m.truncate(end);
63            message_truncated = true;
64        }
65        let empty = serde_json::json!({});
66        let metrics = e.get("metrics").unwrap_or(&empty);
67        if !metrics.is_object() {
68            return Err(invalid());
69        }
70        Some(LastExecution {
71            status,
72            date,
73            status_order: optional(e, "status_order")?,
74            message,
75            message_truncated,
76            metrics: ExecutionMetrics {
77                total_search_duration_ms: optional(metrics, "total_search_duration_ms")?,
78                total_indexing_duration_ms: optional(metrics, "total_indexing_duration_ms")?,
79                total_enrichment_duration_ms: optional(metrics, "total_enrichment_duration_ms")?,
80                execution_gap_duration_s: optional(metrics, "execution_gap_duration_s")?,
81            },
82        })
83    } else {
84        None
85    };
86    Ok(RuleSnapshot {
87        rule: normalize::canonical(&rule),
88        execution,
89    })
90}
91pub async fn snapshot_by_id(t: &Transport, rule_id: &str) -> Result<RuleSnapshot> {
92    let rule = rules::get(t, rule_id).await?;
93    if rule.rule_id()? != rule_id {
94        return Err(Error::new(ErrorKind::NotFound, "rule identity mismatch"));
95    }
96    snapshot(rule)
97}
98pub async fn snapshot_by_selector(t: &Transport, selector: &str) -> Result<RuleSnapshot> {
99    let (_, rule) = selection::to_rule(t, selector).await?;
100    snapshot(rule)
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106    fn recorded() -> Rule {
107        let value: Value = serde_json::from_str(include_str!(concat!(
108            env!("CARGO_MANIFEST_DIR"),
109            "/../../tests/fixtures/traditional-9.5.1/rules_get.json"
110        )))
111        .unwrap();
112        Rule::from_value(value["response"].clone()).unwrap()
113    }
114    #[test]
115    fn execution_decoder_and_character_boundary() {
116        let mut rule = recorded();
117        rule.as_map_mut().remove("execution_summary");
118        assert!(snapshot(rule.clone()).unwrap().execution.is_none());
119        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}}}));
120        let execution = snapshot(rule.clone()).unwrap().execution.unwrap();
121        assert_eq!(execution.message.as_ref().unwrap().len(), 2046);
122        assert!(execution.message_truncated);
123        assert_eq!(execution.metrics.total_search_duration_ms, Some(3.0));
124        for field in ["status", "date"] {
125            let mut bad = rule.clone();
126            bad.as_map_mut()["execution_summary"]["last_execution"]
127                .as_object_mut()
128                .unwrap()
129                .remove(field);
130            assert_eq!(snapshot(bad).unwrap_err().kind, ErrorKind::Http);
131        }
132        for field in [
133            "total_search_duration_ms",
134            "total_indexing_duration_ms",
135            "total_enrichment_duration_ms",
136            "execution_gap_duration_s",
137        ] {
138            let mut bad = rule.clone();
139            bad.as_map_mut()["execution_summary"]["last_execution"]["metrics"][field] =
140                serde_json::json!("bad");
141            assert_eq!(snapshot(bad).unwrap_err().kind, ErrorKind::Http);
142        }
143    }
144}
elasticctl_api/compose/rule_snapshot.rs

elasticctl_api/compose/
rule_snapshot.rs
