dogstatsd/

lib.rs

//! A Rust client for interacting with Dogstatsd
//!
//! Dogstatsd is a custom `StatsD` implementation by `DataDog` for sending metrics and events to their
//! system. Through this client you can report any type of metric you want, tag it, and enjoy your
//! custom metrics.
//!
//! ## Usage
//!
//! Build an options struct and create a client:
//!
//! ```
//! use dogstatsd::{Client, Options, OptionsBuilder};
//! use std::time::Duration;
//!
//! // Binds to a udp socket on an available ephemeral port on 127.0.0.1 for
//! // transmitting, and sends to  127.0.0.1:8125, the default dogstatsd
//! // address.
//! let default_options = Options::default();
//! let default_client = Client::new(default_options).unwrap();
//!
//! // Binds to 127.0.0.1:9000 for transmitting and sends to 10.1.2.3:8125, with a
//! // namespace of "analytics".
//! let custom_options = Options::new("127.0.0.1:9000", "10.1.2.3:8125", "analytics", vec!(String::new()), None, None);
//! let custom_client = Client::new(custom_options).unwrap();
//!
//! // You can also use the OptionsBuilder API to avoid needing to specify every option.
//! let built_options = OptionsBuilder::new().from_addr(String::from("127.0.0.1:9001")).build();
//! let built_client = Client::new(built_options).unwrap();
//! ```
//!
//! Start sending metrics:
//!
//! ```
//! use dogstatsd::{Client, Options, ServiceCheckOptions, ServiceStatus};
//!
//! let client = Client::new(Options::default()).unwrap();
//! let tags = &["env:production"];
//!
//! // Increment a counter
//! client.incr("my_counter", tags).unwrap();
//!
//! // Decrement a counter
//! client.decr("my_counter", tags).unwrap();
//!
//! // Time a block of code (reports in ms)
//! client.time("my_time", tags, || {
//!     // Some time consuming code
//! }).unwrap();
//!
//! // Report your own timing in ms
//! client.timing("my_timing", 500, tags).unwrap();
//!
//! // Report an arbitrary value (a gauge)
//! client.gauge("my_gauge", "12345", tags).unwrap();
//!
//! // Report a sample of a histogram
//! client.histogram("my_histogram", "67890", tags).unwrap();
//!
//! // Report a sample of a distribution
//! client.distribution("distribution", "67890", tags).unwrap();
//!
//! // Report a member of a set
//! client.set("my_set", "13579", tags).unwrap();
//!
//! // Report a service check
//! let service_check_options = ServiceCheckOptions {
//!   hostname: Some("my-host.localhost"),
//!   ..Default::default()
//! };
//! client.service_check("redis.can_connect", ServiceStatus::OK, tags, Some(service_check_options)).unwrap();
//!
//! // Send a custom event
//! client.event("My Custom Event Title", "My Custom Event Body", tags).unwrap();
//!
//! use dogstatsd::{EventOptions, EventPriority, EventAlertType};
//! let event_options = EventOptions::new()
//!     .with_timestamp(1638480000)
//!     .with_hostname("localhost")
//!     .with_priority(EventPriority::Normal)
//!     .with_alert_type(EventAlertType::Error);
//! client.event_with_options("My Custom Event Title", "My Custom Event Body", tags, Some(event_options)).unwrap();
//! ```

#![cfg_attr(feature = "unstable", feature(test))]
#![deny(
    warnings,
    missing_debug_implementations,
    missing_copy_implementations,
    missing_docs
)]
extern crate chrono;

use chrono::Utc;
use std::borrow::Cow;
use std::future::Future;
use std::net::UdpSocket;
#[cfg(unix)]
use std::os::unix::net::UnixDatagram;
use std::sync::mpsc::Sender;
use std::sync::{mpsc, Mutex};
use std::thread;
use std::time::Duration;

pub use self::error::DogstatsdError;
use self::metrics::*;
pub use self::metrics::{EventAlertType, EventPriority, ServiceCheckOptions, ServiceStatus};

mod error;
mod metrics;

/// A type alias for returning a unit type or an error
pub type DogstatsdResult = Result<(), DogstatsdError>;

const DEFAULT_FROM_ADDR: &str = "0.0.0.0:0";
const DEFAULT_TO_ADDR: &str = "127.0.0.1:8125";

/// The struct that represents the options available for the Dogstatsd client.
#[derive(Debug, PartialEq, Clone, Copy)]
pub struct BatchingOptions {
    /// The maximum buffer size in bytes of a batch of events.
    pub max_buffer_size: usize,
    /// The maximum time before sending a batch of events.
    pub max_time: Duration,
    /// The maximum retry attempts if we fail to flush our buffer
    pub max_retry_attempts: usize,
    /// Upon retry, there is an exponential backoff, this value sets the starting value
    pub initial_retry_delay: u64,
}

