refactor(statsd): wrap client in arc

Author: VianneyRuhlmann

datadog-sidecar/src/service/session_info.rs

@@ -30,7 +30,7 @@ pub(crate) struct SessionInfo {
     pub(crate) session_config: Arc<Mutex<Option<libdd_telemetry::config::Config>>>,
     debugger_config: Arc<Mutex<datadog_live_debugger::sender::Config>>,
     tracer_config: Arc<Mutex<tracer::Config>>,
-    dogstatsd: Arc<Mutex<Option<libdd_dogstatsd_client::Client>>>,
+    dogstatsd: Arc<Mutex<Option<libdd_dogstatsd_client::DogStatsDClient>>>,
     remote_config_options: Arc<Mutex<Option<ConfigOptions>>>,
     pub(crate) agent_infos: Arc<Mutex<Option<AgentInfoGuard>>>,
     pub(crate) remote_config_interval: Arc<Mutex<Duration>>,
@@ -210,13 +210,15 @@ impl SessionInfo {
         f(&mut self.get_trace_config());
     }
 
-    pub(crate) fn get_dogstatsd(&self) -> MutexGuard<'_, Option<libdd_dogstatsd_client::Client>> {
+    pub(crate) fn get_dogstatsd(
+        &self,
+    ) -> MutexGuard<'_, Option<libdd_dogstatsd_client::DogStatsDClient>> {
         self.dogstatsd.lock_or_panic()
     }
 
     pub(crate) fn configure_dogstatsd<F>(&self, f: F)
     where
-        F: FnOnce(&mut Option<libdd_dogstatsd_client::Client>),
+        F: FnOnce(&mut Option<libdd_dogstatsd_client::DogStatsDClient>),
     {
         f(&mut self.get_dogstatsd());
     }

datadog-sidecar/src/service/sidecar_server.rs

@@ -47,7 +47,7 @@ use datadog_ipc::ipc_server::OwnedServerConn;
 use datadog_live_debugger::sender::{agent_info_supports_debugger_v2_endpoint, DebuggerType};
 use libdd_capabilities_impl::NativeCapabilities;
 use libdd_common::tag::Tag;
-use libdd_dogstatsd_client::{new, DogStatsDActionOwned};
+use libdd_dogstatsd_client::{DogStatsDActionOwned, DogStatsDClient};
 use libdd_remote_config::fetch::{ConfigInvariants, ConfigOptions, MultiTargetStats};
 use libdd_telemetry::config::{Config, TelemetryEndpoint};
 use libdd_tinybytes as tinybytes;
