elasticctl_api/compose/
readiness.rs

1use super::{
2    MAX_PACKAGES, MAX_PATTERNS, Part, SkipReason, degrade, is_fatal, is_path_segment,
3    snapshot_by_selector,
4};
5use crate::{
6    Rule,
7    data_views::{self, DataViewSummary},
8    fleet::{
9        agent_policies, integration_policies::IntegrationPolicySummary, integration_policy_ops,
10    },
11};
12use elasticctl_core::{Error, ErrorKind, Result, Transport, urlencode};
13use serde_json::Value;
14use std::collections::HashSet;
15
16pub const PATTERN_BYTES: usize = 1_024;
17pub const PACKAGE_NAME_BYTES: usize = 128;
18
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum IndexSource {
21    Index,
22    DataView,
23    None,
24}
25
26#[derive(Debug)]
27pub struct RuleDataView {
28    pub id: String,
29    pub found: bool,
30    pub title: Option<String>,
31    pub name: Option<String>,
32}
33
34#[derive(Debug)]
35pub struct DataViewFacts {
36    pub count: usize,
37    pub rule_data_view: Option<RuleDataView>,
38}
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum PatternState {
42    Matches,
43    NoMatch,
44    Exclusion,
45    Unresolved,
46}
47
48#[derive(Debug)]
49pub struct PatternRow {
50    pub pattern: String,
51    pub state: PatternState,
52    pub counts: Option<(u64, u64, u64)>,
53    pub error: Option<Error>,
54    pub data_view_ids: Option<Vec<String>>,
55}
56
57#[derive(Debug)]
58pub struct PatternFacts {
59    pub total: usize,
60    pub truncated: bool,
61    pub rows: Vec<PatternRow>,
62}
63
64#[derive(Debug)]
65pub struct PackageUse {
66    pub install_status: String,
67    pub installed_version: Option<String>,
68    pub policies: u64,
69}
70
71#[derive(Debug)]
72pub struct PackageRow {
73    pub package: String,
74    pub version: Option<String>,
75    pub status: std::result::Result<PackageUse, Error>,
76}
77
78#[derive(Debug)]
79pub struct IntegrationFacts {
80    pub total: usize,
81    pub truncated: bool,
82    pub rows: Vec<PackageRow>,
83}
84
85#[derive(Debug)]
86pub struct Readiness {
87    pub rule: Rule,
88    pub index_source: IndexSource,
89    pub data_views: Part<DataViewFacts>,
90    pub patterns: Part<PatternFacts>,
91    pub integrations: Part<IntegrationFacts>,
92}
93
94#[derive(Clone, Copy, Debug, PartialEq, Eq)]
95pub enum ReadinessFinding {
96    RuleDisabled,
97    NoIndexSource,
98    DataViewMissing,
99    NoPatternMatches,
100    PatternUnresolved,
101    IntegrationNotInstalled,
102    IntegrationWithoutPolicy,
103    PatternsTruncated,
104    IntegrationsTruncated,
105}
106
107enum Source {
108    Index(Vec<String>),
109    DataView(String),
110    None,
111}
112
113pub async fn rule_readiness(t: &Transport, selector: &str) -> Result<Readiness> {
114    let rule = snapshot_by_selector(t, selector).await?.rule;
115    let source = source(&rule);
116    let index_source = match source {
117        Source::Index(_) => IndexSource::Index,
118        Source::DataView(_) => IndexSource::DataView,
119        Source::None => IndexSource::None,
120    };
121
122    let mut views = None;
123    let data_views = match &source {
124        Source::None => Part::Skipped(SkipReason::NotApplicable),
125        _ => match degrade(data_views::list(t).await)? {
126            Part::Ok(list) => {
127                let rule_data_view = match &source {
128                    Source::DataView(id) => Some(rule_data_view(&list, id)),
129                    _ => None,
130                };
131                let facts = DataViewFacts {
132                    count: list.len(),
133                    rule_data_view,
134                };
135                views = Some(list);
136                Part::Ok(facts)
137            }
138            Part::Unavailable(error) => Part::Unavailable(error),
139            Part::Skipped(reason) => Part::Skipped(reason),
140        },
141    };
142
143    let patterns = match source {
144        Source::None => Part::Skipped(SkipReason::NotApplicable),
145        _ if !t.has_es_url() => Part::Skipped(SkipReason::NotConfigured),
146        Source::Index(patterns) => Part::Ok(resolve_patterns(t, patterns, views.as_deref()).await?),
147        Source::DataView(_) => {
148            let title = match &data_views {
149                Part::Ok(DataViewFacts {
150                    rule_data_view:
151                        Some(RuleDataView {
152                            found: true,
153                            title: Some(title),
154                            ..
155                        }),
156                    ..
157                }) => Some(title.clone()),
158                _ => None,
159            };
160            match title {
161                Some(title) => {
162                    let patterns = title.split(',').map(str::to_owned).collect();
163                    Part::Ok(resolve_patterns(t, patterns, views.as_deref()).await?)
164                }
165                None => Part::Skipped(SkipReason::DependencyUnavailable),
166            }
167        }
168    };
169
170    let integrations = integrations(t, &rule).await?;
171    Ok(Readiness {
172        rule,
173        index_source,
174        data_views,
175        patterns,
176        integrations,
177    })
178}
179
180fn source(rule: &Rule) -> Source {
181    let map = rule.as_map();
182    if let Some(Value::Array(items)) = map.get("index")
183        && !items.is_empty()
184        && let Some(patterns) = items
185            .iter()
186            .map(|v| v.as_str().map(str::to_owned))
187            .collect::<Option<Vec<_>>>()
188    {
189        return Source::Index(patterns);
190    }
191    match map.get("data_view_id") {
192        Some(Value::String(id)) => Source::DataView(id.clone()),
193        _ => Source::None,
194    }
195}
196
197fn rule_data_view(list: &[DataViewSummary], id: &str) -> RuleDataView {
198    match list.iter().find(|view| view.id == id) {
199        Some(view) => RuleDataView {
200            id: id.to_owned(),
201            found: true,
202            title: Some(view.title.clone()),
203            name: view.name.clone(),
204        },
205        None => RuleDataView {
206            id: id.to_owned(),
207            found: false,
208            title: None,
209            name: None,
210        },
211    }
212}
213
214fn data_view_ids(views: &[DataViewSummary], pattern: &str) -> Vec<String> {
215    let mut ids = views
216        .iter()
217        .filter(|view| view.title.split(',').any(|part| part == pattern))
218        .map(|view| view.id.clone())
219        .collect::<Vec<_>>();
220    ids.sort();
221    ids.dedup();
222    ids
223}
224
225fn sendable_pattern(pattern: &str) -> bool {
226    !pattern.trim().is_empty()
227        && pattern.len() <= PATTERN_BYTES
228        && is_path_segment(pattern)
229        && !pattern.chars().any(char::is_control)
230}
231
232async fn resolve_patterns(
233    t: &Transport,
234    patterns: Vec<String>,
235    views: Option<&[DataViewSummary]>,
236) -> Result<PatternFacts> {
237    let total = patterns.len();
238    let mut rows = Vec::new();
239    for pattern in patterns.into_iter().take(MAX_PATTERNS) {
240        let (state, counts, error) = if pattern.starts_with('-') {
241            (PatternState::Exclusion, None, None)
242        } else if !sendable_pattern(&pattern) {
243            (PatternState::Unresolved, None, None)
244        } else {
245            match resolve(t, &pattern).await {
246                Ok(counts) => {
247                    let state = if counts.0 + counts.1 + counts.2 > 0 {
248                        PatternState::Matches
249                    } else {
250                        PatternState::NoMatch
251                    };
252                    (state, Some(counts), None)
253                }
254                Err(e) if is_fatal(e.kind) => return Err(e),
255                Err(e) if e.kind == ErrorKind::NotFound => {
256                    (PatternState::NoMatch, Some((0, 0, 0)), None)
257                }
258                Err(e) => (PatternState::Unresolved, None, Some(e)),
259            }
260        };
261        rows.push(PatternRow {
262            data_view_ids: views.map(|views| data_view_ids(views, &pattern)),
263            pattern,
264            state,
265            counts,
266            error,
267        });
268    }
269    Ok(PatternFacts {
270        total,
271        truncated: total > MAX_PATTERNS,
272        rows,
273    })
274}
275
276async fn resolve(t: &Transport, pattern: &str) -> Result<(u64, u64, u64)> {
277    let body = t
278        .get_absolute_es(&format!("/_resolve/index/{}", urlencode(pattern)))
279        .await?;
280    let count = |key: &str| {
281        body.get(key)
282            .and_then(Value::as_array)
283            .map(|items| items.len() as u64)
284            .ok_or_else(|| Error::new(ErrorKind::Http, "invalid resolve index response"))
285    };
286    Ok((count("indices")?, count("aliases")?, count("data_streams")?))
287}
288
289fn is_package_name(name: &str) -> bool {
290    (1..=PACKAGE_NAME_BYTES).contains(&name.len())
291        && name
292            .bytes()
293            .all(|b| matches!(b, b'a'..=b'z' | b'0'..=b'9' | b'_' | b'-'))
294}
295
296struct RelatedPackages {
297    total: usize,
298    entries: Vec<(String, Option<String>)>,
299}
300
301async fn related_packages(rule: &Rule) -> Result<RelatedPackages> {
302    let invalid = || Error::new(ErrorKind::Http, "invalid related integrations on the rule");
303    let entries = match rule.as_map().get("related_integrations") {
304        None | Some(Value::Null) => {
305            return Ok(RelatedPackages {
306                total: 0,
307                entries: Vec::new(),
308            });
309        }
310        Some(Value::Array(entries)) => entries,
311        Some(_) => return Err(invalid()),
312    };
313    let mut packages: Vec<(String, Option<String>)> = Vec::new();
314    let mut seen = HashSet::new();
315    for (index, entry) in entries.iter().enumerate() {
316        let package = entry
317            .get("package")
318            .and_then(Value::as_str)
319            .ok_or_else(invalid)?;
320        let version = match entry.get("version") {
321            None | Some(Value::Null) => None,
322            Some(Value::String(version)) => Some(version.clone()),
323            Some(_) => return Err(invalid()),
324        };
325        if seen.insert(package) && packages.len() < MAX_PACKAGES {
326            packages.push((package.to_owned(), version));
327        }
328        if (index + 1) % 1_024 == 0 {
329            tokio::task::yield_now().await;
330        }
331    }
332    Ok(RelatedPackages {
333        total: seen.len(),
334        entries: packages,
335    })
336}
337
338async fn integrations(t: &Transport, rule: &Rule) -> Result<Part<IntegrationFacts>> {
339    let packages = match related_packages(rule).await {
340        Ok(packages) => packages,
341        Err(error) => return Ok(Part::Unavailable(error)),
342    };
343    let total = packages.total;
344    if total == 0 {
345        return Ok(Part::Ok(IntegrationFacts {
346            total,
347            truncated: false,
348            rows: Vec::new(),
349        }));
350    }
351    let policies = match integration_policy_ops::list_first_page(t).await {
352        Ok(Some(policies)) => policies,
353        Ok(None) => {
354            return Ok(Part::Unavailable(Error::new(
355                ErrorKind::Unsupported,
356                "more integration policies than one bounded page",
357            )));
358        }
359        Err(e) if is_fatal(e.kind) => return Err(e),
360        Err(e) => return Ok(Part::Unavailable(e)),
361    };
362    let mut rows = Vec::new();
363    for (package, version) in packages.entries {
364        let status = package_use(t, &package, &policies).await?;
365        rows.push(PackageRow {
366            package,
367            version,
368            status,
369        });
370    }
371    Ok(Part::Ok(IntegrationFacts {
372        total,
373        truncated: total > MAX_PACKAGES,
374        rows,
375    }))
376}
377
378async fn package_use(
379    t: &Transport,
380    package: &str,
381    policies: &[IntegrationPolicySummary],
382) -> Result<std::result::Result<PackageUse, Error>> {
383    if !is_package_name(package) {
384        return Ok(Err(Error::new(
385            ErrorKind::Unsupported,
386            "package name outside the accepted grammar",
387        )));
388    }
389    match agent_policies::package_status(t, package).await {
390        Ok(status) => Ok(Ok(PackageUse {
391            install_status: status.status,
392            installed_version: status.installed_version,
393            policies: policies
394                .iter()
395                .filter(|policy| policy.package.name == package)
396                .count() as u64,
397        })),
398        Err(e) if is_fatal(e.kind) => Err(e),
399        Err(e) => Ok(Err(e)),
400    }
401}
402
403impl Readiness {
404    pub fn findings(&self) -> Vec<ReadinessFinding> {
405        use ReadinessFinding::*;
406        let mut findings = Vec::new();
407        if self.rule.as_map().get("enabled") == Some(&Value::Bool(false)) {
408            findings.push(RuleDisabled);
409        }
410        if self.index_source == IndexSource::None {
411            findings.push(NoIndexSource);
412        }
413        if let Part::Ok(DataViewFacts {
414            rule_data_view: Some(view),
415            ..
416        }) = &self.data_views
417            && self.index_source == IndexSource::DataView
418            && !view.found
419        {
420            findings.push(DataViewMissing);
421        }
422        let patterns = match &self.patterns {
423            Part::Ok(facts) => Some(facts),
424            _ => None,
425        };
426        if let Some(facts) = patterns {
427            let mut checked = facts
428                .rows
429                .iter()
430                .filter(|row| row.state != PatternState::Exclusion)
431                .peekable();
432            if !facts.truncated
433                && checked.peek().is_some()
434                && checked.all(|row| row.state == PatternState::NoMatch)
435            {
436                findings.push(NoPatternMatches);
437            }
438            if facts
439                .rows
440                .iter()
441                .any(|row| row.state == PatternState::Unresolved)
442            {
443                findings.push(PatternUnresolved);
444            }
445        }
446        let integrations = match &self.integrations {
447            Part::Ok(facts) => Some(facts),
448            _ => None,
449        };
450        let used = || {
451            integrations
452                .into_iter()
453                .flat_map(|facts| &facts.rows)
454                .filter_map(|row| row.status.as_ref().ok())
455        };
456        if used().any(|u| u.install_status != "installed") {
457            findings.push(IntegrationNotInstalled);
458        }
459        if used().any(|u| u.install_status == "installed" && u.policies == 0) {
460            findings.push(IntegrationWithoutPolicy);
461        }
462        if patterns.is_some_and(|facts| facts.truncated) {
463            findings.push(PatternsTruncated);
464        }
465        if integrations.is_some_and(|facts| facts.truncated) {
466            findings.push(IntegrationsTruncated);
467        }
468        findings
469    }
470
471    pub fn partial(&self) -> bool {
472        let patterns = match &self.patterns {
473            Part::Ok(facts) => facts
474                .rows
475                .iter()
476                .any(|row| row.state == PatternState::Unresolved),
477            part => matches!(part, Part::Unavailable(_)),
478        };
479        let integrations = match &self.integrations {
480            Part::Ok(facts) => facts.rows.iter().any(|row| row.status.is_err()),
481            part => matches!(part, Part::Unavailable(_)),
482        };
483        matches!(self.data_views, Part::Unavailable(_)) || patterns || integrations
484    }
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490
491    #[test]
492    fn pattern_and_package_grammars() {
493        for pattern in ["logs-*", "...", ".a", "a..b", "a/b", "%2e%2e", "../_bulk"] {
494            assert!(sendable_pattern(pattern), "{pattern}");
495        }
496        let long = "a".repeat(PATTERN_BYTES + 1);
497        for pattern in [
498            "",
499            " ",
500            "\t",
501            ".",
502            "..",
503            "a\u{7}b",
504            "a\u{85}b",
505            long.as_str(),
506        ] {
507            assert!(!sendable_pattern(pattern), "{pattern:?}");
508        }
509        assert!(sendable_pattern(&"a".repeat(PATTERN_BYTES)));
510        for name in ["system", "a", "endpoint_2-x", &"a".repeat(128)] {
511            assert!(is_package_name(name), "{name}");
512        }
513        let long = "a".repeat(129);
514        for name in [
515            "",
516            ".",
517            "..",
518            "../x",
519            "A",
520            "a.b",
521            "a/b",
522            "a b",
523            "é",
524            long.as_str(),
525        ] {
526            assert!(!is_package_name(name), "{name}");
527        }
528    }
529}
elasticctl_api/compose/readiness.rs

elasticctl_api/compose/
readiness.rs