/// The struct that represents the options available for the Dogstatsd client.
#[derive(Debug, PartialEq)]
pub struct Options {
    /// The address of the udp socket we'll bind to for sending.
    pub from_addr: String,
    /// The address of the udp socket we'll send metrics and events to.
    pub to_addr: String,
    /// A namespace to prefix all metrics with, joined with a '.'.
    pub namespace: String,
    /// Default tags to include with every request.
    pub default_tags: Vec<String>,
    /// OPTIONAL, if defined, will use UDS instead of UDP and will ignore UDP options
    pub socket_path: Option<String>,
    /// OPTIONAL, if defined, will utilize batching for sending metrics
    pub batching_options: Option<BatchingOptions>,
}

impl Default for Options {
    /// Create a new options struct with all the default settings.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::Options;
    ///
    ///   let options = Options::default();
    ///
    ///   assert_eq!(
    ///       Options {
    ///           from_addr: "0.0.0.0:0".into(),
    ///           to_addr: "127.0.0.1:8125".into(),
    ///           namespace: String::new(),
    ///           default_tags: vec!(),
    ///           socket_path: None,
    ///           batching_options: None,
    ///       },
    ///       options
    ///   )
    /// ```
    fn default() -> Self {
        Options {
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            default_tags: vec![],
            socket_path: None,
            batching_options: None,
        }
    }
}

impl Options {
    /// Create a new options struct by supplying values for all fields.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::Options;
    ///
    ///   let options = Options::new("127.0.0.1:9000", "127.0.0.1:9001", "", vec!(String::new()), None, None);
    /// ```
    pub fn new(
        from_addr: &str,
        to_addr: &str,
        namespace: &str,
        default_tags: Vec<String>,
        socket_path: Option<String>,
        batching_options: Option<BatchingOptions>,
    ) -> Self {
        Options {
            from_addr: from_addr.into(),
            to_addr: to_addr.into(),
            namespace: namespace.into(),
            default_tags,
            socket_path,
            batching_options,
        }
    }

    fn merge_with_system_tags(default_tags: Vec<String>) -> Vec<String> {
        let mut merged_tags = default_tags;

        if !merged_tags.iter().any(|tag| tag.starts_with("env:")) {
            if let Ok(env) = std::env::var("DD_ENV") {
                merged_tags.push(format!("env:{}", env));
            }
        }
        if !merged_tags.iter().any(|tag| tag.starts_with("service:")) {
            if let Ok(service) = std::env::var("DD_SERVICE") {
                merged_tags.push(format!("service:{}", service));
            }
        }
        if !merged_tags.iter().any(|tag| tag.starts_with("version:")) {
            if let Ok(version) = std::env::var("DD_VERSION") {
                merged_tags.push(format!("version:{}", version));
            }
        }

        merged_tags
    }
}

/// Struct that allows build an `Options` for available for the Dogstatsd client.
#[derive(Default, Debug)]
pub struct OptionsBuilder {
    /// The address of the udp socket we'll bind to for sending.
    from_addr: Option<String>,
    /// The address of the udp socket we'll send metrics and events to.
    to_addr: Option<String>,
    /// A namespace to prefix all metrics with, joined with a '.'.
    namespace: Option<String>,
    /// Default tags to include with every request.
    default_tags: Vec<String>,
    /// OPTIONAL, if defined, will use UDS instead of UDP and will ignore UDP options
    socket_path: Option<String>,
    /// OPTIONAL, if defined, will utilize batching for sending metrics
    batching_options: Option<BatchingOptions>,
}

impl OptionsBuilder {
    /// Create a new `OptionsBuilder` struct.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new();
    /// ```
    pub fn new() -> Self {
        Self::default()
    }

    /// Will allow the builder to generate an `Options` struct with the provided value.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new().from_addr(String::from("127.0.0.1:9000"));
    /// ```
    pub fn from_addr(&mut self, from_addr: String) -> &mut OptionsBuilder {
        self.from_addr = Some(from_addr);
        self
    }

    /// Will allow the builder to generate an `Options` struct with the provided value.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new().to_addr(String::from("127.0.0.1:9001"));
    /// ```
    pub fn to_addr(&mut self, to_addr: String) -> &mut OptionsBuilder {
        self.to_addr = Some(to_addr);
        self
    }

    /// Will allow the builder to generate an `Options` struct with the provided value.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new().namespace(String::from("mynamespace"));
    /// ```
    pub fn namespace(&mut self, namespace: String) -> &mut OptionsBuilder {
        self.namespace = Some(namespace);
        self
    }

    /// Will allow the builder to generate an `Options` struct with the provided value. Can be called multiple times to add multiple `default_tags` to the `Options`.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new().default_tag(String::from("tag1:tav1val")).default_tag(String::from("tag2:tag2val"));
    /// ```
    pub fn default_tag(&mut self, default_tag: String) -> &mut OptionsBuilder {
        self.default_tags.push(default_tag);
        self
    }