@@ -753,7 +753,7 @@ impl SidecarInterface for ConnectionSidecarHandler {
             *endpoint = config.otlp_metrics_endpoint.clone();
         });
         session.configure_dogstatsd(|dogstatsd| {
-            let d = new(config.dogstatsd_endpoint.clone()).ok();
+            let d = DogStatsDClient::new(config.dogstatsd_endpoint.clone()).ok();
             *dogstatsd = d;
         });
         session.modify_debugger_config(|cfg| {

libdd-data-pipeline/src/trace_exporter/builder.rs

@@ -22,7 +22,7 @@ use crate::trace_exporter::{
 use arc_swap::ArcSwap;
 use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability};
 use libdd_common::{parse_uri, tag, Endpoint};
-use libdd_dogstatsd_client::new;
+use libdd_dogstatsd_client::DogStatsDClient;
 use libdd_shared_runtime::SharedRuntime;
 #[cfg(not(target_arch = "wasm32"))]
 use libdd_shared_runtime::{BlockingRuntime, ForkSafeRuntime};
@@ -577,7 +577,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
 
         let dogstatsd = self
             .dogstatsd_url
-            .and_then(|u| new(Endpoint::from_slice(&u)).ok().map(Arc::new));
+            .and_then(|u| DogStatsDClient::new(Endpoint::from_slice(&u)).ok());
 
         let base_url = self.url.as_deref().unwrap_or(DEFAULT_AGENT_URL);
 

libdd-data-pipeline/src/trace_exporter/metrics.rs

@@ -4,19 +4,19 @@
 use crate::health_metrics::{HealthMetric, SendResult};
 use either::Either;
 use libdd_common::tag::Tag;
-use libdd_dogstatsd_client::{Client, DogStatsDAction};
+use libdd_dogstatsd_client::{DogStatsDAction, DogStatsDClient};
 use tracing::debug;
 
 /// Handles emission of health metrics to DogStatsD
 #[derive(Debug)]
 pub(crate) struct MetricsEmitter<'a> {
-    dogstatsd: Option<&'a Client>,
+    dogstatsd: Option<&'a DogStatsDClient>,
     common_tags: &'a [Tag],
 }
 
 impl<'a> MetricsEmitter<'a> {
     /// Create a new MetricsEmitter
-    pub(crate) fn new(dogstatsd: Option<&'a Client>, common_tags: &'a [Tag]) -> Self {
+    pub(crate) fn new(dogstatsd: Option<&'a DogStatsDClient>, common_tags: &'a [Tag]) -> Self {
         Self {
             dogstatsd,
             common_tags,

libdd-data-pipeline/src/trace_exporter/mod.rs

@@ -42,7 +42,7 @@ use http::Uri;
 use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability};
 use libdd_common::tag::Tag;
 use libdd_common::Endpoint;
-use libdd_dogstatsd_client::Client;
+use libdd_dogstatsd_client::DogStatsDClient;
 #[cfg(not(target_arch = "wasm32"))]
 use libdd_shared_runtime::BlockingRuntime;
 use libdd_shared_runtime::{SharedRuntime, WorkerHandle};
@@ -253,7 +253,7 @@ pub struct TraceExporter<
     serializer: TraceSerializer,
     shared_runtime: Arc<R>,
     /// None if dogstatsd is disabled
-    dogstatsd: Option<Arc<Client>>,
+    dogstatsd: Option<DogStatsDClient>,
     common_stats_tags: Vec<Tag>,
     client_computed_top_level: bool,
     client_side_stats: StatsComputationConfig,
@@ -567,15 +567,15 @@ impl<
     /// Emit a health metric to dogstatsd
     fn emit_metric(&self, metric: HealthMetric, custom_tags: Option<Vec<&Tag>>) {
         if self.health_metrics_enabled {
-            let emitter = MetricsEmitter::new(self.dogstatsd.as_deref(), &self.common_stats_tags);
+            let emitter = MetricsEmitter::new(self.dogstatsd.as_ref(), &self.common_stats_tags);
             emitter.emit(metric, custom_tags);
         }
     }
 
     /// Emit all health metrics from a SendResult
     fn emit_send_result(&self, result: &SendResult) {
         if self.health_metrics_enabled {
-            let emitter = MetricsEmitter::new(self.dogstatsd.as_deref(), &self.common_stats_tags);
+            let emitter = MetricsEmitter::new(self.dogstatsd.as_ref(), &self.common_stats_tags);
             emitter.emit_from_send_result(result);
         }
     }

libdd-data-pipeline/src/trace_exporter/stats.rs

@@ -51,7 +51,7 @@ pub(crate) struct StatsContext<'a, R: SharedRuntime> {
     pub shared_runtime: &'a R,
     pub stats_cardinality_limit: Option<usize>,
     /// Optional DogStatsD client forwarded to the [`StatsExporter`].
-    pub dogstatsd: Option<std::sync::Arc<libdd_dogstatsd_client::Client>>,
+    pub dogstatsd: Option<libdd_dogstatsd_client::DogStatsDClient>,
     /// Optional telemetry handle forwarded to the [`StatsExporter`].
     #[cfg(feature = "telemetry")]
     pub telemetry: Option<libdd_telemetry::worker::TelemetryWorkerHandle>,

libdd-dogstatsd-client/src/action.rs

@@ -0,0 +1,82 @@
+// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+use libdd_common::tag::Tag;
+use serde::{Deserialize, Serialize};
+use std::fmt::Debug;
+
+/// The `DogStatsDActionOwned` enum gathers the metric types that can be sent to the DogStatsD
+/// server. This type takes ownership of the relevant data to support the sidecar better.
+/// For documentation on the dogstatsd metric types: https://docs.datadoghq.com/metrics/types/?tab=count#metric-types
+///
+/// Originally I attempted to combine this type with `DogStatsDAction` but this GREATLY complicates
+/// the types to the point of insanity. I was unable to come up with a satisfactory approach that
+/// allows both the data-pipeline and sidecar crates to use the same type. If a future rustacean
+/// wants to take a stab and open a PR please do so!
+#[derive(Debug, Serialize, Deserialize)]
+pub enum DogStatsDActionOwned {
+    #[allow(missing_docs)]
+    Count(String, i64, Vec<Tag>),
+    #[allow(missing_docs)]
+    Distribution(String, f64, Vec<Tag>),
+    #[allow(missing_docs)]
+    Gauge(String, f64, Vec<Tag>),
+    #[allow(missing_docs)]
+    Histogram(String, f64, Vec<Tag>),
+    /// Cadence only support i64 type as value
+    /// but Golang implementation uses string (https://github.com/DataDog/datadog-go/blob/331d24832f7eac97b091efd696278fe2c4192b29/statsd/statsd.go#L230)
+    /// and PHP implementation uses float or string (https://github.com/DataDog/php-datadogstatsd/blob/0efdd1c38f6d3dd407efbb899ad1fd2e5cd18085/src/DogStatsd.php#L251)
+    Set(String, i64, Vec<Tag>),
+}
+
+/// The `DogStatsDAction` enum gathers the metric types that can be sent to the DogStatsD server.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum DogStatsDAction<'a, T: AsRef<str>, V: IntoIterator<Item = &'a Tag>> {
+    // TODO: instead of AsRef<str> we can accept a marker Trait that users of this crate implement
+    #[allow(missing_docs)]
+    Count(T, i64, V),
+    #[allow(missing_docs)]
+    Distribution(T, f64, V),
+    #[allow(missing_docs)]
+    Gauge(T, f64, V),
+    #[allow(missing_docs)]
+    Histogram(T, f64, V),
+    /// Cadence only support i64 type as value
+    /// but Golang implementation uses string (https://github.com/DataDog/datadog-go/blob/331d24832f7eac97b091efd696278fe2c4192b29/statsd/statsd.go#L230)
+    /// and PHP implementation uses float or string (https://github.com/DataDog/php-datadogstatsd/blob/0efdd1c38f6d3dd407efbb899ad1fd2e5cd18085/src/DogStatsd.php#L251)
+    Set(T, i64, V),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_owned_sync() {
+        // This test ensures that if a new variant is added to either `DogStatsDActionOwned` or
+        // `DogStatsDAction` this test will NOT COMPILE to act as a reminder that BOTH locations
+        // must be updated.
+        let owned_act = DogStatsDActionOwned::Count("test".to_string(), 1, vec![]);
+        match owned_act {
+            DogStatsDActionOwned::Count(_, _, _) => {}
+            DogStatsDActionOwned::Distribution(_, _, _) => {}
+            DogStatsDActionOwned::Gauge(_, _, _) => {}
+            DogStatsDActionOwned::Histogram(_, _, _) => {}
+            DogStatsDActionOwned::Set(_, _, _) => {}
+        }
+
+        let act = DogStatsDAction::Count("test".to_string(), 1, vec![]);
+        match act {
+            DogStatsDAction::Count(_, _, _) => {}
+            DogStatsDAction::Distribution(_, _, _) => {}
+            DogStatsDAction::Gauge(_, _, _) => {}
+            DogStatsDAction::Histogram(_, _, _) => {}
+            DogStatsDAction::Set(_, _, _) => {}
+        }
+        // TODO: when std::mem::variant_count is in stable we can do this instead
+        // assert_eq!(
+        //     std::mem::variant_count::<DogStatsDActionOwned>(),
+        //     std::mem::variant_count::<DogStatsDAction<String, Vec<&Tag>>>(),
+        //     "DogStatsDActionOwned and DogStatsDAction should have the same number of variants,
+        // did you forget to update one?", );
+    }
+}