diff --git a/datadog-sidecar/src/service/session_info.rs b/datadog-sidecar/src/service/session_info.rs index fbe88b206e..a19bce2579 100644 --- a/datadog-sidecar/src/service/session_info.rs +++ b/datadog-sidecar/src/service/session_info.rs @@ -30,7 +30,7 @@ pub(crate) struct SessionInfo { pub(crate) session_config: Arc>>, debugger_config: Arc>, tracer_config: Arc>, - dogstatsd: Arc>>, + dogstatsd: Arc>>, remote_config_options: Arc>>, pub(crate) agent_infos: Arc>>, pub(crate) remote_config_interval: Arc>, @@ -210,13 +210,15 @@ impl SessionInfo { f(&mut self.get_trace_config()); } - pub(crate) fn get_dogstatsd(&self) -> MutexGuard<'_, Option> { + pub(crate) fn get_dogstatsd( + &self, + ) -> MutexGuard<'_, Option> { self.dogstatsd.lock_or_panic() } pub(crate) fn configure_dogstatsd(&self, f: F) where - F: FnOnce(&mut Option), + F: FnOnce(&mut Option), { f(&mut self.get_dogstatsd()); } diff --git a/datadog-sidecar/src/service/sidecar_server.rs b/datadog-sidecar/src/service/sidecar_server.rs index 4565781e33..f56deca7aa 100644 --- a/datadog-sidecar/src/service/sidecar_server.rs +++ b/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| { diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index eef1c9e3f9..552cdfb1f7 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/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 TraceExporterBuilder { 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); diff --git a/libdd-data-pipeline/src/trace_exporter/metrics.rs b/libdd-data-pipeline/src/trace_exporter/metrics.rs index 776ddee598..024197116d 100644 --- a/libdd-data-pipeline/src/trace_exporter/metrics.rs +++ b/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, diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 08f69e8b99..1722c2dbe1 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/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, /// None if dogstatsd is disabled - dogstatsd: Option>, + dogstatsd: Option, common_stats_tags: Vec, client_computed_top_level: bool, client_side_stats: StatsComputationConfig, @@ -567,7 +567,7 @@ impl< /// Emit a health metric to dogstatsd fn emit_metric(&self, metric: HealthMetric, custom_tags: Option>) { 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); } } @@ -575,7 +575,7 @@ impl< /// 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); } } diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 8d7c7a47f1..00c9ea2b70 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/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, /// Optional DogStatsD client forwarded to the [`StatsExporter`]. - pub dogstatsd: Option>, + pub dogstatsd: Option, /// Optional telemetry handle forwarded to the [`StatsExporter`]. #[cfg(feature = "telemetry")] pub telemetry: Option, diff --git a/libdd-dogstatsd-client/src/action.rs b/libdd-dogstatsd-client/src/action.rs new file mode 100644 index 0000000000..2edd654232 --- /dev/null +++ b/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), + #[allow(missing_docs)] + Distribution(String, f64, Vec), + #[allow(missing_docs)] + Gauge(String, f64, Vec), + #[allow(missing_docs)] + Histogram(String, f64, Vec), + /// 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), +} + +/// 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, V: IntoIterator> { + // TODO: instead of AsRef 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::(), + // std::mem::variant_count::>>(), + // "DogStatsDActionOwned and DogStatsDAction should have the same number of variants, + // did you forget to update one?", ); + } +} diff --git a/libdd-dogstatsd-client/src/client/mod.rs b/libdd-dogstatsd-client/src/client/mod.rs new file mode 100644 index 0000000000..b06cd3641e --- /dev/null +++ b/libdd-dogstatsd-client/src/client/mod.rs @@ -0,0 +1,242 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use crate::action::{DogStatsDAction, DogStatsDActionOwned}; +use cadence::prelude::*; +use cadence::{Metric, MetricBuilder, QueuingMetricSink, StatsdClient}; +use libdd_common::tag::Tag; +use libdd_common::Endpoint; +use sink::create_udp_sink; +#[cfg(unix)] +use sink::create_unix_sink; +use std::sync::{Arc, OnceLock}; +use tracing::error; + +/// Provides transport sink which can be wrapped in buffered sink +mod sink; + +// Queue with a maximum capacity of 32K elements +const QUEUE_SIZE: usize = 32 * 1024; + +/// A dogstatsd-client that flushes stats to a given endpoint. +/// +/// This client can be cloned and shared between threads. +#[derive(Debug, Default, Clone)] +pub struct DogStatsDClient { + inner: Arc, +} + +/// Inner struct of [DogStatsDClient] to be wrapped in [Arc]. +#[derive(Debug, Default)] +struct InnerClient { + client: OnceLock, + endpoint: Endpoint, +} + +impl DogStatsDClient { + /// Build a new client instance pointed at the provided endpoint. + /// Returns error if the provided endpoint is not valid. + pub fn new(endpoint: Endpoint) -> anyhow::Result { + // defer initialization of the client until the first metric is sent and we definitely know + // the client is going to be used to communicate with the endpoint. + Ok(Self { + inner: Arc::new(InnerClient { + endpoint, + ..Default::default() + }), + }) + } + + /// Send a vector of DogStatsDActionOwned, this is the same as `send` except it uses the + /// "owned" version of DogStatsDAction. See the docs for DogStatsDActionOwned for details. + pub fn send_owned(&self, actions: Vec) { + match self.get_or_init_client() { + Ok(client) => { + for action in actions { + if let Err(err) = match action { + DogStatsDActionOwned::Count(metric, value, tags) => { + Self::do_send(client.count_with_tags(metric.as_ref(), value), &tags) + } + DogStatsDActionOwned::Distribution(metric, value, tags) => Self::do_send( + client.distribution_with_tags(metric.as_ref(), value), + &tags, + ), + DogStatsDActionOwned::Gauge(metric, value, tags) => { + Self::do_send(client.gauge_with_tags(metric.as_ref(), value), &tags) + } + DogStatsDActionOwned::Histogram(metric, value, tags) => { + Self::do_send(client.histogram_with_tags(metric.as_ref(), value), &tags) + } + DogStatsDActionOwned::Set(metric, value, tags) => { + Self::do_send(client.set_with_tags(metric.as_ref(), value), &tags) + } + } { + error!(?err, "Error while sending metric"); + } + } + } + Err(e) => { + error!("Failed to acquire dogstatsd client lock: {e}"); + } + }; + } + + /// Send a vector of DogStatsDAction, this is the same as `send_owned` except it only borrows + /// the provided values. See the docs for DogStatsDActionOwned for details. + pub fn send<'a, T: AsRef, V: IntoIterator>( + &self, + actions: Vec>, + ) { + match self.get_or_init_client() { + Ok(client) => { + for action in actions { + if let Err(err) = match action { + DogStatsDAction::Count(metric, value, tags) => { + let metric_builder = client.count_with_tags(metric.as_ref(), value); + Self::do_send(metric_builder, tags) + } + DogStatsDAction::Distribution(metric, value, tags) => Self::do_send( + client.distribution_with_tags(metric.as_ref(), value), + tags, + ), + DogStatsDAction::Gauge(metric, value, tags) => { + Self::do_send(client.gauge_with_tags(metric.as_ref(), value), tags) + } + DogStatsDAction::Histogram(metric, value, tags) => { + Self::do_send(client.histogram_with_tags(metric.as_ref(), value), tags) + } + DogStatsDAction::Set(metric, value, tags) => { + Self::do_send(client.set_with_tags(metric.as_ref(), value), tags) + } + } { + error!(?err, "Error while sending metric"); + } + } + } + Err(e) => { + error!(?e, "Failed to get client"); + } + } + } + + fn get_or_init_client(&self) -> anyhow::Result<&StatsdClient> { + match self.inner.client.get() { + Some(client) => Ok(client), + None => { + let client = Self::create_client(&self.inner.endpoint)?; + Ok(self.inner.client.get_or_init(|| client)) + } + } + } + + fn create_client(endpoint: &Endpoint) -> anyhow::Result { + match endpoint.url.scheme_str() { + #[cfg(unix)] + Some("unix") => Ok(StatsdClient::from_sink( + "", + QueuingMetricSink::with_capacity(create_unix_sink(endpoint)?, QUEUE_SIZE), + )), + _ => Ok(StatsdClient::from_sink( + "", + QueuingMetricSink::with_capacity(create_udp_sink(endpoint)?, QUEUE_SIZE), + )), + } + } + + fn do_send<'m, 't, T, V: IntoIterator>( + mut builder: MetricBuilder<'m, '_, T>, + tags: V, + ) -> anyhow::Result<()> + where + T: Metric + From, + 't: 'm, + { + for tag in tags { + builder = builder.with_tag_value(tag.as_ref()); + } + builder.try_send()?; + Ok(()) + } +} + +#[cfg(test)] +mod test { + use super::*; + use crate::action::DogStatsDAction::{Count, Distribution, Gauge, Histogram, Set}; + use libdd_common::{tag, Endpoint}; + use std::net; + use std::sync::Arc; + use std::time::Duration; + + #[test] + #[cfg_attr(miri, ignore)] + fn test_flusher() { + let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind host socket"); + let _ = socket.set_read_timeout(Some(Duration::from_millis(500))); + + let flusher = DogStatsDClient::new(Endpoint::from_slice( + socket.local_addr().unwrap().to_string().as_str(), + )) + .unwrap(); + flusher.send(vec![ + Count("test_count", 3, &vec![tag!("foo", "bar")]), + Count("test_neg_count", -2, &vec![]), + Distribution("test_distribution", 4.2, &vec![]), + Gauge("test_gauge", 7.6, &vec![]), + Histogram("test_histogram", 8.0, &vec![]), + Set("test_set", 9, &vec![tag!("the", "end")]), + Set("test_neg_set", -1, &vec![]), + ]); + + fn read(socket: &net::UdpSocket) -> String { + let mut buf = [0; 100]; + socket.recv(&mut buf).expect("No data"); + let datagram = String::from_utf8_lossy(buf.strip_suffix(&[0]).unwrap()); + datagram.trim_matches(char::from(0)).to_string() + } + + assert_eq!("test_count:3|c|#foo:bar", read(&socket)); + assert_eq!("test_neg_count:-2|c", read(&socket)); + assert_eq!("test_distribution:4.2|d", read(&socket)); + assert_eq!("test_gauge:7.6|g", read(&socket)); + assert_eq!("test_histogram:8|h", read(&socket)); + assert_eq!("test_set:9|s|#the:end", read(&socket)); + assert_eq!("test_neg_set:-1|s", read(&socket)); + } + + #[tokio::test] + #[cfg_attr(miri, ignore)] + async fn test_thread_safety() { + let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind host socket"); + let _ = socket.set_read_timeout(Some(Duration::from_millis(500))); + let endpoint = Endpoint::from_slice(socket.local_addr().unwrap().to_string().as_str()); + let flusher = Arc::new(DogStatsDClient::new(endpoint.clone()).unwrap()); + + { + assert!(flusher.inner.client.get().is_none()); + } + + let tasks: Vec<_> = (0..10) + .map(|_| { + let flusher_clone = Arc::clone(&flusher); + tokio::spawn(async move { + flusher_clone.send(vec![ + Count("test_count", 3, &vec![tag!("foo", "bar")]), + Count("test_neg_count", -2, &vec![]), + Distribution("test_distribution", 4.2, &vec![]), + Gauge("test_gauge", 7.6, &vec![]), + Histogram("test_histogram", 8.0, &vec![]), + Set("test_set", 9, &vec![tag!("the", "end")]), + Set("test_neg_set", -1, &vec![]), + ]); + + assert!(flusher_clone.inner.client.get().is_some()); + }) + }) + .collect(); + + for task in tasks { + task.await.unwrap(); + } + } +} diff --git a/libdd-dogstatsd-client/src/client/sink.rs b/libdd-dogstatsd-client/src/client/sink.rs new file mode 100644 index 0000000000..5bc6a45f46 --- /dev/null +++ b/libdd-dogstatsd-client/src/client/sink.rs @@ -0,0 +1,94 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use anyhow::anyhow; +use cadence::UdpMetricSink; +#[cfg(unix)] +use cadence::UnixMetricSink; +#[cfg(unix)] +use libdd_common::connector::uds::socket_path_from_uri; +use libdd_common::Endpoint; +use std::net::{ToSocketAddrs, UdpSocket}; +#[cfg(unix)] +use std::os::unix::net::UnixDatagram; + +#[cfg(unix)] +pub(super) fn create_unix_sink(endpoint: &Endpoint) -> anyhow::Result { + let socket = + UnixDatagram::unbound().map_err(|e| anyhow!("failed to make unbound unix port: {}", e))?; + socket + .set_nonblocking(true) + .map_err(|e| anyhow!("failed to set socket to nonblocking: {}", e))?; + Ok(UnixMetricSink::from( + socket_path_from_uri(&endpoint.url) + .map_err(|e| anyhow!("failed to build socket path from uri: {}", e))?, + socket, + )) +} + +pub(super) fn create_udp_sink(endpoint: &Endpoint) -> anyhow::Result { + let host = endpoint.url.host().ok_or(anyhow!("invalid host"))?; + let port = endpoint.url.port().ok_or(anyhow!("invalid port"))?.as_u16(); + + let server_address = (host, port) + .to_socket_addrs()? + .next() + .ok_or(anyhow!("invalid address"))?; + + let socket = if server_address.is_ipv4() { + UdpSocket::bind("0.0.0.0:0").map_err(|e| anyhow!("failed to bind to 0.0.0.0:0: {}", e))? + } else { + UdpSocket::bind("[::]:0").map_err(|e| anyhow!("failed to bind to [::]:0: {}", e))? + }; + socket.set_nonblocking(true)?; + + UdpMetricSink::from((host, port), socket) + .map_err(|e| anyhow!("failed to build UdpMetricSink: {}", e)) +} + +#[cfg(test)] +mod tests { + use super::*; + #[cfg(unix)] + use http::Uri; + #[cfg(unix)] + use libdd_common::connector::uds::socket_path_to_uri; + use libdd_common::Endpoint; + + #[test] + #[cfg_attr(miri, ignore)] + fn test_create_udp_sink() { + let res = create_udp_sink(&Endpoint::default()); + assert!(res.is_err()); + assert_eq!("invalid host", res.unwrap_err().to_string().as_str()); + + let res = create_udp_sink(&Endpoint::from_slice("localhost:99999")); + assert!(res.is_err()); + assert_eq!("invalid port", res.unwrap_err().to_string().as_str()); + + let res = create_udp_sink(&Endpoint::from_slice("localhost:80")); + assert!(res.is_ok()); + + let res = create_udp_sink(&Endpoint::from_slice("http://localhost:80")); + assert!(res.is_ok()); + } + + #[test] + #[cfg(unix)] + #[cfg_attr(miri, ignore)] + fn test_create_unix_sink() { + let res = create_unix_sink(&Endpoint::from_url( + "unix://localhost:80".parse::().unwrap(), + )); + assert!(res.is_err()); + assert_eq!( + "failed to build socket path from uri: invalid url", + res.unwrap_err().to_string().as_str() + ); + + let res = create_unix_sink(&Endpoint::from_url( + socket_path_to_uri("/path/to/a/socket.sock".as_ref()).unwrap(), + )); + assert!(res.is_ok()); + } +} diff --git a/libdd-dogstatsd-client/src/lib.rs b/libdd-dogstatsd-client/src/lib.rs index f41d9e837a..4dd019f1d4 100644 --- a/libdd-dogstatsd-client/src/lib.rs +++ b/libdd-dogstatsd-client/src/lib.rs @@ -10,406 +10,10 @@ //! dogstatsd-client implements a client to emit metrics to a dogstatsd server. //! This is made use of in at least the data-pipeline and sidecar crates. -use libdd_common::tag::Tag; -use libdd_common::Endpoint; -use serde::{Deserialize, Serialize}; -use std::fmt::Debug; -use tracing::error; +/// DogStatsD action types that can be sent to a DogStatsD server. +pub mod action; +/// DogStatsD client and supporting sink logic. +pub mod client; -use anyhow::anyhow; -use cadence::prelude::*; -#[cfg(unix)] -use cadence::UnixMetricSink; -use cadence::{Metric, MetricBuilder, QueuingMetricSink, StatsdClient, UdpMetricSink}; -#[cfg(unix)] -use libdd_common::connector::uds::socket_path_from_uri; -use std::net::{ToSocketAddrs, UdpSocket}; -#[cfg(unix)] -use std::os::unix::net::UnixDatagram; -use std::sync::{Arc, Mutex}; - -// Queue with a maximum capacity of 32K elements -const QUEUE_SIZE: usize = 32 * 1024; - -/// 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), - #[allow(missing_docs)] - Distribution(String, f64, Vec), - #[allow(missing_docs)] - Gauge(String, f64, Vec), - #[allow(missing_docs)] - Histogram(String, f64, Vec), - /// 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), -} - -/// 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, V: IntoIterator> { - // TODO: instead of AsRef 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), -} - -/// A dogstatsd-client that flushes stats to a given endpoint. -#[derive(Debug, Default)] -pub struct Client { - client: Mutex>>, - endpoint: Option, -} - -/// Build a new flusher instance pointed at the provided endpoint. -/// Returns error if the provided endpoint is not valid. -pub fn new(endpoint: Endpoint) -> anyhow::Result { - // defer initialization of the client until the first metric is sent and we definitely know the - // client is going to be used to communicate with the endpoint. - Ok(Client { - endpoint: Some(endpoint), - ..Default::default() - }) -} - -impl Client { - /// Send a vector of DogStatsDActionOwned, this is the same as `send` except it uses the "owned" - /// version of DogStatsDAction. See the docs for DogStatsDActionOwned for details. - pub fn send_owned(&self, actions: Vec) { - let client_opt = match self.get_or_init_client() { - Ok(client) => client, - Err(e) => { - error!(?e, "Failed to get client"); - return; - } - }; - - if let Some(client) = &*client_opt { - for action in actions { - if let Err(err) = match action { - DogStatsDActionOwned::Count(metric, value, tags) => { - do_send(client.count_with_tags(metric.as_ref(), value), &tags) - } - DogStatsDActionOwned::Distribution(metric, value, tags) => { - do_send(client.distribution_with_tags(metric.as_ref(), value), &tags) - } - DogStatsDActionOwned::Gauge(metric, value, tags) => { - do_send(client.gauge_with_tags(metric.as_ref(), value), &tags) - } - DogStatsDActionOwned::Histogram(metric, value, tags) => { - do_send(client.histogram_with_tags(metric.as_ref(), value), &tags) - } - DogStatsDActionOwned::Set(metric, value, tags) => { - do_send(client.set_with_tags(metric.as_ref(), value), &tags) - } - } { - error!(?err, "Error while sending metric"); - } - } - } - } - - /// Send a vector of DogStatsDAction, this is the same as `send_owned` except it only borrows - /// the provided values.See the docs for DogStatsDActionOwned for details. - pub fn send<'a, T: AsRef, V: IntoIterator>( - &self, - actions: Vec>, - ) { - let client_opt = match self.get_or_init_client() { - Ok(client) => client, - Err(e) => { - error!(?e, "Failed to get client"); - return; - } - }; - if let Some(client) = &*client_opt { - for action in actions { - if let Err(err) = match action { - DogStatsDAction::Count(metric, value, tags) => { - let metric_builder = client.count_with_tags(metric.as_ref(), value); - do_send(metric_builder, tags) - } - DogStatsDAction::Distribution(metric, value, tags) => { - do_send(client.distribution_with_tags(metric.as_ref(), value), tags) - } - DogStatsDAction::Gauge(metric, value, tags) => { - do_send(client.gauge_with_tags(metric.as_ref(), value), tags) - } - DogStatsDAction::Histogram(metric, value, tags) => { - do_send(client.histogram_with_tags(metric.as_ref(), value), tags) - } - DogStatsDAction::Set(metric, value, tags) => { - do_send(client.set_with_tags(metric.as_ref(), value), tags) - } - } { - error!(?err, "Error while sending metric"); - } - } - } - } - - fn get_or_init_client(&self) -> anyhow::Result>> { - if let Some(endpoint) = &self.endpoint { - let mut client_guard = self - .client - .lock() - .map_err(|e| anyhow!("Failed to acquire dogstatsd client lock: {e}"))?; - return if client_guard.is_some() { - Ok(client_guard.clone()) - } else { - let client = Arc::new(Some(create_client(endpoint)?)); - *client_guard = client.clone(); - Ok(client) - }; - } - - Ok(None.into()) - } -} - -fn do_send<'m, 't, T, V: IntoIterator>( - mut builder: MetricBuilder<'m, '_, T>, - tags: V, -) -> anyhow::Result<()> -where - T: Metric + From, - 't: 'm, -{ - let mut tags_iter = tags.into_iter(); - let mut tag_opt = tags_iter.next(); - #[allow(clippy::unwrap_used)] - while tag_opt.is_some() { - builder = builder.with_tag_value(tag_opt.unwrap().as_ref()); - tag_opt = tags_iter.next(); - } - builder.try_send()?; - Ok(()) -} - -fn create_client(endpoint: &Endpoint) -> anyhow::Result { - match endpoint.url.scheme_str() { - #[cfg(unix)] - Some("unix") => { - let socket = UnixDatagram::unbound() - .map_err(|e| anyhow!("failed to make unbound unix port: {}", e))?; - socket - .set_nonblocking(true) - .map_err(|e| anyhow!("failed to set socket to nonblocking: {}", e))?; - let sink = QueuingMetricSink::with_capacity( - UnixMetricSink::from( - socket_path_from_uri(&endpoint.url) - .map_err(|e| anyhow!("failed to build socket path from uri: {}", e))?, - socket, - ), - QUEUE_SIZE, - ); - - Ok(StatsdClient::from_sink("", sink)) - } - _ => { - let host = endpoint.url.host().ok_or(anyhow!("invalid host"))?; - let port = endpoint.url.port().ok_or(anyhow!("invalid port"))?.as_u16(); - - let server_address = (host, port) - .to_socket_addrs()? - .next() - .ok_or(anyhow!("invalid address"))?; - - let socket = if server_address.is_ipv4() { - UdpSocket::bind("0.0.0.0:0") - .map_err(|e| anyhow!("failed to bind to 0.0.0.0:0: {}", e))? - } else { - UdpSocket::bind("[::]:0").map_err(|e| anyhow!("failed to bind to [::]:0: {}", e))? - }; - socket.set_nonblocking(true)?; - - let sink = QueuingMetricSink::with_capacity( - UdpMetricSink::from((host, port), socket) - .map_err(|e| anyhow!("failed to build UdpMetricSink: {}", e))?, - QUEUE_SIZE, - ); - - Ok(StatsdClient::from_sink("", sink)) - } - } -} - -#[cfg(test)] -mod test { - use crate::DogStatsDAction::{Count, Distribution, Gauge, Histogram, Set}; - use crate::{create_client, new, DogStatsDActionOwned}; - #[cfg(unix)] - use http::Uri; - #[cfg(unix)] - use libdd_common::connector::uds::socket_path_to_uri; - use libdd_common::{tag, Endpoint}; - use std::net; - use std::sync::Arc; - use std::time::Duration; - - #[test] - #[cfg_attr(miri, ignore)] - fn test_flusher() { - let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind host socket"); - let _ = socket.set_read_timeout(Some(Duration::from_millis(500))); - - let flusher = new(Endpoint::from_slice( - socket.local_addr().unwrap().to_string().as_str(), - )) - .unwrap(); - flusher.send(vec![ - Count("test_count", 3, &vec![tag!("foo", "bar")]), - Count("test_neg_count", -2, &vec![]), - Distribution("test_distribution", 4.2, &vec![]), - Gauge("test_gauge", 7.6, &vec![]), - Histogram("test_histogram", 8.0, &vec![]), - Set("test_set", 9, &vec![tag!("the", "end")]), - Set("test_neg_set", -1, &vec![]), - ]); - - fn read(socket: &net::UdpSocket) -> String { - let mut buf = [0; 100]; - socket.recv(&mut buf).expect("No data"); - let datagram = String::from_utf8_lossy(buf.strip_suffix(&[0]).unwrap()); - datagram.trim_matches(char::from(0)).to_string() - } - - assert_eq!("test_count:3|c|#foo:bar", read(&socket)); - assert_eq!("test_neg_count:-2|c", read(&socket)); - assert_eq!("test_distribution:4.2|d", read(&socket)); - assert_eq!("test_gauge:7.6|g", read(&socket)); - assert_eq!("test_histogram:8|h", read(&socket)); - assert_eq!("test_set:9|s|#the:end", read(&socket)); - assert_eq!("test_neg_set:-1|s", read(&socket)); - } - - #[test] - #[cfg_attr(miri, ignore)] - fn test_create_client_udp() { - let res = create_client(&Endpoint::default()); - assert!(res.is_err()); - assert_eq!("invalid host", res.unwrap_err().to_string().as_str()); - - let res = create_client(&Endpoint::from_slice("localhost:99999")); - assert!(res.is_err()); - assert_eq!("invalid port", res.unwrap_err().to_string().as_str()); - - let res = create_client(&Endpoint::from_slice("localhost:80")); - assert!(res.is_ok()); - - let res = create_client(&Endpoint::from_slice("http://localhost:80")); - assert!(res.is_ok()); - } - - #[test] - #[cfg(unix)] - #[cfg_attr(miri, ignore)] - fn test_create_client_unix_domain_socket() { - let res = create_client(&Endpoint::from_url( - "unix://localhost:80".parse::().unwrap(), - )); - assert!(res.is_err()); - assert_eq!( - "failed to build socket path from uri: invalid url", - res.unwrap_err().to_string().as_str() - ); - - let res = create_client(&Endpoint::from_url( - socket_path_to_uri("/path/to/a/socket.sock".as_ref()).unwrap(), - )); - assert!(res.is_ok()); - } - - #[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 = Count("test".to_string(), 1, vec![]); - match act { - Count(_, _, _) => {} - Distribution(_, _, _) => {} - Gauge(_, _, _) => {} - Histogram(_, _, _) => {} - Set(_, _, _) => {} - } - // TODO: when std::mem::variant_count is in stable we can do this instead - // assert_eq!( - // std::mem::variant_count::(), - // std::mem::variant_count::>>(), - // "DogStatsDActionOwned and DogStatsDAction should have the same number of variants, - // did you forget to update one?", ); - } - - #[tokio::test] - #[cfg_attr(miri, ignore)] - async fn test_thread_safety() { - let socket = net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind host socket"); - let _ = socket.set_read_timeout(Some(Duration::from_millis(500))); - let endpoint = Endpoint::from_slice(socket.local_addr().unwrap().to_string().as_str()); - let flusher = Arc::new(new(endpoint.clone()).unwrap()); - - { - let client = flusher - .client - .lock() - .expect("failed to obtain lock on client"); - assert!(client.is_none()); - } - - let tasks: Vec<_> = (0..10) - .map(|_| { - let flusher_clone = Arc::clone(&flusher); - tokio::spawn(async move { - flusher_clone.send(vec![ - Count("test_count", 3, &vec![tag!("foo", "bar")]), - Count("test_neg_count", -2, &vec![]), - Distribution("test_distribution", 4.2, &vec![]), - Gauge("test_gauge", 7.6, &vec![]), - Histogram("test_histogram", 8.0, &vec![]), - Set("test_set", 9, &vec![tag!("the", "end")]), - Set("test_neg_set", -1, &vec![]), - ]); - - let client = flusher_clone - .client - .lock() - .expect("failed to obtain lock on client within send thread"); - assert!(client.is_some()); - }) - }) - .collect(); - - for task in tasks { - task.await.unwrap(); - } - } -} +pub use action::{DogStatsDAction, DogStatsDActionOwned}; +pub use client::DogStatsDClient; diff --git a/libdd-trace-stats/src/stats_exporter.rs b/libdd-trace-stats/src/stats_exporter.rs index c771fd5b2b..101af45751 100644 --- a/libdd-trace-stats/src/stats_exporter.rs +++ b/libdd-trace-stats/src/stats_exporter.rs @@ -107,7 +107,7 @@ pub struct StatsExporter< )>, /// Optional DogStatsD client. #[cfg(feature = "dogstatsd")] - dogstatsd: Option>, + dogstatsd: Option, } impl @@ -133,7 +133,7 @@ impl #[cfg(feature = "telemetry")] telemetry: Option< libdd_telemetry::worker::TelemetryWorkerHandle, >, - #[cfg(feature = "dogstatsd")] dogstatsd: Option>, + #[cfg(feature = "dogstatsd")] dogstatsd: Option, ) -> Self { #[cfg(feature = "telemetry")] let telemetry = telemetry.map(|handle| { @@ -690,10 +690,9 @@ mod tests { .unwrap(); let addr = socket.local_addr().unwrap().to_string(); - let dogstatsd_client = Arc::new( - libdd_dogstatsd_client::new(libdd_common::Endpoint::from_slice(&addr)) - .expect("failed to create dogstatsd client"), - ); + let dogstatsd_client = + libdd_dogstatsd_client::DogStatsDClient::new(libdd_common::Endpoint::from_slice(&addr)) + .expect("failed to create dogstatsd client"); // get_test_concentrator() has no cardinality collapse: collapsed_spans will be 0. let stats_exporter = StatsExporter::::new( @@ -742,10 +741,9 @@ mod tests { .unwrap(); let addr = socket.local_addr().unwrap().to_string(); - let dogstatsd_client = Arc::new( - libdd_dogstatsd_client::new(libdd_common::Endpoint::from_slice(&addr)) - .expect("failed to create dogstatsd client"), - ); + let dogstatsd_client = + libdd_dogstatsd_client::DogStatsDClient::new(libdd_common::Endpoint::from_slice(&addr)) + .expect("failed to create dogstatsd client"); let stats_exporter = StatsExporter::::new( BUCKETS_DURATION,