    /// Will allow the builder to generate an `Options` struct with the provided value.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///
    ///   let options_builder = OptionsBuilder::new().default_tag(String::from("tag1:tav1val")).default_tag(String::from("tag2:tav2val"));
    /// ```
    pub fn socket_path(&mut self, socket_path: Option<String>) -> &mut OptionsBuilder {
        self.socket_path = socket_path;
        self
    }

    /// Will allow the builder to generate an `Options` struct with the provided value.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{ OptionsBuilder, BatchingOptions };
    ///   use std::time::Duration;
    ///
    ///   let options_builder = OptionsBuilder::new().batching_options(BatchingOptions { max_buffer_size: 8000, max_time: Duration::from_millis(3000), max_retry_attempts: 3, initial_retry_delay: 10 });
    /// ```
    pub fn batching_options(&mut self, batching_options: BatchingOptions) -> &mut OptionsBuilder {
        self.batching_options = Some(batching_options);
        self
    }

    /// Will construct an `Options` with all of the provided values and fall back to the default values if they aren't provided.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::OptionsBuilder;
    ///   use dogstatsd::Options;
    ///
    ///   let options = OptionsBuilder::new().namespace(String::from("mynamespace")).default_tag(String::from("tag1:tav1val")).build();
    ///
    ///   assert_eq!(
    ///       Options {
    ///           from_addr: "0.0.0.0:0".into(),
    ///           to_addr: "127.0.0.1:8125".into(),
    ///           namespace: String::from("mynamespace"),
    ///           default_tags: vec!(String::from("tag1:tav1val")),
    ///           socket_path: None,
    ///           batching_options: None,
    ///       },
    ///       options
    ///   )
    /// ```
    pub fn build(&self) -> Options {
        Options::new(
            self.from_addr
                .as_ref()
                .unwrap_or(&String::from(DEFAULT_FROM_ADDR)),
            self.to_addr
                .as_ref()
                .unwrap_or(&String::from(DEFAULT_TO_ADDR)),
            self.namespace.as_ref().unwrap_or(&String::default()),
            self.default_tags.to_vec(),
            self.socket_path.clone(),
            self.batching_options,
        )
    }
}

#[derive(Debug)]
enum SocketType {
    Udp(UdpSocket),
    #[cfg(unix)]
    Uds(UnixDatagram),
    BatchableUdp(Mutex<Sender<batch_processor::Message>>),
    #[cfg(unix)]
    BatchableUds(Mutex<Sender<batch_processor::Message>>),
}

/// The client struct that handles sending metrics to the Dogstatsd server.
#[derive(Debug)]
pub struct Client {
    socket: SocketType,
    from_addr: String,
    to_addr: String,
    namespace: String,
    default_tags: Vec<u8>,
}

impl PartialEq for Client {
    fn eq(&self, other: &Self) -> bool {
        // Ignore `socket`, which will never be the same
        self.from_addr == other.from_addr
            && self.to_addr == other.to_addr
            && self.namespace == other.namespace
            && self.default_tags == other.default_tags
    }
}

impl Drop for Client {
    fn drop(&mut self) {
        match &self.socket {
            SocketType::BatchableUdp(tx_channel) => {
                let _ = tx_channel
                    .lock()
                    .unwrap()
                    .send(batch_processor::Message::Shutdown);
            }
            #[cfg(unix)]
            SocketType::BatchableUds(tx_channel) => {
                let _ = tx_channel
                    .lock()
                    .unwrap()
                    .send(batch_processor::Message::Shutdown);
            }
            _ => {}
        }
    }
}

