StatsD and Graphite

use statsd.qweruse statsd_connect, statsd_count from statsd      // Quillo 0.6: only these

// A shop's metrics sent to StatsD with DogStatsD tags, events and service checks, buffered and// flushed once a second; the same over TCP and to Graphite; and a small StatsD server of its own,// taking packets with udp.listen and reading them with statsd_parse.////   STATSD_URL=udp://127.0.0.1:8125 quillo run statsd.qwe      then open http://localhost:8080/order//// Needs something that takes StatsD: Telegraf's statsd input (datadog_extensions = true), the// Datadog agent, or statsd itself. STATSD_TCP_URL and GRAPHITE_URL (graphite://host:2003) for// /tcp; STATSD_LISTEN_PORT is the port this program listens on itself (9125).use "statsd.qwer"let opened = [    statsd_connect("app", env("STATSD_URL", "udp://127.0.0.1:8125"), { flavor: "dogstatsd", prefix: "shop.", tags: { env: env("APP_ENV", "dev") } }),    statsd_connect("tcp", env("STATSD_TCP_URL", "tcp://127.0.0.1:8125"), { prefix: "shop.", buffer: 0 }),    statsd_connect("graphite", env("GRAPHITE_URL", "graphite://127.0.0.1:2003"), { prefix: "shop." }),]// What the buffer holds goes out once a second, packed into as few packets as fit.every(1000, () => statsd_flush("app"))route GET /order ? { total: float = 42.5, country: str = "FR", customer: str = "c-1001" } {    let started = now()    let lines = [        statsd_count("app", "orders", 1, { tags: { country } })?,        statsd_histogram("app", "order.total", total, { tags: { country } })?,        statsd_set("app", "customers", customer, {})?,        statsd_gauge("app", "carts.open", 7, {})?,        statsd_gauge("app", "stock.sku42", -1, { delta: true })?,        statsd_timing("app", "checkout.ms", now() - started, { rate: 1.0 })?,    ]    { lines, buffered: statsd_stats("app")?.buffered, sent: statsd_flush("app")? }}route GET /deploy {    let event = statsd_event("app", "Deployed 1.4.2", "by ci\nall checks green", { alert_type: "success", aggregation_key: "deploys", tags: ["service:shop"] })?    let check = statsd_service_check("app", "db.reachable", "ok", { message: "replica lag 20 ms" })?    statsd_flush("app")?    { event, check }}// Over TCP, one line per metric, and to Graphite (buffered: flushed here).route GET /tcp {    statsd_count("tcp", "emails.sent", 3, {})?    statsd_timing("tcp", "email.ms", 118.5, {})?    statsd_gauge("graphite", "queue.depth", 12, {})?    statsd_count("graphite", "orders", 1, { tags: { country: "FR" } })?    statsd_flush("graphite")?    { tcp: statsd_stats("tcp")?, graphite: statsd_stats("graphite")?, line: graphite_line("shop.queue.depth", 12, 1761800000, {})? }}// Lines without sending them, in each flavor.route GET /lines = {    statsd: statsd_line("count", "page.views", 1, { rate: 0.25 })?,    dogstatsd: statsd_line("timing", "db.query", 3.75, { flavor: "dogstatsd", tags: { table: "orders" } })?,    graphite: graphite_line("shop.orders", 1, 1761800000, { country: "FR" })?,    no_tags_in_plain: statsd_line("count", "x", 1, { tags: { a: "b" } }).error,}// A StatsD server of its own: what comes in on the port is read and kept (the last 200 metrics).let received = []udp.listen(int(env("STATSD_LISTEN_PORT", "9125")), (p) => {    let all = received + statsd_parse(p.text ?? "")    received = slice(all, -200, len(all))})route GET /loopback {    statsd_connect("self", `udp://127.0.0.1:${env("STATSD_LISTEN_PORT", "9125")}`, { flavor: "dogstatsd" })?    statsd_count("self", "signups", 2, { tags: { plan: "pro" } })?    statsd_gauge("self", "temperature", 21.5, {})?    statsd_service_check("self", "worker", 2, { message: "stuck" })?    statsd_close("self")?    sleep(300)    { received: received | filter((m) => m.metric == "signups" || m.metric == "temperature" || m.type == "service_check") }}

