saluki_components/destinations/
dogstatsd_client_telemetry.rs

1use async_trait::async_trait;
2use saluki_core::{
3    accounting::{MemoryBounds, MemoryBoundsBuilder},
4    components::{destinations::*, BuildContext, ComponentContext},
5    data_model::event::{
6        metric::{Metric, MetricValues},
7        Event, EventType,
8    },
9};
10use saluki_error::GenericError;
11use saluki_metrics::{static_metrics, Counter};
12use tokio::select;
13use tracing::debug;
14
15#[derive(Clone, Copy)]
16struct ClientTelemetryTags {
17    client: &'static str,
18    client_transport: &'static str,
19}
20
21impl ClientTelemetryTags {
22    fn from_metric(metric: &Metric) -> Self {
23        let mut tags = Self {
24            client: "unknown",
25            client_transport: "unknown",
26        };
27        for tag in metric.context().tags() {
28            match tag.name() {
29                "client" => tags.client = normalize_client_library(tag.value()),
30                "client_transport" => tags.client_transport = normalize_client_transport(tag.value()),
31                _ => {}
32            }
33            if tags.client != "unknown" && tags.client_transport != "unknown" {
34                break;
35            }
36        }
37        tags
38    }
39}
40
41pub(super) fn normalize_client_library(client: Option<&str>) -> &'static str {
42    match client {
43        Some("go") => "go",
44        Some("py") => "py",
45        Some("java") => "java",
46        Some("ruby") => "ruby",
47        Some("csharp") => "csharp",
48        Some("php") => "php",
49        Some("rust") => "rust",
50        _ => "unknown",
51    }
52}
53
54pub(super) fn normalize_client_transport(transport: Option<&str>) -> &'static str {
55    match transport {
56        Some("udp") => "udp",
57        Some("uds") => "uds",
58        Some("uds-stream") => "uds-stream",
59        Some("uds-datagram") => "uds-datagram",
60        Some("pipe") => "pipe",
61        Some("namedpipe") => "namedpipe",
62        Some("named_pipe") => "named_pipe",
63        Some("custom") => "custom",
64        Some("http") => "http",
65        _ => "unknown",
66    }
67}
68
69#[static_metrics(prefix = dogstatsd_client_telemetry, labels(component_id, component_type))]
70#[derive(Clone)]
71struct ClientTelemetryCounters {
72    #[metric(mapped(client, client_transport))]
73    bytes_sent: Counter,
74    #[metric(mapped(client, client_transport))]
75    bytes_dropped: Counter,
76    #[metric(mapped(client, client_transport))]
77    bytes_dropped_queue: Counter,
78    #[metric(mapped(client, client_transport))]
79    bytes_dropped_writer: Counter,
80}
81
82enum ClientTelemetryMetric {
83    Sent,
84    Dropped,
85    DroppedQueue,
86    DroppedWriter,
87}
88
89/// Configuration for the DogStatsD client telemetry destination.
90#[derive(Default)]
91pub struct DogStatsDClientTelemetryConfiguration;
92
93#[async_trait]
94impl DestinationBuilder for DogStatsDClientTelemetryConfiguration {
95    fn input_event_type(&self) -> EventType {
96        EventType::Metric
97    }
98
99    async fn build(&self, context: BuildContext) -> Result<Box<dyn Destination + Send>, GenericError> {
100        Ok(Box::new(DogStatsDClientTelemetry::new(context.component_context())))
101    }
102}
103
104impl MemoryBounds for DogStatsDClientTelemetryConfiguration {
105    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
106        builder
107            .minimum()
108            .with_single_value::<DogStatsDClientTelemetry>("destination struct");
109    }
110}
111
112/// Mirrors supported DogStatsD client telemetry metrics into ADP internal telemetry.
113pub struct DogStatsDClientTelemetry {
114    counters: ClientTelemetryCounters,
115}
116
117impl DogStatsDClientTelemetry {
118    pub(super) fn new(component_context: &ComponentContext) -> Self {
119        Self {
120            counters: ClientTelemetryCounters::new(
121                component_context.component_id(),
122                component_context.component_type().as_str(),
123            ),
124        }
125    }
126
127    pub(super) fn record_metric(&self, metric: &Metric) {
128        let metric_kind = match metric.context().name().as_ref() {
129            "datadog.dogstatsd.client.bytes_sent" => ClientTelemetryMetric::Sent,
130            "datadog.dogstatsd.client.bytes_dropped" => ClientTelemetryMetric::Dropped,
131            "datadog.dogstatsd.client.bytes_dropped_queue" => ClientTelemetryMetric::DroppedQueue,
132            "datadog.dogstatsd.client.bytes_dropped_writer" => ClientTelemetryMetric::DroppedWriter,
133            _ => return,
134        };
135
136        if let MetricValues::Rate(values, _) = metric.values() {
137            // A delayed aggregate flush can contain several closed time buckets in one metric. Separate tag contexts
138            // arrive as separate metrics and accumulate into counters with their client dimensions preserved.
139            let tags = ClientTelemetryTags::from_metric(metric);
140            let counter = match metric_kind {
141                ClientTelemetryMetric::Sent => self.counters.bytes_sent(tags.client, tags.client_transport),
142                ClientTelemetryMetric::Dropped => self.counters.bytes_dropped(tags.client, tags.client_transport),
143                ClientTelemetryMetric::DroppedQueue => {
144                    self.counters.bytes_dropped_queue(tags.client, tags.client_transport)
145                }
146                ClientTelemetryMetric::DroppedWriter => {
147                    self.counters.bytes_dropped_writer(tags.client, tags.client_transport)
148                }
149            };
150            for (_, value) in values {
151                if value.is_finite() && value >= 0.0 && value.fract() == 0.0 && value <= u64::MAX as f64 {
152                    counter.increment(value as u64);
153                }
154            }
155        }
156    }
157}
158
159#[async_trait]
160impl Destination for DogStatsDClientTelemetry {
161    async fn run(self: Box<Self>, mut context: DestinationContext) -> Result<(), GenericError> {
162        let mut health = context.take_health_handle();
163        health.mark_ready();
164        debug!("DogStatsD client telemetry destination started.");
165
166        loop {
167            select! {
168                _ = health.live() => continue,
169                maybe_events = context.events().next() => match maybe_events {
170                    Some(events) => {
171                        for event in events {
172                            if let Event::Metric(metric) = event {
173                                self.record_metric(&metric);
174                            }
175                        }
176                    },
177                    None => break,
178                },
179            }
180        }
181
182        debug!("DogStatsD client telemetry destination stopped.");
183        Ok(())
184    }
185}
saluki_components/destinations/dogstatsd_client_telemetry.rs

saluki_components/destinations/
dogstatsd_client_telemetry.rs