impl Client {
    /// Create a new client from an options struct.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    /// ```
    pub fn new(options: Options) -> Result<Self, DogstatsdError> {
        let fn_create_tx_channel = |socket: SocketType,
                                    batching_options: BatchingOptions,
                                    to_addr: String,
                                    socket_path: Option<String>|
         -> Mutex<Sender<batch_processor::Message>> {
            let (tx, rx) = mpsc::channel();
            thread::spawn(move || {
                batch_processor::process_events(batching_options, to_addr, socket, socket_path, rx);
            });
            Mutex::from(tx)
        };

        let socket = if options.socket_path.is_some() {
            #[cfg(unix)]
            {
                let socket_path = options.socket_path.clone().expect("checked is_some above");

                // The follow scenarios can occur:
                // - socket does not exist yet: We will call .bind(...) to create one
                // - socket exists, but no listener: We will retry attempting to connect
                //   however, if no listener subscribes to the socket within retries, we will
                //   fail to initialize
                // - socket exists, with a listener: Calling .connect(...) will work successfully
                let mut uds_socket = UnixDatagram::unbound()?;
                match uds_socket.connect(socket_path.clone()) {
                    Ok(socket) => socket,
                    Err(e) => {
                        println!(
                            "Couldn't connect to uds socket.. attempting to re-create by binding directly: {e:?}"
                        );
                        uds_socket = UnixDatagram::bind(socket_path.clone())?;
                    }
                };
                uds_socket.set_nonblocking(true)?;

                let wrapped_socket = SocketType::Uds(uds_socket);
                if let Some(batching_options) = options.batching_options {
                    SocketType::BatchableUds(fn_create_tx_channel(
                        wrapped_socket,
                        batching_options,
                        options.to_addr.clone(),
                        Some(socket_path),
                    ))
                } else {
                    wrapped_socket
                }
            }
            #[cfg(not(unix))]
            {
                return Err(DogstatsdError::from(std::io::Error::new(
                    std::io::ErrorKind::Unsupported,
                    "Unix domain sockets are not supported on this platform",
                )));
            }
        } else {
            let wrapped_socket = SocketType::Udp(UdpSocket::bind(&options.from_addr)?);
            if let Some(batching_options) = options.batching_options {
                SocketType::BatchableUdp(fn_create_tx_channel(
                    wrapped_socket,
                    batching_options,
                    options.to_addr.clone(),
                    None,
                ))
            } else {
                wrapped_socket
            }
        };

        let default_tags = Options::merge_with_system_tags(options.default_tags);

        Ok(Client {
            socket,
            from_addr: options.from_addr,
            to_addr: options.to_addr,
            namespace: options.namespace,
            default_tags: default_tags.join(",").into_bytes(),
        })
    }

