elasticctl_api/compose/
readiness.rs

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

elasticctl_api/compose/
readiness.rs
