otel-arrow-dfe-engine 0.61.0

Async pipeline engine
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//! Node-scoped metrics for completion emission from effect handlers.
//!
//! These counters are recorded from the node point of view: they count
//! completions that a node successfully routes into the shared pipeline
//! completion lane through `notify_ack` / `notify_nack`.

use crate::entity_context::NodeTelemetryHandle;
use otel_arrow_dfe_config::MetricLevel;
use otel_arrow_dfe_telemetry::error::Error as TelemetryError;
use otel_arrow_dfe_telemetry::instrument::Counter;
use otel_arrow_dfe_telemetry::metrics::{MetricSet, MetricSetSnapshot};
use otel_arrow_dfe_telemetry::reporter::MetricsReporter;
use otel_arrow_dfe_telemetry_macros::metric_set;
use std::sync::{Arc, Mutex};

/// Node-scoped completion counters emitted from effect handlers.
#[metric_set(name = "node.completion_emission")]
#[derive(Debug, Default, Clone)]
pub(crate) struct CompletionEmissionMetrics {
    /// Count of successful `notify_ack` routings into the shared completion lane.
    #[metric(name = "notify_ack.routed", unit = "{message}")]
    pub notify_ack_routed: Counter<u64>,
    /// Count of successful `notify_nack` routings into the shared completion lane.
    #[metric(name = "notify_nack.routed", unit = "{message}")]
    pub notify_nack_routed: Counter<u64>,
}

pub(crate) struct CompletionEmissionMetricsState {
    metrics: MetricSet<CompletionEmissionMetrics>,
}

impl CompletionEmissionMetricsState {
    fn new(telemetry_handle: &NodeTelemetryHandle) -> Self {
        Self {
            metrics: telemetry_handle.register_metric_set::<CompletionEmissionMetrics>(),
        }
    }

    pub(crate) fn record_notify_ack_routed(&mut self) {
        self.metrics.notify_ack_routed.inc();
    }

    pub(crate) fn record_notify_nack_routed(&mut self) {
        self.metrics.notify_nack_routed.inc();
    }

    pub(crate) fn report(
        &mut self,
        metrics_reporter: &mut MetricsReporter,
    ) -> Result<(), TelemetryError> {
        metrics_reporter.report(&mut self.metrics)
    }

    pub(crate) fn snapshot(&self) -> Option<MetricSetSnapshot> {
        (!self.metrics.is_empty()).then(|| self.metrics.snapshot())
    }

    #[cfg(test)]
    pub(crate) fn counts(&self) -> (u64, u64) {
        (
            self.metrics.notify_ack_routed.get(),
            self.metrics.notify_nack_routed.get(),
        )
    }
}

pub(crate) type CompletionEmissionMetricsHandle = Arc<Mutex<CompletionEmissionMetricsState>>;

pub(crate) fn make_completion_emission_metrics(
    telemetry_handle: &Option<NodeTelemetryHandle>,
    level: MetricLevel,
) -> Option<CompletionEmissionMetricsHandle> {
    if level >= MetricLevel::Normal {
        telemetry_handle
            .as_ref()
            .map(|telemetry| Arc::new(Mutex::new(CompletionEmissionMetricsState::new(telemetry))))
    } else {
        None
    }
}