    /// Increment a StatsD counter
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.incr("counter", &["tag:counter"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn incr<'a, I, S, T>(&self, stat: S, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&CountMetric::Incr(stat.into().as_ref(), 1), tags)
    }

    /// Increment a StatsD counter by the provided amount
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.incr_by_value("counter", 123, &["tag:counter"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn incr_by_value<'a, I, S, T>(&self, stat: S, value: i64, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&CountMetric::Incr(stat.into().as_ref(), value), tags)
    }

    /// Decrement a StatsD counter
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.decr("counter", &["tag:counter"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn decr<'a, I, S, T>(&self, stat: S, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&CountMetric::Decr(stat.into().as_ref(), 1), tags)
    }

    /// Decrement a StatsD counter by the provided amount
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.decr_by_value("counter", 23, &["tag:counter"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn decr_by_value<'a, I, S, T>(&self, stat: S, value: i64, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&CountMetric::Decr(stat.into().as_ref(), value), tags)
    }

    /// Make an arbitrary change to a StatsD counter
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.count("counter", 42, &["tag:counter"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn count<'a, I, S, T>(&self, stat: S, count: i64, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&CountMetric::Arbitrary(stat.into().as_ref(), count), tags)
    }

    /// Time how long it takes for a block of code to execute.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///   use std::thread;
    ///   use std::time::Duration;
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.time("timer", &["tag:time"], || {
    ///       thread::sleep(Duration::from_millis(200))
    ///   }).unwrap_or_else(|(_, e)| println!("Encountered error: {}", e))
    /// ```
    pub fn time<'a, F, O, I, S, T>(
        &self,
        stat: S,
        tags: I,
        block: F,
    ) -> Result<O, (O, DogstatsdError)>
    where
        F: FnOnce() -> O,
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        let start_time = Utc::now();
        let output = block();
        let end_time = Utc::now();
        let stat = stat.into();
        let metric = TimeMetric::new(stat.as_ref(), &start_time, &end_time);
        match self.send(&metric, tags) {
            Ok(()) => Ok(output),
            Err(error) => Err((output, error)),
        }
    }

    /// Time how long it takes for an async block of code to execute.
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///   use std::thread;
    ///   use std::time::Duration;
    ///
    /// # async fn do_work() {}
    ///   async fn timer() {
    ///       let client = Client::new(Options::default()).unwrap();
    ///       client.async_time("timer", &["tag:time"], do_work)
    ///       .await
    ///       .unwrap_or_else(|(_, e)| println!("Encountered error: {}", e))
    ///   }
    /// ```
    pub async fn async_time<'a, Fn, Fut, O, I, S, T>(
        &self,
        stat: S,
        tags: I,
        block: Fn,
    ) -> Result<O, (O, DogstatsdError)>
    where
        Fn: FnOnce() -> Fut,
        Fut: Future<Output = O>,
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        let start_time = Utc::now();
        let output = block().await;
        let end_time = Utc::now();
        let stat = stat.into();
        match self.send(
            &TimeMetric::new(stat.as_ref(), &start_time, &end_time),
            tags,
        ) {
            Ok(()) => Ok(output),
            Err(error) => Err((output, error)),
        }
    }

    /// Send your own timing metric in milliseconds
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.timing("timing", 350, &["tag:timing"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn timing<'a, I, S, T>(&self, stat: S, ms: i64, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(&TimingMetric::new(stat.into().as_ref(), ms), tags)
    }

    /// Report an arbitrary value as a gauge
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.gauge("gauge", "12345", &["tag:gauge"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn gauge<'a, I, S, SS, T>(&self, stat: S, val: SS, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(
            &GaugeMetric::new(stat.into().as_ref(), val.into().as_ref()),
            tags,
        )
    }

    /// Report a value in a histogram
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.histogram("histogram", "67890", &["tag:histogram"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn histogram<'a, I, S, SS, T>(&self, stat: S, val: SS, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(
            &HistogramMetric::new(stat.into().as_ref(), val.into().as_ref()),
            tags,
        )
    }

    /// Report a value in a distribution
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.distribution("distribution", "67890", &["tag:distribution"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn distribution<'a, I, S, SS, T>(&self, stat: S, val: SS, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(
            &DistributionMetric::new(stat.into().as_ref(), val.into().as_ref()),
            tags,
        )
    }

    /// Report a value in a set
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.set("set", "13579", &["tag:set"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn set<'a, I, S, SS, T>(&self, stat: S, val: SS, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(
            &SetMetric::new(stat.into().as_ref(), val.into().as_ref()),
            tags,
        )
    }

    /// Report the status of a service
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options, ServiceStatus, ServiceCheckOptions};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.service_check("redis.can_connect", ServiceStatus::OK, &["tag:service"], None)
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    ///
    ///   let options = ServiceCheckOptions {
    ///     hostname: Some("my-host.localhost"),
    ///     ..Default::default()
    ///   };
    ///   client.service_check("redis.can_connect", ServiceStatus::OK, &["tag:service"], Some(options))
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    ///
    ///   let all_options = ServiceCheckOptions {
    ///     hostname: Some("my-host.localhost"),
    ///     timestamp: Some(1510326433),
    ///     message: Some("Message about the check or service")
    ///   };
    ///   client.service_check("redis.can_connect", ServiceStatus::OK, &["tag:service"], Some(all_options))
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    /// ```
    pub fn service_check<'a, I, S, T>(
        &self,
        stat: S,
        val: ServiceStatus,
        tags: I,
        options: Option<ServiceCheckOptions>,
    ) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        let unwrapped_options = options.unwrap_or_default();
        self.send(
            &ServiceCheck::new(stat.into().as_ref(), val, unwrapped_options),
            tags,
        )
    }

    /// Send a custom event as a title and a body
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   client.event("Event Title", "Event Body", &["tag:event"])
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    ///
    /// ```
    pub fn event<'a, I, S, SS, T>(&self, title: S, text: SS, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        self.send(
            &Event::new(title.into().as_ref(), text.into().as_ref()),
            tags,
        )
    }

    /// Send a custom event as a title and a body
    ///
    /// # Examples
    ///
    /// ```
    ///   use dogstatsd::{Client, Options, EventOptions, EventAlertType, EventPriority};
    ///
    ///   let client = Client::new(Options::default()).unwrap();
    ///   let event_options = EventOptions::new()
    ///     .with_timestamp(1638480000)
    ///     .with_hostname("localhost")
    ///     .with_priority(EventPriority::Normal)
    ///     .with_alert_type(EventAlertType::Error);
    ///   client.event_with_options("My Custom Event Title", "My Custom Event Body", &["tag:event"], Some(event_options))
    ///       .unwrap_or_else(|e| println!("Encountered error: {}", e));
    pub fn event_with_options<'a, I, S, SS, T>(
        &self,
        title: S,
        text: SS,
        tags: I,
        options: Option<EventOptions<'a>>,
    ) -> DogstatsdResult
    where
        I: IntoIterator<Item = T>,
        S: Into<Cow<'a, str>>,
        SS: Into<Cow<'a, str>>,
        T: AsRef<str>,
    {
        let title_owned = title.into();
        let text_owned = text.into();
        let mut event = Event::new(title_owned.as_ref(), text_owned.as_ref());

        // Apply additional options if provided
        if let Some(options) = options {
            if let Some(timestamp) = options.timestamp {
                event = event.with_timestamp(timestamp);
            }
            if let Some(hostname) = options.hostname {
                event = event.with_hostname(hostname);
            }
            if let Some(aggregation_key) = options.aggregation_key {
                event = event.with_aggregation_key(aggregation_key);
            }
            if let Some(priority) = options.priority {
                event = event.with_priority(priority);
            }
            if let Some(source_type_name) = options.source_type_name {
                event = event.with_source_type_name(source_type_name);
            }
            if let Some(alert_type) = options.alert_type {
                event = event.with_alert_type(alert_type);
            }
        }

        self.send(&event, tags)
    }

    fn send<I, M, S>(&self, metric: &M, tags: I) -> DogstatsdResult
    where
        I: IntoIterator<Item = S>,
        M: Metric,
        S: AsRef<str>,
    {
        let formatted_metric = format_for_send(metric, &self.namespace, tags, &self.default_tags);
        match &self.socket {
            SocketType::Udp(socket) => {
                socket.send_to(formatted_metric.as_slice(), &self.to_addr)?;
            }
            #[cfg(unix)]
            SocketType::Uds(socket) => {
                socket.send(formatted_metric.as_slice())?;
            }
            SocketType::BatchableUdp(tx_channel) => {
                tx_channel
                    .lock()
                    .expect("Mutex poisoned...")
                    .send(batch_processor::Message::Data(formatted_metric))
                    .unwrap_or_else(|error| {
                        println!("Exception occurred when writing to channel: {:?}", error);
                    });
            }
            #[cfg(unix)]
            SocketType::BatchableUds(tx_channel) => {
                tx_channel
                    .lock()
                    .expect("Mutex poisoned...")
                    .send(batch_processor::Message::Data(formatted_metric))
                    .unwrap_or_else(|error| {
                        println!("Exception occurred when writing to channel: {:?}", error);
                    });
            }
        }
        Ok(())
    }
}