//! StatsD, DogStatsD and Graphite metrics sent over UDP or TCP: counters, gauges, timers, histograms, distributions and sets with sample rates and tags, DogStatsD events and service checks, a client-side buffer flushed when a packet is full or when asked, and StatsD lines read back for a program that receives them.//! version: 1.0.0//!//! [dependencies]use serde_json::{json, Value};use std::collections::HashMap;use std::io::Write;use std::net::{SocketAddr, TcpStream, ToSocketAddrs, UdpSocket};use std::sync::Mutex;use std::time::Duration;#[derive(Clone, Copy, PartialEq)]enum Flavor {    Statsd,    Dog,    Graphite,}enum Wire {    Udp(UdpSocket),    Tcp(Option<TcpStream>),}struct Client {    url: String,    host: String,    port: u16,    addr: SocketAddr,    flavor: Flavor,    wire: Wire,    prefix: String,    tags: Vec<(String, String)>,    max: usize,    buf: Vec<String>,    buf_len: usize,    timeout: Duration,    lines: u64,    packets: u64,    sampled_out: u64,    errors: u64,    last_error: Option<String>,}/// The clients, by the name given to `statsd_connect`.static CLIENTS: Mutex<Option<HashMap<String, Client>>> = Mutex::new(None);fn with<T>(name: &str, f: impl FnOnce(&mut Client) -> Result<T, String>) -> Result<T, String> {    let mut g = CLIENTS.lock().unwrap_or_else(|e| e.into_inner());    let map = g.get_or_insert_with(HashMap::new);    match map.get_mut(name) {        Some(c) => f(c),        None => Err(format!("no StatsD client named `{}`: call statsd_connect(\"{}\", \"udp://127.0.0.1:8125\", {{}}) first", name, name)),    }}fn resolve(host: &str, port: u16) -> Result<SocketAddr, String> {    let h = host.trim_start_matches('[').trim_end_matches(']');    let addrs: Vec<SocketAddr> = (h, port).to_socket_addrs().map_err(|e| format!("{}: {}", host, e))?.collect();    addrs.iter().find(|a| a.is_ipv4()).or(addrs.first()).cloned().ok_or(format!("{} has no address", host))}/// `udp://host:8125`, `tcp://host:8125`, `graphite://host:2003` (TCP), `graphite+udp://host:2003`.fn parse_url(url: &str) -> Result<(bool, Option<Flavor>, String, u16), String> {    let (scheme, rest) = url.split_once("://").ok_or(format!("`{}` is not a URL like udp://127.0.0.1:8125", url))?;    let (tcp, flavor, default_port) = match scheme.to_ascii_lowercase().as_str() {        "udp" | "statsd" => (false, None, 8125),        "tcp" => (true, None, 8125),        "graphite" | "graphite+tcp" => (true, Some(Flavor::Graphite), 2003),        "graphite+udp" => (false, Some(Flavor::Graphite), 2003),        s => return Err(format!("`{}`: the scheme is udp://, tcp://, graphite:// or graphite+udp://, not {}://", url, s)),    };    let rest = rest.trim_end_matches('/');    let (host, port) = if let Some(r) = rest.strip_prefix('[') {        let (h, p) = r.split_once(']').ok_or(format!("`{}`: an IPv6 address is closed with ]", url))?;        (h.to_string(), p.trim_start_matches(':').to_string())    } else {        match rest.rsplit_once(':') {            Some((h, p)) => (h.to_string(), p.to_string()),            None => (rest.to_string(), String::new()),        }    };    if host.is_empty() {        return Err(format!("`{}` has no host", url));    }    let port = if port.is_empty() { default_port } else { port.parse::<u16>().map_err(|_| format!("`{}`: the port `{}` is not a number from 1 to 65535", url, port))? };    Ok((tcp, flavor, host, port))}fn tag_list(v: &Value) -> Result<Vec<(String, String)>, String> {    let mut out = vec![];    match v {        Value::Null => {}        Value::Object(m) => {            for (k, x) in m {                let val = match x {                    Value::String(s) => s.clone(),                    Value::Null => String::new(),                    other => other.to_string(),                };                out.push((clean_tag(k), clean_tag(&val)));            }        }        Value::Array(a) => {            for x in a {                let s = x.as_str().ok_or("tags are a map { env: \"prod\" } or a list [\"env:prod\"]")?;                match s.split_once(':') {                    Some((k, v)) => out.push((clean_tag(k), clean_tag(v))),                    None => out.push((clean_tag(s), String::new())),                }            }        }        _ => return Err("tags are a map { env: \"prod\" } or a list [\"env:prod\"]".into()),    }    Ok(out)}fn clean_tag(s: &str) -> String {    s.chars().map(|c| if matches!(c, '|' | ',' | '#' | '\n' | '\r' | ';' | '=') { '_' } else { c }).collect()}fn clean_metric(s: &str, flavor: Flavor) -> String {    s.chars()        .map(|c| match flavor {            Flavor::Graphite if c.is_whitespace() || c == ';' => '_',            _ if matches!(c, ':' | '|' | '@' | '#' | '\n' | '\r') || c.is_whitespace() => '_',            _ => c,        })        .collect()}fn number(v: &Value, what: &str) -> Result<String, String> {    match v {        Value::Number(n) if n.is_i64() || n.is_u64() => Ok(n.to_string()),        Value::Number(n) => {            let f = n.as_f64().unwrap_or(0.0);            if !f.is_finite() {                return Err(format!("the {} is not a finite number", what));            }            Ok(format!("{}", f))        }        Value::String(s) if s.trim().parse::<f64>().map(|f| f.is_finite()).unwrap_or(false) => Ok(s.trim().to_string()),        _ => Err(format!("the {} is a number, not {}", what, v)),    }}fn rate_of(o: &Value) -> Result<f64, String> {    match &o["rate"] {        Value::Null => Ok(1.0),        v => {            let r = v.as_f64().ok_or("the rate is a number between 0 and 1, like 0.1")?;            if !(r > 0.0 && r <= 1.0) {                return Err(format!("the rate is between 0 (not included) and 1, not {}", r));            }            Ok(r)        }    }}fn flavor_of(s: &str) -> Result<Flavor, String> {    match s.to_ascii_lowercase().as_str() {        "" | "statsd" | "etsy" => Ok(Flavor::Statsd),        "dogstatsd" | "datadog" | "dog" => Ok(Flavor::Dog),        "graphite" | "carbon" => Ok(Flavor::Graphite),        x => Err(format!("the flavor is \"statsd\", \"dogstatsd\" or \"graphite\", not \"{}\"", x)),    }}fn unix_now() -> i64 {    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs() as i64).unwrap_or(0)}/// A random number in [0, 1), for sample rates: xorshift seeded by the clock.fn chance() -> f64 {    static STATE: Mutex<u64> = Mutex::new(0);    let mut s = STATE.lock().unwrap_or_else(|e| e.into_inner());    if *s == 0 {        *s = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(88172645463325252) | 1;    }    let mut x = *s;    x ^= x << 13;    x ^= x >> 7;    x ^= x << 17;    *s = x;    (x >> 11) as f64 / (1u64 << 53) as f64}/// One line in the client's flavor. `kind` is c, g, ms, h, d or s.fn format_line(flavor: Flavor, prefix: &str, global: &[(String, String)], kind: &str, metric: &str, value: &Value, o: &Value) -> Result<String, String> {    if metric.trim().is_empty() {        return Err("the metric has no name".into());    }    let name = clean_metric(&format!("{}{}", prefix, metric), flavor);    let mut tags = global.to_vec();    tags.extend(tag_list(&o["tags"])?);    let rate = rate_of(o)?;    let val = if kind == "s" {        match value {            Value::String(s) => clean_metric(s, Flavor::Statsd),            v => number(v, "set member")?,        }    } else {        let n = number(value, "value")?;        if kind == "g" && o["delta"].as_bool() == Some(true) && !n.starts_with('-') { format!("+{}", n) } else { n }    };    if flavor == Flavor::Graphite {        if kind == "s" {            return Err("Graphite has no sets: count or gauge the value instead".into());        }        let mut path = name;        for (k, v) in &tags {            path.push_str(&format!(";{}={}", k, if v.is_empty() { "true" } else { v }));        }        let ts = o["timestamp"].as_i64().unwrap_or_else(unix_now);        return Ok(format!("{} {} {}", path, val, ts));    }    if kind == "d" && flavor != Flavor::Dog {        return Err("distributions are DogStatsD's: connect with { flavor: \"dogstatsd\" }, or use statsd_histogram".into());    }    let mut line = format!("{}:{}|{}", name, val, kind);    if rate < 1.0 {        line.push_str(&format!("|@{}", rate));    }    if !tags.is_empty() {        if flavor != Flavor::Dog {            return Err("plain StatsD has no tags: connect with { flavor: \"dogstatsd\" } (Telegraf and Datadog read them), or put them in the name".into());        }        line.push_str("|#");        line.push_str(&tags.iter().map(|(k, v)| if v.is_empty() { k.clone() } else { format!("{}:{}", k, v) }).collect::<Vec<_>>().join(","));    }    if let Some(ts) = o["timestamp"].as_i64() {        if flavor == Flavor::Dog {            line.push_str(&format!("|T{}", ts));        }    }    Ok(line)}fn tags_suffix(tags: &[(String, String)]) -> String {    if tags.is_empty() {        String::new()    } else {        format!("|#{}", tags.iter().map(|(k, v)| if v.is_empty() { k.clone() } else { format!("{}:{}", k, v) }).collect::<Vec<_>>().join(","))    }}fn one_line(s: &str) -> String {    s.replace("\r\n", "\\n").replace('\n', "\\n").replace('\r', "")}impl Client {    fn send_raw(&mut self, payload: &str) -> Result<(), String> {        let r = match &mut self.wire {            Wire::Udp(sock) => sock.send_to(payload.as_bytes(), self.addr).map(|_| ()).map_err(|e| e.to_string()),            Wire::Tcp(slot) => {                let mut tries = 0;                loop {                    tries += 1;                    if slot.is_none() {                        let addr = resolve(&self.host, self.port).unwrap_or(self.addr);                        match TcpStream::connect_timeout(&addr, self.timeout) {                            Ok(s) => {                                let _ = s.set_write_timeout(Some(self.timeout));                                let _ = s.set_nodelay(true);                                *slot = Some(s);                            }                            Err(e) => break Err(format!("cannot reach {} at {}: {}", self.url, addr, e)),                        }                    }                    let mut data = payload.to_string();                    data.push('\n');                    match slot.as_mut().unwrap().write_all(data.as_bytes()) {                        Ok(()) => break Ok(()),                        Err(e) => {                            *slot = None;                            if tries >= 2 {                                break Err(format!("{}: {}", self.url, e));                            }                        }                    }                }            }        };        match &r {            Ok(()) => self.packets += 1,            Err(e) => {                self.errors += 1;                self.last_error = Some(e.clone());            }        }        r    }    fn flush(&mut self) -> Result<(u64, u64), String> {        if self.buf.is_empty() {            return Ok((0, 0));        }        let lines = std::mem::take(&mut self.buf);        self.buf_len = 0;        let n = lines.len() as u64;        self.send_raw(&lines.join("\n"))?;        self.lines += n;        Ok((n, 1))    }    fn push(&mut self, line: String) -> Result<(), String> {        if self.max == 0 || line.len() >= self.max {            self.flush()?;            self.send_raw(&line)?;            self.lines += 1;            return Ok(());        }        let extra = if self.buf.is_empty() { line.len() } else { line.len() + 1 };        if self.buf_len + extra > self.max {            self.flush()?;            self.buf_len = line.len();        } else {            self.buf_len += extra;        }        self.buf.push(line);        Ok(())    }}fn metric(name: &str, kind: &str, metric: &str, value: &Value, options: &Value) -> Result<Option<String>, String> {    with(name, |c| {        let line = format_line(c.flavor, &c.prefix, &c.tags, kind, metric, value, options)?;        if rate_of(options)? < 1.0 && chance() >= rate_of(options)? {            c.sampled_out += 1;            return Ok(None);        }        c.push(line.clone())?;        Ok(Some(line))    })}/// Opens (or replaces) the client `name` to `url`: `udp://host:8125`, `tcp://host:8125`, `graphite://host:2003` (plaintext over TCP) or `graphite+udp://host:2003`./// Options: `flavor` ("statsd", "dogstatsd" or "graphite"), `prefix` ("shop."), `tags` (sent with every metric: a map or a list), `buffer` (bytes per packet; 1432 for UDP, 8192 for TCP; 0 sends each metric at once), `timeout_ms` (TCP, 2000). TCP connects on the first send, and again after an error.pub fn statsd_connect(name: &str, url: &str, options: serde_json::Value) -> Result<Value, String> {    let (tcp, url_flavor, host, port) = parse_url(url)?;    let flavor = match options["flavor"].as_str() {        Some(f) => flavor_of(f)?,        None => url_flavor.unwrap_or(Flavor::Statsd),    };    let addr = resolve(&host, port)?;    let tags = tag_list(&options["tags"])?;    if !tags.is_empty() && flavor == Flavor::Statsd {        return Err("plain StatsD has no tags: add { flavor: \"dogstatsd\" } to send them".into());    }    let max = match &options["buffer"] {        Value::Null => if tcp { 8192 } else { 1432 },        v => v.as_i64().filter(|n| (0..=65_000).contains(n)).ok_or("the buffer is a number of bytes from 0 (no buffer) to 65000")? as usize,    };    let timeout = Duration::from_millis(options["timeout_ms"].as_i64().unwrap_or(2000).clamp(50, 60_000) as u64);    let wire = if tcp {        Wire::Tcp(None)    } else {        let bind = if addr.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" };        Wire::Udp(UdpSocket::bind(bind).map_err(|e| format!("cannot open a UDP socket: {}", e))?)    };    let client = Client {        url: url.to_string(),        host,        port,        addr,        flavor,        wire,        prefix: options["prefix"].as_str().unwrap_or("").to_string(),        tags,        max,        buf: vec![],        buf_len: 0,        timeout,        lines: 0,        packets: 0,        sampled_out: 0,        errors: 0,        last_error: None,    };    let mut g = CLIENTS.lock().unwrap_or_else(|e| e.into_inner());    let map = g.get_or_insert_with(HashMap::new);    if let Some(mut old) = map.remove(name) {        let _ = old.flush();    }    map.insert(name.to_string(), client);    Ok(json!({        "name": name, "url": url, "address": addr.to_string(), "transport": if tcp { "tcp" } else { "udp" },        "flavor": match flavor { Flavor::Statsd => "statsd", Flavor::Dog => "dogstatsd", Flavor::Graphite => "graphite" },        "buffer": max,    }))}/// Adds `value` to the counter `metric`. Options: `rate` (0.1 sends one in ten, marked `|@0.1` so the server scales it back), `tags`. Gives the line, or null when the sample rate left it out.pub fn statsd_count(name: &str, metric: &str, value: i64, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "c", metric, &json!(value), &options)}/// Sets the gauge `metric` to `value` (int or float); `{ delta: true }` moves it by `value` instead (`+3`, `-2`). Options: `tags`, `rate`.pub fn statsd_gauge(name: &str, metric: &str, value: serde_json::Value, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "g", metric, &value, &options)}/// Records a time of `ms` milliseconds for `metric` (the server keeps its count, mean, percentiles). Options: `rate`, `tags`.pub fn statsd_timing(name: &str, metric: &str, ms: serde_json::Value, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "ms", metric, &ms, &options)}/// Records `value` in the histogram `metric` (`|h`: DogStatsD's histogram, a timer to Etsy's StatsD). Options: `rate`, `tags`.pub fn statsd_histogram(name: &str, metric: &str, value: serde_json::Value, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "h", metric, &value, &options)}/// Records `value` in the distribution `metric` (DogStatsD's `|d`, aggregated across hosts by Datadog). Options: `rate`, `tags`.pub fn statsd_distribution(name: &str, metric: &str, value: serde_json::Value, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "d", metric, &value, &options)}/// Adds `member` to the set `metric`: the server counts the distinct members it saw in each interval (unique visitors). Options: `tags`.pub fn statsd_set(name: &str, metric: &str, member: &str, options: serde_json::Value) -> Result<Option<String>, String> {    self::metric(name, "s", metric, &json!(member), &options)}/// Sends a DogStatsD event (`_e{..}:title|text`). Options: `timestamp` (Unix seconds), `hostname`, `aggregation_key`, `priority` ("normal" or "low"), `source_type_name`, `alert_type` ("error", "warning", "info" or "success"), `tags`. Gives the line.pub fn statsd_event(name: &str, title: &str, text: &str, options: serde_json::Value) -> Result<String, String> {    with(name, |c| {        if c.flavor != Flavor::Dog {            return Err("events are DogStatsD's: connect with { flavor: \"dogstatsd\" }".into());        }        if title.is_empty() {            return Err("an event needs a title".into());        }        let (t, x) = (one_line(title), one_line(text));        let mut line = format!("_e{{{},{}}}:{}|{}", t.len(), x.len(), t, x);        if let Some(ts) = options["timestamp"].as_i64() {            line.push_str(&format!("|d:{}", ts));        }        for (key, code) in [("hostname", "h"), ("aggregation_key", "k"), ("priority", "p"), ("source_type_name", "s"), ("alert_type", "t")] {            if let Some(v) = options[key].as_str() {                if key == "priority" && !matches!(v, "normal" | "low") {                    return Err(format!("the priority is \"normal\" or \"low\", not \"{}\"", v));                }                if key == "alert_type" && !matches!(v, "error" | "warning" | "info" | "success") {                    return Err(format!("the alert_type is \"error\", \"warning\", \"info\" or \"success\", not \"{}\"", v));                }                line.push_str(&format!("|{}:{}", code, clean_tag(v)));            }        }        let mut tags = c.tags.clone();        tags.extend(tag_list(&options["tags"])?);        line.push_str(&tags_suffix(&tags));        c.push(line.clone())?;        Ok(line)    })}/// Sends a DogStatsD service check (`_sc|name|status`): `status` is 0 to 3 or "ok", "warning", "critical", "unknown". Options: `timestamp`, `hostname`, `message`, `tags`. Gives the line.pub fn statsd_service_check(name: &str, check: &str, status: serde_json::Value, options: serde_json::Value) -> Result<String, String> {    let code = match &status {        Value::Number(n) => n.as_i64().filter(|s| (0..=3).contains(s)),        Value::String(s) => ["ok", "warning", "critical", "unknown"].iter().position(|x| x.eq_ignore_ascii_case(s)).map(|i| i as i64),        _ => None,    }    .ok_or(format!("the status is 0 to 3, or \"ok\", \"warning\", \"critical\" or \"unknown\", not {}", status))?;    with(name, |c| {        if c.flavor != Flavor::Dog {            return Err("service checks are DogStatsD's: connect with { flavor: \"dogstatsd\" }".into());        }        let mut line = format!("_sc|{}|{}", clean_metric(&format!("{}{}", c.prefix, check), Flavor::Dog), code);        if let Some(ts) = options["timestamp"].as_i64() {            line.push_str(&format!("|d:{}", ts));        }        if let Some(h) = options["hostname"].as_str() {            line.push_str(&format!("|h:{}", clean_tag(h)));        }        let mut tags = c.tags.clone();        tags.extend(tag_list(&options["tags"])?);        line.push_str(&tags_suffix(&tags));        // The message comes last, and `m:` must not appear in it.        if let Some(m) = options["message"].as_str() {            line.push_str(&format!("|m:{}", one_line(m).replace("m:", "m\\:")));        }        c.push(line.clone())?;        Ok(line)    })}/// Sends what the client's buffer holds now, in one packet: `{ lines, packets }`. Call it from an `every` tick, after a batch, or before the program stops.pub fn statsd_flush(name: &str) -> Result<Value, String> {    with(name, |c| {        let (lines, packets) = c.flush()?;        Ok(json!({ "lines": lines, "packets": packets }))    })}/// What the client has done: `{ lines, packets, buffered, sampled_out, errors, last_error }`. Over UDP, a line sent is not a line received: nothing comes back.pub fn statsd_stats(name: &str) -> Result<Value, String> {    with(name, |c| {        Ok(json!({            "url": c.url, "lines": c.lines, "packets": c.packets, "buffered": c.buf.len(),            "sampled_out": c.sampled_out, "errors": c.errors, "last_error": c.last_error,        }))    })}/// Flushes the client and forgets it; whether there was one.pub fn statsd_close(name: &str) -> Result<bool, String> {    let mut g = CLIENTS.lock().unwrap_or_else(|e| e.into_inner());    match g.get_or_insert_with(HashMap::new).remove(name) {        Some(mut c) => {            c.flush()?;            Ok(true)        }        None => Ok(false),    }}/// One metric's line without sending it: `kind` is "count", "gauge", "timing", "histogram", "distribution" or "set". Options as the senders take, plus `flavor` and `prefix`.pub fn statsd_line(kind: &str, metric: &str, value: serde_json::Value, options: serde_json::Value) -> Result<String, String> {    let k = match kind {        "count" | "c" | "counter" => "c",        "gauge" | "g" => "g",        "timing" | "ms" | "timer" => "ms",        "histogram" | "h" => "h",        "distribution" | "d" => "d",        "set" | "s" => "s",        _ => return Err(format!("the kind is count, gauge, timing, histogram, distribution or set, not `{}`", kind)),    };    let flavor = flavor_of(options["flavor"].as_str().unwrap_or(""))?;    format_line(flavor, options["prefix"].as_str().unwrap_or(""), &[], k, metric, &value, &options)}/// A Graphite plaintext line, `path value timestamp`: `timestamp` 0 for now; `tags` (a map) are written Graphite 1.1's way, `path;key=value`.pub fn graphite_line(path: &str, value: serde_json::Value, timestamp: i64, tags: serde_json::Value) -> Result<String, String> {    let mut o = json!({ "tags": tags });    if timestamp > 0 {        o["timestamp"] = json!(timestamp);    }    format_line(Flavor::Graphite, "", &[], "g", path, &value, &o)}fn parse_one(line: &str) -> Result<Value, String> {    if let Some(rest) = line.strip_prefix("_e{") {        let (lens, body) = rest.split_once("}:").ok_or("an event is _e{title length,text length}:title|text")?;        let (tl, xl) = lens.split_once(',').ok_or("an event is _e{title length,text length}:title|text")?;        let tl: usize = tl.parse().map_err(|_| "an event's title length is a number")?;        let xl: usize = xl.parse().map_err(|_| "an event's text length is a number")?;        let b = body.as_bytes();        if b.len() < tl + 1 + xl || b.get(tl) != Some(&b'|') {            return Err("an event's lengths do not match its title and text".into());        }        let title = String::from_utf8_lossy(&b[..tl]).to_string();        let text = String::from_utf8_lossy(&b[tl + 1..tl + 1 + xl]).to_string();        let mut ev = json!({ "type": "event", "title": title, "text": text.replace("\\n", "\n"), "tags": {} });        for part in String::from_utf8_lossy(&b[tl + 1 + xl..]).split('|').filter(|p| !p.is_empty()) {            if let Some(t) = part.strip_prefix('#') {                ev["tags"] = parse_tags(t);            } else if let Some((k, v)) = part.split_once(':') {                let key = match k { "d" => "timestamp", "h" => "hostname", "k" => "aggregation_key", "p" => "priority", "s" => "source_type_name", "t" => "alert_type", o => o };                ev[key] = if k == "d" { v.parse::<i64>().map(Value::from).unwrap_or(json!(v)) } else { json!(v) };            }        }        return Ok(ev);    }    if let Some(rest) = line.strip_prefix("_sc|") {        let mut parts = rest.split('|');        let name = parts.next().unwrap_or("");        let status: i64 = parts.next().and_then(|s| s.parse().ok()).filter(|s| (0..=3).contains(s)).ok_or("a service check's status is 0 to 3")?;        let mut sc = json!({ "type": "service_check", "name": name, "status": status, "tags": {} });        for part in parts {            if let Some(t) = part.strip_prefix('#') {                sc["tags"] = parse_tags(t);            } else if let Some(m) = part.strip_prefix("m:") {                sc["message"] = json!(m.replace("m\\:", "m:"));            } else if let Some(d) = part.strip_prefix("d:") {                sc["timestamp"] = json!(d.parse::<i64>().unwrap_or(0));            } else if let Some(h) = part.strip_prefix("h:") {                sc["hostname"] = json!(h);            }        }        return Ok(sc);    }    let (name, rest) = line.split_once(':').ok_or("a StatsD line is name:value|type")?;    let mut parts = rest.split('|');    let value = parts.next().unwrap_or("");    let kind = parts.next().ok_or("a StatsD line is name:value|type")?;    let kind_word = match kind { "c" => "count", "g" => "gauge", "ms" => "timing", "h" => "histogram", "d" => "distribution", "s" => "set", o => return Err(format!("`{}` is not a StatsD type (c, g, ms, h, d, s)", o)) };    let mut out = json!({ "type": kind_word, "metric": name, "rate": 1.0, "tags": {} });    if kind == "s" {        out["value"] = json!(value);    } else {        let f: f64 = value.parse().map_err(|_| format!("`{}` is not a number", value))?;        out["value"] = if f.fract() == 0.0 && f.abs() < 9e15 && !value.contains('.') { json!(f as i64) } else { json!(f) };        if kind == "g" && (value.starts_with('+') || value.starts_with('-')) {            out["delta"] = json!(true);        }    }    for part in parts {        if let Some(r) = part.strip_prefix('@') {            out["rate"] = json!(r.parse::<f64>().map_err(|_| format!("`{}` is not a sample rate", r))?);        } else if let Some(t) = part.strip_prefix('#') {            out["tags"] = parse_tags(t);        } else if let Some(t) = part.strip_prefix('T') {            out["timestamp"] = json!(t.parse::<i64>().unwrap_or(0));        } else if let Some(c) = part.strip_prefix("c:") {            out["container"] = json!(c);        }    }    Ok(out)}fn parse_tags(t: &str) -> Value {    let mut m = serde_json::Map::new();    for tag in t.split(',').filter(|x| !x.is_empty()) {        match tag.split_once(':') {            Some((k, v)) => m.insert(k.to_string(), json!(v)),            None => m.insert(tag.to_string(), json!("")),        };    }    Value::Object(m)}/// Reads a StatsD or DogStatsD packet (one line or many), as a server receives it, for a program that takes metrics itself with `udp.listen`: a list of `{ type, metric, value, rate, tags }`, events and service checks too. A line that cannot be read is `{ type: "error", line, error }`.pub fn statsd_parse(packet: &str) -> Vec<Value> {    packet        .split('\n')        .map(|l| l.trim_end_matches('\r'))        .filter(|l| !l.trim().is_empty())        .map(|l| parse_one(l).unwrap_or_else(|e| json!({ "type": "error", "line": l, "error": e })))        .collect()}