tempest 0.1.1

Realtime message handling framework inspired by Apache Storm and built with Actix
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

use crate::metric::backend::prelude::*;
use std::net::{ToSocketAddrs, UdpSocket};

/// The backend for writing metrics
/// to a Statsd server
#[allow(dead_code)]
pub(crate) struct Statsd {
    addr: String,
    prefix: Option<String>,
    socket: UdpSocket,
    target: MetricTarget,
    name_delimiter: Format,
    label_separator: Format,
    label_delimiter: Format,
}

impl Statsd {
    pub(crate) fn new(target: MetricTarget) -> BackendResult<Self> {
        let socket = match UdpSocket::bind("0.0.0.0:0") {
            Ok(_socket) => _socket,
            Err(err) => return Err(BackendError::from_io(err)),
        };

        let mut _prefix = None;
        let mut addr = "".to_string();
        match &target {
            MetricTarget::Statsd { host, prefix } => {
                addr = host.to_string();
                _prefix = prefix.clone();
            }
            _ => {}
        }

        // Try converting this address once
        // socket.send_to will automatically
        // do this each time, before sending
        &addr
            .to_socket_addrs()?
            .next()
            .ok_or_else(|| BackendError::from_addr())?;

        Ok(Statsd {
            addr: addr,
            prefix: _prefix,
            socket: socket,
            target: target,
            name_delimiter: PERIOD,
            label_separator: EMPTY,
            label_delimiter: EMPTY,
        })
    }

    fn send(&self, metrics: &str) {
        match self.socket.send_to(&metrics.as_bytes(), &self.addr) {
            Ok(bytes) => {
                trace!("statd Ok({}) bytes written", bytes,);
            }
            Err(err) => {
                error!("statd error {:?}", err);
            }
        }
    }
}

impl Backend for Statsd {
    fn write(&mut self, mut msg: Msg) {
        // Build the metric name
        // Add the config override
        let mut name = vec![];
        // Add the prefix
        if let Some(p) = &self.prefix {
            name.push(p.to_string());
        }
        // Add the root prefix
        name.push(msg.root_prefix);

        // just clone the met
        if msg.metrics.names.len() > 0 {
            name.append(&mut msg.metrics.names.clone());
        }
        let name = name.join(&self.name_delimiter);

        let mut out = String::new();
        let iter_mut = msg.metrics.bucket.map.iter_mut();
        let mut i = 0;
        let len = iter_mut.len();
        for (_key, metric) in iter_mut {
            let key = vec![name.clone(), metric.names.join(&self.name_delimiter)]
                .join(&self.name_delimiter);

            out.push_str(&key);
            out.push_str(COLON);
            if let FormatedMetric::Statsd(v) = &metric.to_value(MetricFormat::Statsd) {
                out.push_str(v);
            };
            out.push_str(PIPE);
            out.push_str(&metric.kind.as_statsd_str());
            i += 1;
            if i < len {
                out.push_str(LBR);
            }
        }
        trace!("statsd send: {:?}", &out);
        self.send(&out);
    }
}

/// StatsdActor
///
pub(crate) struct StatsdActor {
    pub statsd: Statsd,
}
impl StatsdActor {}

impl Actor for StatsdActor {
    type Context = Context<Self>;
    fn started(&mut self, _ctx: &mut Context<Self>) {}
}

impl Handler<Msg> for StatsdActor {
    type Result = ();

    fn handle(&mut self, msg: Msg, _ctx: &mut Context<Self>) {
        self.statsd.write(msg);
    }
}