/// Configuration options for an `Event`.
///
/// `EventOptions` provides additional optional metadata that can be attached
/// to an event, enabling greater flexibility and contextual information
/// for event handling and monitoring systems.
///
/// # Example
///
/// ```rust
/// use dogstatsd::{EventOptions, EventAlertType, EventPriority};
///
/// let options = EventOptions {
///     timestamp: Some(1638480000),
///     hostname: Some("localhost"),
///     aggregation_key: Some("service_down"),
///     priority: Some(EventPriority::Normal),
///     source_type_name: Some("monitoring"),
///     alert_type: Some(EventAlertType::Error),
/// };
/// ```
///
#[derive(Default, Clone, Copy, Debug)]
pub struct EventOptions<'a> {
    /// Optional Unix timestamp representing the event time. The default is the current Unix epoch timestamp.
    pub timestamp: Option<u64>,
    /// Optional hostname associated with the event.
    pub hostname: Option<&'a str>,
    /// Optional key for grouping related events.
    pub aggregation_key: Option<&'a str>,
    /// Optional priority level of the event, e.g., `"low"` or `"normal"`.
    pub priority: Option<EventPriority>,
    /// Optional source type name of the event, e.g., `"monitoring"`.
    pub source_type_name: Option<&'a str>,
    /// Optional alert type for the event, e.g., `"error"`, `"warning"`,  `"info"`, `"success"`. Default `"info"`.
    pub alert_type: Option<EventAlertType>,
}

impl<'a> EventOptions<'a> {
    /// Creates a new `EventOptions` instance with all fields set to `None`.
    pub fn new() -> Self {
        EventOptions {
            timestamp: None,
            hostname: None,
            aggregation_key: None,
            priority: None,
            source_type_name: None,
            alert_type: None,
        }
    }
    /// Sets the `hostname` for the event.
    pub fn with_timestamp(mut self, timestamp: u64) -> Self {
        self.timestamp = Some(timestamp);
        self
    }

    /// Sets the `hostname` for the event.
    pub fn with_hostname(mut self, hostname: &'a str) -> Self {
        self.hostname = Some(hostname);
        self
    }

    /// Sets the `aggregation_key` for the event.
    pub fn with_aggregation_key(mut self, aggregation_key: &'a str) -> Self {
        self.aggregation_key = Some(aggregation_key);
        self
    }

    /// Sets the `priority` for the event.
    pub fn with_priority(mut self, priority: EventPriority) -> Self {
        self.priority = Some(priority);
        self
    }

    /// Sets the `source_type_name` for the event.
    pub fn with_source_type_name(mut self, source_type_name: &'a str) -> Self {
        self.source_type_name = Some(source_type_name);
        self
    }

    /// Sets the `alert_type` for the event.
    pub fn with_alert_type(mut self, alert_type: EventAlertType) -> Self {
        self.alert_type = Some(alert_type);
        self
    }
}

mod batch_processor {
    use std::sync::mpsc::{Receiver, RecvTimeoutError};
    use std::time::{Duration, Instant};

    use retry::{delay::jitter, delay::Exponential, retry};

    use crate::{BatchingOptions, SocketType};

    pub(crate) enum Message {
        Data(Vec<u8>),
        Shutdown,
    }

