esdiag 0.16.4

Elastic Stack diagnostic collector and processor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License 2.0;
// you may not use this file except in compliance with the Elastic License 2.0.

use super::super::{DocumentExporter, ElasticsearchMetadata, Lookups, ProcessorSummary};
use super::SearchableSnapshotsStats;
use crate::exporter::Exporter;
use rayon::prelude::*;
use serde::Serialize;
use serde_json::{Value, value::RawValue};

impl DocumentExporter<Lookups, ElasticsearchMetadata> for SearchableSnapshotsStats {
    async fn documents_export(
        self,
        exporter: &Exporter,
        _lookups: &Lookups,
        metadata: &ElasticsearchMetadata,
    ) -> ProcessorSummary {
        let data_stream = "metrics-searchable_snapshot-esdiag".to_string();
        let searchable_snapshots_stats_metadata = metadata.for_data_stream(&data_stream).as_meta_doc();

        let mut indices: Vec<_> = self.indices.into_par_iter().collect();

        let searchable_snapshot_stats: Vec<Value> = indices
            .par_drain(..)
            .flat_map(|(index_name, mut index_stats)| {
                index_stats
                    .total
                    .par_drain(..)
                    .map(|index_stats| {
                        serde_json::to_value(SearchableSnapshotStatsDoc {
                            metadata: searchable_snapshots_stats_metadata.clone(),
                            index: IndexName {
                                name: index_name.clone(),
                            },
                            searchable_snapshot: index_stats,
                        })
                        .unwrap_or_default()
                    })
                    .collect::<Vec<Value>>()
            })
            .collect();

        tracing::debug!("searchable_snapshot_stats docs: {}", searchable_snapshot_stats.len());

        let mut summary = ProcessorSummary::new(data_stream.clone());
        match exporter.send(data_stream, searchable_snapshot_stats).await {
            Ok(batch) => summary.add_batch(batch),
            Err(err) => tracing::error!("Failed to send searchable snapshots stats: {}", err),
        }
        summary
    }
}

// Serializing data structures

#[derive(Clone, Serialize)]
pub struct SearchableSnapshotStatsDoc {
    #[serde(flatten)]
    metadata: Value,
    index: IndexName,
    searchable_snapshot: Box<RawValue>,
}

#[derive(Clone, Serialize)]
pub struct IndexName {
    pub name: String,
}