esdiag 0.16.4

// 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::{PendingTask, PendingTasks};
use crate::exporter::Exporter;
use rayon::prelude::*;
use serde::Serialize;
use serde_json::Value;

impl DocumentExporter<Lookups, ElasticsearchMetadata> for PendingTasks {
    async fn documents_export(
        self,
        exporter: &Exporter,
        _lookups: &Lookups,
        metadata: &ElasticsearchMetadata,
    ) -> ProcessorSummary {
        tracing::debug!("processing pending tasks");
        let data_stream = "metrics-task.pending-esdiag".to_string();
        let metadata = metadata.for_data_stream(&data_stream).as_meta_doc();

        let mut pending_tasks: Vec<PendingTask> = self.tasks.into_par_iter().collect();

        let pending_tasks: Vec<Value> = pending_tasks
            .par_drain(..)
            .filter_map(|task| {
                serde_json::to_value(PendingTaskDoc {
                    task,
                    metadata: metadata.clone(),
                })
                .ok()
            })
            .collect();

        tracing::debug!("pending task docs: {}", pending_tasks.len());

        let mut summary = match pending_tasks.is_empty() {
            false => ProcessorSummary::new(data_stream.clone()),
            true => return ProcessorSummary::new(data_stream.clone()),
        };
        match exporter.send(data_stream, pending_tasks).await {
            Ok(batch) => summary.add_batch(batch),
            Err(err) => tracing::error!("Failed to send pending tasks: {}", err),
        }
        summary
    }
}

#[derive(Serialize)]
struct PendingTaskDoc {
    task: PendingTask,
    #[serde(flatten)]
    metadata: Value,
}