    fn send_to_socket_with_retries(
        batching_options: &BatchingOptions,
        socket: &SocketType,
        data: &Vec<u8>,
        to_addr: &String,
        _socket_path: &Option<String>,
    ) {
        retry(
            Exponential::from_millis(batching_options.initial_retry_delay)
                .map(jitter)
                .take(batching_options.max_retry_attempts),
            || -> Result<(), std::io::Error> {
                match socket {
                    SocketType::Udp(socket) => {
                        socket.send_to(data.as_slice(), to_addr)?;
                    }
                    #[cfg(unix)]
                    SocketType::Uds(socket) => {
                        if let Err(error) = socket.send(data.as_slice()) {
                            // Per https://doc.rust-lang.org/stable/std/os/unix/net/struct.UnixDatagram.html#method.send
                            // If send fails, it is due to a connection issue, so just attempt
                            // to reconnect
                            let socket_path_unwrapped = _socket_path
                                .as_ref()
                                .expect("Only invoked if socket path is defined.");
                            socket.connect(socket_path_unwrapped)?;

                            return Err(error);
                        }
                    }
                    SocketType::BatchableUdp(_tx_channel) => {
                        panic!("Logic Error - socket type should not be batchable.");
                    }
                    #[cfg(unix)]
                    SocketType::BatchableUds(_tx_channel) => {
                        panic!("Logic Error - socket type should not be batchable.");
                    }
                }

                Ok(())
            },
        )
        .unwrap_or_else(|error| {
            println!(
                "Failed to send within retry policy... Dropping metrics: {:?}",
                error
            )
        });
    }

    fn flush_buffer(
        batching_options: &BatchingOptions,
        socket: &SocketType,
        buffer: &mut Vec<u8>,
        to_addr: &String,
        socket_path: &Option<String>,
    ) {
        if buffer.is_empty() {
            return;
        }

        send_to_socket_with_retries(batching_options, socket, buffer, to_addr, socket_path);
        buffer.clear();
    }

    pub(crate) fn process_events(
        batching_options: BatchingOptions,
        to_addr: String,
        socket: SocketType,
        socket_path: Option<String>,
        rx: Receiver<Message>,
    ) {
        let mut buffer: Vec<u8> = vec![];
        let mut batch_started_at: Option<Instant> = None;

        loop {
            let message = match batch_started_at {
                Some(started_at) => match rx
                    .recv_timeout(remaining_batch_time(started_at, batching_options.max_time))
                {
                    Ok(message) => Ok(message),
                    Err(RecvTimeoutError::Timeout) => {
                        flush_buffer(
                            &batching_options,
                            &socket,
                            &mut buffer,
                            &to_addr,
                            &socket_path,
                        );
                        batch_started_at = None;
                        continue;
                    }
                    Err(RecvTimeoutError::Disconnected) => Err(RecvTimeoutError::Disconnected),
                },
                None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected),
            };

            match message {
                Ok(Message::Data(data)) => {
                    let next_buffer_size = buffer.len() + data.len() + 1;

                    if !buffer.is_empty() && next_buffer_size > batching_options.max_buffer_size {
                        flush_buffer(
                            &batching_options,
                            &socket,
                            &mut buffer,
                            &to_addr,
                            &socket_path,
                        );
                        batch_started_at = None;
                    }

                    if buffer.is_empty() {
                        batch_started_at = Some(Instant::now());
                    }

                    buffer.extend(data);
                    buffer.push(b'\n');

                    if buffer.len() >= batching_options.max_buffer_size {
                        flush_buffer(
                            &batching_options,
                            &socket,
                            &mut buffer,
                            &to_addr,
                            &socket_path,
                        );
                        batch_started_at = None;
                    }
                }
                Ok(Message::Shutdown) => {
                    flush_buffer(
                        &batching_options,
                        &socket,
                        &mut buffer,
                        &to_addr,
                        &socket_path,
                    );
                    break;
                }
                Err(e) => {
                    println!("Exception occurred when reading from channel: {:?}", e);
                    break;
                }
            }
        }
    }

    fn remaining_batch_time(started_at: Instant, max_time: Duration) -> Duration {
        max_time
            .checked_sub(started_at.elapsed())
            .unwrap_or(Duration::ZERO)
    }
}

#[cfg(test)]
mod tests {
    use metrics::GaugeMetric;
    use serial_test::serial;

    use super::*;

    #[test]
    fn test_options_default() {
        let options = Options::default();
        let expected_options = Options {
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            ..Default::default()
        };

        assert_eq!(expected_options, options)
    }

    #[test]
    fn test_options_builder_none() {
        let options = OptionsBuilder::new().build();
        let expected_options = Options {
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            ..Default::default()
        };

        assert_eq!(expected_options, options);
    }

    #[test]
    fn teset_options_builder_all() {
        let options = OptionsBuilder::new()
            .from_addr("127.0.0.2:0".into())
            .to_addr("127.0.0.2:8125".into())
            .namespace("mynamespace".into())
            .default_tag(String::from("tag1:tag1val"))
            .build();
        let expected_options = Options {
            from_addr: "127.0.0.2:0".into(),
            to_addr: "127.0.0.2:8125".into(),
            namespace: "mynamespace".into(),
            default_tags: ["tag1:tag1val".into()].to_vec(),
            socket_path: None,
            batching_options: None,
        };

        assert_eq!(expected_options, options);
    }

    #[test]
    #[serial]
    fn test_new() {
        let client = Client::new(Options::default()).unwrap();
        let expected_client = Client {
            socket: SocketType::Udp(UdpSocket::bind(DEFAULT_FROM_ADDR).unwrap()),
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            default_tags: String::new().into_bytes(),
        };

        assert_eq!(expected_client, client)
    }

    #[test]
    #[serial]
    fn test_new_default_tags() {
        let options = Options::new(
            DEFAULT_FROM_ADDR,
            DEFAULT_TO_ADDR,
            "",
            vec![String::from("tag1:tag1val")],
            None,
            None,
        );
        let client = Client::new(options).unwrap();
        let expected_client = Client {
            socket: SocketType::Udp(UdpSocket::bind(DEFAULT_FROM_ADDR).unwrap()),
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            default_tags: String::from("tag1:tag1val").into_bytes(),
        };

        assert_eq!(expected_client, client)
    }

    #[test]
    #[serial]
    fn test_system_tags() {
        let options = Options::new(
            DEFAULT_FROM_ADDR,
            DEFAULT_TO_ADDR,
            "",
            vec![String::from("tag1:tag1val"), String::from("version:0.0.2")],
            None,
            None,
        );

        let client = with_default_system_tags(|| Client::new(options).unwrap());

        dbg!(String::from_utf8_lossy(client.default_tags.as_ref()));

        let expected_client = Client {
            socket: SocketType::Udp(UdpSocket::bind(DEFAULT_FROM_ADDR).unwrap()),
            from_addr: DEFAULT_FROM_ADDR.into(),
            to_addr: DEFAULT_TO_ADDR.into(),
            namespace: String::new(),
            default_tags: String::from("tag1:tag1val,version:0.0.2,env:production,service:service")
                .into_bytes(),
        };

        assert_eq!(expected_client, client)
    }

    #[test]
    fn test_send() {
        let options = Options::new("127.0.0.1:9001", "127.0.0.1:9002", "", vec![], None, None);
        let client = Client::new(options).unwrap();
        // Shouldn't panic or error
        client
            .send(&GaugeMetric::new("gauge", "1234"), ["tag1", "tag2"])
            .unwrap();
    }

    fn with_default_system_tags<T, F: FnOnce() -> T>(f: F) -> T {
        unsafe {
            std::env::set_var("DD_ENV", "production");
            std::env::set_var("DD_SERVICE", "service");
            std::env::set_var("DD_VERSION", "0.0.1");
        }
        let t = f();
        unsafe {
            std::env::remove_var("DD_ENV");
            std::env::remove_var("DD_SERVICE");
            std::env::remove_var("DD_VERSION");
        }
        t
    }
}

#[cfg(all(feature = "unstable", test))]
mod bench {
    extern crate test;

    use super::*;

    use self::test::Bencher;

    #[bench]
    fn bench_incr(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = &["name1:value1"];
        b.iter(|| {
            client.incr("bench.incr", tags).unwrap();
        })
    }

    #[bench]
    fn bench_decr(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = &["name1:value1"];
        b.iter(|| {
            client.decr("bench.decr", tags).unwrap();
        })
    }

    #[bench]
    fn bench_count(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = &["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client.count("bench.count", i, tags).unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_timing(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = &["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client.timing("bench.timing", i, tags).unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_gauge(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client.gauge("bench.guage", &i.to_string(), &tags).unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_histogram(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client
                .histogram("bench.histogram", &i.to_string(), &tags)
                .unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_distribution(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client
                .distribution("bench.distribution", &i.to_string(), &tags)
                .unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_set(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let mut i = 0;
        b.iter(|| {
            client.set("bench.set", &i.to_string(), &tags).unwrap();
            i += 1;
        })
    }

    #[bench]
    fn bench_service_check(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let all_options = ServiceCheckOptions {
            hostname: Some("my-host.localhost"),
            timestamp: Some(1510326433),
            message: Some("Message about the check or service"),
        };
        b.iter(|| {
            client
                .service_check(
                    "bench.service_check",
                    ServiceStatus::Critical,
                    &tags,
                    Some(all_options),
                )
                .unwrap();
        })
    }

    #[bench]
    fn bench_event(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        b.iter(|| {
            client
                .event("Test Event Title", "Test Event Message", &tags, None)
                .unwrap();
        })
    }

    fn bench_event_options(b: &mut Bencher) {
        let options = Options::default();
        let client = Client::new(options).unwrap();
        let tags = vec!["name1:value1"];
        let event_options = EventOptions::new()
            .with_timestamp(1638480000)
            .with_hostname("localhost")
            .with_priority(EventPriority::Normal)
            .with_alert_type(EventAlertType::Error);

        b.iter(|| {
            client
                .event(
                    "Test Event Title",
                    "Test Event Message",
                    &tags,
                    Some(event_options),
                )
                .unwrap();
        })
    }
}