mirror of https://github.com/fralalonde/dipstick
Context -> Scope
This commit is contained in:
parent
4c4b82a9da
commit
0143cea743
|
@ -0,0 +1,7 @@
|
|||
# Latest changes + history
|
||||
|
||||
## version 0.7.0
|
||||
|
||||
- Add `Delegate` mechanism to allow runtime (re)configuration of metrics
|
||||
- Enhance macros to allow metrics of different types within a single block
|
||||
|
|
@ -113,7 +113,7 @@ Metric definition macros are just `lazy_static!` wrappers.
|
|||
Where necessary, metrics can be defined _ad-hoc_:
|
||||
```rust,skt-run
|
||||
let user_name = "john_day";
|
||||
let app_metrics = app_metrics(to_log().with_cache(512));
|
||||
let app_metrics = app_metrics(to_log()).with_cache(512);
|
||||
app_metrics.gauge(format!("gauge_for_user_{}", user_name)).value(44);
|
||||
```
|
||||
Defining a cache is optional but will speed up re-definition of common ad-hoc metrics.
|
||||
|
|
|
@ -9,34 +9,34 @@ use std::thread::sleep;
|
|||
use dipstick::*;
|
||||
|
||||
fn main() {
|
||||
let metrics = to_stdout();
|
||||
|
||||
let counter = metrics.counter("counter_a");
|
||||
let timer = metrics.timer("timer_a");
|
||||
let gauge = metrics.gauge("gauge_a");
|
||||
let marker = metrics.marker("marker_a");
|
||||
let context = to_buffered_stdout();
|
||||
|
||||
loop {
|
||||
// add counts forever, non-stop
|
||||
println!("\n------- open scope");
|
||||
|
||||
let ref mut scope = metrics.open_scope(true);
|
||||
let metrics = context.open_scope();
|
||||
|
||||
counter.count(scope, 11);
|
||||
counter.count(scope, 12);
|
||||
counter.count(scope, 13);
|
||||
let counter = metrics.counter("counter_a");
|
||||
let timer = metrics.timer("timer_a");
|
||||
let gauge = metrics.gauge("gauge_a");
|
||||
let marker = metrics.marker("marker_a");
|
||||
|
||||
counter.count(11);
|
||||
counter.count(12);
|
||||
counter.count(13);
|
||||
|
||||
timer.interval_us(scope, 11_000_000);
|
||||
timer.interval_us(scope, 12_000_000);
|
||||
timer.interval_us(scope, 13_000_000);
|
||||
timer.interval_us(11_000_000);
|
||||
timer.interval_us(12_000_000);
|
||||
timer.interval_us(13_000_000);
|
||||
|
||||
sleep(Duration::from_millis(1000));
|
||||
|
||||
gauge.value(scope, 11);
|
||||
gauge.value(scope, 12);
|
||||
gauge.value(scope, 13);
|
||||
gauge.value(11);
|
||||
gauge.value(12);
|
||||
gauge.value(13);
|
||||
|
||||
marker.mark(scope);
|
||||
marker.mark();
|
||||
|
||||
sleep(Duration::from_millis(1000));
|
||||
|
|
@ -11,9 +11,9 @@ fn main() {
|
|||
|
||||
pub fn raw_write() {
|
||||
// setup dual metric channels
|
||||
let metrics_log = to_log();
|
||||
let metrics_log = to_log().open_scope();
|
||||
|
||||
// define and send metrics using raw channel API
|
||||
let counter = metrics_log.define_metric(Kind::Counter, "count_a", FULL_SAMPLING_RATE);
|
||||
metrics_log.open_scope(true).write(&counter, 1);
|
||||
metrics_log.write(&counter, 1);
|
||||
}
|
||||
|
|
|
@ -117,34 +117,31 @@ mod bench {
|
|||
use super::*;
|
||||
use test;
|
||||
use core::Kind::*;
|
||||
use output::*;
|
||||
|
||||
#[bench]
|
||||
fn aggregate_marker(b: &mut test::Bencher) {
|
||||
let sink = aggregate(summary, to_void());
|
||||
let sink: AppMetrics<Aggregate> = aggregate(summary, to_void()).into();
|
||||
let metric = sink.define_metric(Marker, "event_a", 1.0);
|
||||
let scope = sink.open_scope(false);
|
||||
b.iter(|| test::black_box(scope.write(&metric, 1)));
|
||||
b.iter(|| test::black_box(sink.write(&metric, 1)));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn aggregate_counter(b: &mut test::Bencher) {
|
||||
let sink = aggregate(summary, to_void());
|
||||
let sink: AppMetrics<Aggregate> = aggregate(summary, to_void()).into();
|
||||
let metric = sink.define_metric(Counter, "count_a", 1.0);
|
||||
let scope = sink.open_scope(false);
|
||||
b.iter(|| test::black_box(scope.write(&metric, 1)));
|
||||
b.iter(|| test::black_box(sink.write(&metric, 1)));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn reset_marker(b: &mut test::Bencher) {
|
||||
let sink = aggregate(summary, to_void());
|
||||
let sink: AppMetrics<Aggregate> = aggregate(summary, to_void()).into();
|
||||
let metric = sink.define_metric(Marker, "marker_a", 1.0);
|
||||
b.iter(|| test::black_box(metric.reset()));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn reset_counter(b: &mut test::Bencher) {
|
||||
let sink = aggregate(summary, to_void());
|
||||
let sink: AppMetrics<Aggregate> = aggregate(summary, to_void()).into();
|
||||
let metric = sink.define_metric(Counter, "count_a", 1.0);
|
||||
b.iter(|| test::black_box(metric.reset()));
|
||||
}
|
||||
|
|
|
@ -165,9 +165,9 @@ mod bench {
|
|||
|
||||
use super::*;
|
||||
use test;
|
||||
use core::Kind::*;
|
||||
use aggregate::*;
|
||||
use publish::*;
|
||||
use output::*;
|
||||
|
||||
#[bench]
|
||||
fn dispatch_marker_to_aggregate(b: &mut test::Bencher) {
|
||||
|
|
|
@ -58,10 +58,7 @@ pub struct AppCounter<M> {
|
|||
|
||||
impl<M> AppCounter<M> {
|
||||
/// Record a value count.
|
||||
pub fn count<V>(&self, count: V)
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
pub fn count<V: ToPrimitive>(&self, count: V) {
|
||||
self.scope.write(&self.metric, count.to_u64().unwrap());
|
||||
}
|
||||
}
|
||||
|
@ -77,10 +74,7 @@ pub struct AppGauge<M> {
|
|||
|
||||
impl<M> AppGauge<M> {
|
||||
/// Record a value point for this gauge.
|
||||
pub fn value<V>(&self, value: V)
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
pub fn value<V: ToPrimitive>(&self, value: V) {
|
||||
self.scope.write(&self.metric, value.to_u64().unwrap());
|
||||
}
|
||||
}
|
||||
|
@ -102,12 +96,8 @@ pub struct AppTimer<M> {
|
|||
impl<M> AppTimer<M> {
|
||||
/// Record a microsecond interval for this timer
|
||||
/// Can be used in place of start()/stop() if an external time interval source is used
|
||||
pub fn interval_us<V>(&self, interval_us: V) -> V
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
self.scope
|
||||
.write(&self.metric, interval_us.to_u64().unwrap());
|
||||
pub fn interval_us<V: ToPrimitive>(&self, interval_us: V) -> V {
|
||||
self.scope.write(&self.metric, interval_us.to_u64().unwrap());
|
||||
interval_us
|
||||
}
|
||||
|
||||
|
@ -168,8 +158,9 @@ impl<M> AppMetrics<M>
|
|||
where
|
||||
M: Clone + Send + Sync + 'static,
|
||||
{
|
||||
/// Define a raw metric.
|
||||
#[inline]
|
||||
fn define_metric(&self, kind: Kind, name: &str, rate: Rate) -> M {
|
||||
pub fn define_metric(&self, kind: Kind, name: &str, rate: Rate) -> M {
|
||||
(self.define_metric_fn)(kind, name, rate)
|
||||
}
|
||||
|
||||
|
@ -222,6 +213,12 @@ where
|
|||
let scope = self.single_scope.clone();
|
||||
schedule(period, move || scope.flush())
|
||||
}
|
||||
|
||||
/// Record a raw metric value.
|
||||
pub fn write(&self, metric: &M, value: Value) {
|
||||
self.single_scope.write(metric, value);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
//// Dispatch / Receiver impl
|
||||
|
|
|
@ -47,9 +47,9 @@ impl<M: Send + Sync + Clone + 'static> WithAsyncQueue for LocalMetrics<M> {
|
|||
}
|
||||
});
|
||||
|
||||
Arc::new(move |buffered| {
|
||||
Arc::new(move || {
|
||||
// open next scope, make it Arc to move across queue
|
||||
let next_scope: ControlScopeFn<M> = next(buffered);
|
||||
let next_scope: ControlScopeFn<M> = next();
|
||||
let sender = sender.clone();
|
||||
|
||||
// forward any scope command through the channel
|
||||
|
|
|
@ -63,7 +63,7 @@ pub enum Kind {
|
|||
pub type DefineMetricFn<M> = Arc<Fn(Kind, &str, Rate) -> M + Send + Sync>;
|
||||
|
||||
/// A function trait that opens a new metric capture scope.
|
||||
pub type OpenScopeFn<M> = Arc<Fn(bool) -> ControlScopeFn<M> + Send + Sync>;
|
||||
pub type OpenScopeFn<M> = Arc<Fn() -> ControlScopeFn<M> + Send + Sync>;
|
||||
|
||||
/// A function trait that writes to or flushes a certain scope.
|
||||
pub type ControlScopeFn<M> = Arc<InnerControlScopeFn<M>>;
|
||||
|
@ -112,7 +112,7 @@ impl<M> InnerControlScopeFn<M> {
|
|||
/// Write a value to this scope.
|
||||
///
|
||||
/// ```rust
|
||||
/// let ref mut scope = dipstick::to_log().open_scope(false);
|
||||
/// let scope = dipstick::to_log().open_scope();
|
||||
/// scope.write(&"counter".to_string(), 6);
|
||||
/// ```
|
||||
///
|
||||
|
@ -121,10 +121,11 @@ impl<M> InnerControlScopeFn<M> {
|
|||
(self.scope_fn)(Write(metric, value))
|
||||
}
|
||||
|
||||
/// Flush this scope, if buffered.
|
||||
/// Flush this scope.
|
||||
/// Has no effect if scope is unbuffered.
|
||||
///
|
||||
/// ```rust
|
||||
/// let ref mut scope = dipstick::to_log().open_scope(true);
|
||||
/// let scope = dipstick::to_log().open_scope();
|
||||
/// scope.flush();
|
||||
/// ```
|
||||
///
|
||||
|
|
109
src/graphite.rs
109
src/graphite.rs
|
@ -22,6 +22,21 @@ app_metrics!{
|
|||
}
|
||||
}
|
||||
|
||||
// TODO enable fine config
|
||||
//struct GraphiteConfig<ADDR = ToSocketAddrs + Debug + Clone> {
|
||||
// to_socket_address: ADDR,
|
||||
// buffer_bytes: Option<usize>,
|
||||
//}
|
||||
//
|
||||
//impl From<ToSocketAddrs> for GraphiteConfig {
|
||||
// fn from<ADDR: ToSocketAddrs + Debug + Clone>(addr: ADDR) -> GraphiteConfig {
|
||||
// GraphiteConfig {
|
||||
// to_socket_address: addr,
|
||||
// buffer_bytes: None,
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
|
||||
/// Send metrics to a graphite server at the address and port provided.
|
||||
pub fn to_graphite<ADDR>(address: ADDR) -> error::Result<LocalMetrics<Graphite>>
|
||||
where
|
||||
|
@ -29,45 +44,64 @@ where
|
|||
{
|
||||
debug!("Connecting to graphite {:?}", address);
|
||||
let socket = Arc::new(RwLock::new(RetrySocket::new(address.clone())?));
|
||||
Ok(LocalMetrics::new(
|
||||
move |kind, name, rate| {
|
||||
let mut prefix = String::with_capacity(32);
|
||||
prefix.push_str(name);
|
||||
prefix.push(' ');
|
||||
|
||||
let mut scale = match kind {
|
||||
// timers are in µs, lets give graphite milliseconds for consistency with statsd
|
||||
Kind::Timer => 1000,
|
||||
_ => 1,
|
||||
};
|
||||
|
||||
if rate < FULL_SAMPLING_RATE {
|
||||
// graphite does not do sampling, so we'll upsample before sending
|
||||
let upsample = (1.0 / rate).round() as u64;
|
||||
warn!(
|
||||
"Metric {:?} '{}' being sampled at rate {} will be upsampled \
|
||||
by a factor of {} when sent to graphite.",
|
||||
kind, name, rate, upsample
|
||||
);
|
||||
scale *= upsample;
|
||||
}
|
||||
|
||||
Graphite { prefix, scale }
|
||||
},
|
||||
move |buffered| {
|
||||
let buf = ScopeBuffer {
|
||||
buffer: Arc::new(RwLock::new(String::new())),
|
||||
socket: socket.clone(),
|
||||
buffered,
|
||||
};
|
||||
control_scope(move |cmd| match cmd {
|
||||
ScopeCmd::Write(metric, value) => buf.write(metric, value),
|
||||
ScopeCmd::Flush => buf.flush(),
|
||||
})
|
||||
},
|
||||
Ok(metrics_context(
|
||||
move |kind, name, rate| graphite_metric(kind, name, rate),
|
||||
move || graphite_scope(&socket, false),
|
||||
))
|
||||
}
|
||||
|
||||
/// Send metrics to a graphite server at the address and port provided.
|
||||
pub fn to_buffered_graphite<ADDR>(address: ADDR) -> error::Result<LocalMetrics<Graphite>>
|
||||
where
|
||||
ADDR: ToSocketAddrs + Debug + Clone,
|
||||
{
|
||||
debug!("Connecting to graphite {:?}", address);
|
||||
let socket = Arc::new(RwLock::new(RetrySocket::new(address.clone())?));
|
||||
|
||||
Ok(metrics_context(
|
||||
move |kind, name, rate| graphite_metric(kind, name, rate),
|
||||
move || graphite_scope(&socket, true),
|
||||
))
|
||||
}
|
||||
|
||||
fn graphite_metric(kind: Kind, name: &str, rate: Rate) -> Graphite {
|
||||
let mut prefix = String::with_capacity(32);
|
||||
prefix.push_str(name);
|
||||
prefix.push(' ');
|
||||
|
||||
let mut scale = match kind {
|
||||
// timers are in µs, lets give graphite milliseconds for consistency with statsd
|
||||
Kind::Timer => 1000,
|
||||
_ => 1,
|
||||
};
|
||||
|
||||
if rate < FULL_SAMPLING_RATE {
|
||||
// graphite does not do sampling, so we'll upsample before sending
|
||||
let upsample = (1.0 / rate).round() as u64;
|
||||
warn!(
|
||||
"Metric {:?} '{}' being sampled at rate {} will be upsampled \
|
||||
by a factor of {} when sent to graphite.",
|
||||
kind, name, rate, upsample
|
||||
);
|
||||
scale *= upsample;
|
||||
}
|
||||
|
||||
Graphite { prefix, scale }
|
||||
}
|
||||
|
||||
fn graphite_scope(socket: &Arc<RwLock<RetrySocket>>, buffered: bool) -> ControlScopeFn<Graphite> {
|
||||
let buf = ScopeBuffer {
|
||||
buffer: Arc::new(RwLock::new(String::new())),
|
||||
socket: socket.clone(),
|
||||
buffered,
|
||||
};
|
||||
control_scope(move |cmd| match cmd {
|
||||
ScopeCmd::Write(metric, value) => buf.write(metric, value),
|
||||
ScopeCmd::Flush => buf.flush(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Its hard to see how a single scope could get more metrics than this.
|
||||
// TODO make configurable?
|
||||
const BUFFER_FLUSH_THRESHOLD: usize = 65_536;
|
||||
|
@ -161,11 +195,10 @@ mod bench {
|
|||
|
||||
#[bench]
|
||||
pub fn timer_graphite(b: &mut test::Bencher) {
|
||||
let sd = to_graphite("localhost:8125").unwrap();
|
||||
let sd = to_graphite("localhost:8125").unwrap().open_scope();
|
||||
let timer = sd.define_metric(Kind::Timer, "timer", 1000000.0);
|
||||
let scope = sd.open_scope(false);
|
||||
|
||||
b.iter(|| test::black_box(scope.write(&timer, 2000)));
|
||||
b.iter(|| test::black_box(sd.write(&timer, 2000)));
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -1,205 +0,0 @@
|
|||
//! Decouple metric definition from configuration with trait objects.
|
||||
|
||||
use core::*;
|
||||
use local_metrics::*;
|
||||
use output::*;
|
||||
use namespace::*;
|
||||
use registry::*;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, RwLock, Weak};
|
||||
|
||||
use atomic_refcell::*;
|
||||
|
||||
|
||||
/// Create a new dispatch point for metrics.
|
||||
/// All dispatch points are automatically entered in the dispatch registry.
|
||||
pub fn local_delegate() -> LocalSend {
|
||||
let delegation_point = LocalSend {
|
||||
inner_send: Arc::new(RwLock::new(InnerLocalSend {
|
||||
active_metrics: HashMap::new(),
|
||||
recv: registry.get,
|
||||
last_metric_id: 0,
|
||||
})),
|
||||
};
|
||||
register_local_delegation(delegation_point.clone());
|
||||
delegation_point
|
||||
}
|
||||
|
||||
/// Dynamic counterpart of a `Dispatcher`.
|
||||
/// Adapter to LocalMetrics<_> of unknown type.
|
||||
pub trait LocalRecv {
|
||||
/// Register a new metric.
|
||||
/// Only one metric of a certain name will be defined.
|
||||
/// Observer must return a MetricHandle that uniquely identifies the metric.
|
||||
fn define_metric(&self, kind: Kind, name: &str, rate: Rate) -> LocalRecvMetric;
|
||||
|
||||
/// Flush the receiver's scope.
|
||||
fn open_scope(&self, buffered: bool) -> Arc<LocalRecvScope + Send + Sync>;
|
||||
}
|
||||
|
||||
/// A dynamically dispatched metric.
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub trait LocalRecvScope {
|
||||
fn write(&self, metric: &LocalRecvMetric, value: Value);
|
||||
fn flush(&self);
|
||||
}
|
||||
|
||||
pub struct LocalRecvMetric (u64);
|
||||
|
||||
/// Shortcut name because `AppMetrics<Dispatch>`
|
||||
/// looks better than `AppMetrics<Arc<DispatcherMetric>>`.
|
||||
pub type LocalDelegate = Arc<LocalSendMetric>;
|
||||
|
||||
/// A dynamically dispatched metric.
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub struct LocalSendMetric {
|
||||
kind: Kind,
|
||||
name: String,
|
||||
rate: Rate,
|
||||
metric_id: usize,
|
||||
#[derivative(Debug = "ignore")]
|
||||
send: LocalSend,
|
||||
}
|
||||
|
||||
/// Dispatcher weak ref does not prevent dropping but still needs to be cleaned out.
|
||||
impl Drop for LocalSendMetric {
|
||||
fn drop(&mut self) {
|
||||
self.dispatcher.drop_metric(self)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/// Dispatcher weak ref does not prevent dropping but still needs to be cleaned out.
|
||||
impl Drop for LocalSendMetric {
|
||||
fn drop(&mut self) {
|
||||
self.send.drop_metric(self)
|
||||
}
|
||||
}
|
||||
|
||||
/// A dynamic dispatch point for app and lib metrics.
|
||||
/// Decouples metrics definition from backend configuration.
|
||||
/// Allows defining metrics before a concrete type has been selected.
|
||||
/// Allows replacing metrics backend on the fly at runtime.
|
||||
#[derive(Clone)]
|
||||
pub struct LocalSend {
|
||||
inner_send: Arc<RwLock<InnerLocalSend>>,
|
||||
}
|
||||
|
||||
struct InnerLocalSend {
|
||||
recv: Box<LocalRecv + Send + Sync>,
|
||||
active_metrics: HashMap<String, Weak<LocalSendMetric>>,
|
||||
last_metric_id: usize,
|
||||
}
|
||||
|
||||
impl From<&'static str> for LocalMetrics<LocalDelegate> {
|
||||
fn from(prefix: &'static str) -> LocalMetrics<LocalDelegate> {
|
||||
let app_metrics: LocalMetrics<LocalDelegate> = local_delegate().into();
|
||||
app_metrics.with_prefix(prefix)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<LocalSend> for LocalMetrics<LocalDelegate> {
|
||||
fn from(send: LocalSend) -> LocalMetrics<LocalDelegate> {
|
||||
let send_1 = send.clone();
|
||||
LocalMetrics::new(
|
||||
// define metric
|
||||
Arc::new(move |kind, name, rate| send.define_metric(kind, name, rate)),
|
||||
// write / flush metric
|
||||
Arc::new(move |buffered| send.open_scope(buffered))
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl LocalSend {
|
||||
/// Install a new metric receiver, replacing the previous one.
|
||||
pub fn set_receiver<IS: Into<LocalMetrics<T>>, T: Send + Sync + Clone + 'static>(
|
||||
&self,
|
||||
receiver: IS,
|
||||
) {
|
||||
let receiver: Box<LocalRecv + Send + Sync> = Box::new(receiver.into());
|
||||
let inner: &mut InnerLocalSend =
|
||||
&mut *self.inner_send.write().expect("Lock Metrics Send");
|
||||
|
||||
for mut metric in inner.active_metrics.values() {
|
||||
if let Some(metric) = metric.upgrade() {
|
||||
let receiver_metric =
|
||||
receiver.box_metric(metric.kind, metric.name.as_ref(), metric.rate);
|
||||
*metric.receiver.borrow_mut() = receiver_metric;
|
||||
}
|
||||
}
|
||||
// TODO return old receiver (swap, how?)
|
||||
inner.recv = receiver;
|
||||
}
|
||||
|
||||
fn define_metric(&self, kind: Kind, name: &str, rate: Rate) -> LocalDelegate {
|
||||
let mut inner = self.inner_send.write().expect("Lock Metrics Send");
|
||||
inner.metrics.get(name)
|
||||
.and_then(|metric_ref| Weak::upgrade(metric_ref))
|
||||
.unwrap_or_else(|| {
|
||||
let recv_metric = inner.recv.define_metric(kind, name, rate);
|
||||
let new_metric = Arc::new(LocalSendMetric {
|
||||
kind,
|
||||
name: name.to_string(),
|
||||
rate,
|
||||
metric_id: inner.last_metric_id += 1,
|
||||
send: send.clone(),
|
||||
});
|
||||
inner.metrics.insert(
|
||||
new_metric.name.clone(),
|
||||
Arc::downgrade(&new_metric),
|
||||
);
|
||||
new_metric
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
pub fn open_scope(&self, buffered: bool) -> Arc<ControlScopeFn<LocalDelegate>> {
|
||||
let mut inner = self.inner_send.write().expect("Lock Metrics Send");
|
||||
let write_scope = inner.recv.open_scope(buffered);
|
||||
let flush_scope = write_scope.clone();
|
||||
Arc::new(move |cmd| {
|
||||
match cmd {
|
||||
ScopeCmd::Write(metric, value) => write_scope.write(metric, value),
|
||||
ScopeCmd::Flush => flush_scope.flush(),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn drop_metric(&self, metric: &LocalSendMetric) {
|
||||
let mut inner = self.inner_send.write().expect("Lock Metrics Send");
|
||||
if inner.metrics.remove(&metric.name).is_none() {
|
||||
panic!("Could not remove DelegatingMetric weak ref from delegation point")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "bench")]
|
||||
mod bench {
|
||||
|
||||
use super::*;
|
||||
use test;
|
||||
use core::Kind::*;
|
||||
use aggregate::*;
|
||||
use publish::*;
|
||||
|
||||
#[bench]
|
||||
fn dispatch_marker_to_aggregate(b: &mut test::Bencher) {
|
||||
let dispatch = local_delegate();
|
||||
let sink: LocalMetrics<LocalDelegate> = dispatch.clone().into();
|
||||
dispatch.set_receiver(aggregate(summary, to_void()));
|
||||
let metric = sink.marker("event_a");
|
||||
b.iter(|| test::black_box(metric.mark()));
|
||||
}
|
||||
|
||||
#[bench]
|
||||
fn dispatch_marker_to_void(b: &mut test::Bencher) {
|
||||
let dispatch = local_delegate();
|
||||
let sink: LocalMetrics<LocalDelegate> = dispatch.into();
|
||||
let metric = sink.marker("event_a");
|
||||
b.iter(|| test::black_box(metric.mark()));
|
||||
}
|
||||
|
||||
}
|
|
@ -1,12 +1,10 @@
|
|||
//! Chain of command for unscoped metrics.
|
||||
|
||||
use core::*;
|
||||
use core::Kind::*;
|
||||
use app_metrics::AppMetrics;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use cache::*;
|
||||
use namespace::*;
|
||||
|
||||
/// A pair of functions composing a twin "chain of command".
|
||||
|
@ -22,12 +20,6 @@ pub struct LocalMetrics<M> {
|
|||
}
|
||||
|
||||
impl<M> LocalMetrics<M> {
|
||||
/// Define a new metric.
|
||||
#[allow(unused_variables)]
|
||||
pub fn define_metric(&self, kind: Kind, name: &str, sampling: Rate) -> M {
|
||||
(self.define_metric_fn)(kind, name, sampling)
|
||||
}
|
||||
|
||||
/// Open a new metric scope.
|
||||
/// Scope metrics allow an application to emit per-operation statistics,
|
||||
/// For example, producing a per-request performance log.
|
||||
|
@ -37,68 +29,29 @@ impl<M> LocalMetrics<M> {
|
|||
///
|
||||
/// ```rust
|
||||
/// use dipstick::*;
|
||||
/// let scope_metrics = to_log();
|
||||
/// let scope_metrics = to_log().open_scope();
|
||||
/// let request_counter = scope_metrics.counter("scope_counter");
|
||||
/// {
|
||||
/// let ref mut request_scope = scope_metrics.open_scope(true);
|
||||
/// request_counter.count(request_scope, 42);
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
pub fn open_scope(&self, buffered: bool) -> ControlScopeFn<M> {
|
||||
(self.scope_metric_fn)(buffered)
|
||||
pub fn open_scope(&self) -> AppMetrics<M> {
|
||||
AppMetrics::new(self.define_metric_fn.clone(), (self.scope_metric_fn)())
|
||||
}
|
||||
|
||||
/// Open a buffered scope.
|
||||
#[inline]
|
||||
pub fn buffered_scope(&self) -> ControlScopeFn<M> {
|
||||
self.open_scope(true)
|
||||
}
|
||||
}
|
||||
|
||||
/// Open an unbuffered scope.
|
||||
#[inline]
|
||||
pub fn unbuffered_scope(&self) -> ControlScopeFn<M> {
|
||||
self.open_scope(false)
|
||||
/// Create a new metric chain with the provided metric definition and scope creation functions.
|
||||
pub fn metrics_context<MF, WF, M>(make_metric: MF, make_scope: WF) -> LocalMetrics<M>
|
||||
where
|
||||
MF: Fn(Kind, &str, Rate) -> M + Send + Sync + 'static,
|
||||
WF: Fn() -> ControlScopeFn<M> + Send + Sync + 'static,
|
||||
{
|
||||
LocalMetrics {
|
||||
define_metric_fn: Arc::new(make_metric),
|
||||
scope_metric_fn: Arc::new(make_scope),
|
||||
}
|
||||
}
|
||||
|
||||
impl<M: Send + Sync + Clone + 'static> LocalMetrics<M> {
|
||||
/// Create a new metric chain with the provided metric definition and scope creation functions.
|
||||
pub fn new<MF, WF>(make_metric: MF, make_scope: WF) -> Self
|
||||
where
|
||||
MF: Fn(Kind, &str, Rate) -> M + Send + Sync + 'static,
|
||||
WF: Fn(bool) -> ControlScopeFn<M> + Send + Sync + 'static,
|
||||
{
|
||||
LocalMetrics {
|
||||
// capture the provided closures in Arc to provide cheap clones
|
||||
define_metric_fn: Arc::new(make_metric),
|
||||
scope_metric_fn: Arc::new(make_scope),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get an event counter of the provided name.
|
||||
pub fn marker<AS: AsRef<str>>(&self, name: AS) -> LocalMarker<M> {
|
||||
let metric = self.define_metric(Marker, name.as_ref(), 1.0);
|
||||
LocalMarker { metric }
|
||||
}
|
||||
|
||||
/// Get a counter of the provided name.
|
||||
pub fn counter<AS: AsRef<str>>(&self, name: AS) -> LocalCounter<M> {
|
||||
let metric = self.define_metric(Counter, name.as_ref(), 1.0);
|
||||
LocalCounter { metric }
|
||||
}
|
||||
|
||||
/// Get a timer of the provided name.
|
||||
pub fn timer<AS: AsRef<str>>(&self, name: AS) -> LocalTimer<M> {
|
||||
let metric = self.define_metric(Timer, name.as_ref(), 1.0);
|
||||
LocalTimer { metric }
|
||||
}
|
||||
|
||||
/// Get a gauge of the provided name.
|
||||
pub fn gauge<AS: AsRef<str>>(&self, name: AS) -> LocalGauge<M> {
|
||||
let metric = self.define_metric(Gauge, name.as_ref(), 1.0);
|
||||
LocalGauge { metric }
|
||||
}
|
||||
|
||||
/// Intercept both metric definition and scope creation, possibly changing the metric type.
|
||||
pub fn mod_both<MF, N>(&self, mod_fn: MF) -> LocalMetrics<N>
|
||||
|
@ -124,20 +77,12 @@ impl<M: Send + Sync + Clone + 'static> LocalMetrics<M> {
|
|||
scope_metric_fn: mod_fn(self.scope_metric_fn.clone()),
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
impl<M> From<LocalMetrics<M>> for AppMetrics<M> {
|
||||
fn from(metrics: LocalMetrics<M>) -> AppMetrics<M> {
|
||||
AppMetrics::new(metrics.define_metric_fn.clone(), metrics.open_scope(false))
|
||||
}
|
||||
}
|
||||
|
||||
impl<M: Send + Sync + Clone + 'static> WithCache for LocalMetrics<M> {
|
||||
fn with_cache(&self, cache_size: usize) -> Self {
|
||||
LocalMetrics {
|
||||
define_metric_fn: add_cache(cache_size, self.define_metric_fn.clone()),
|
||||
scope_metric_fn: self.scope_metric_fn.clone(),
|
||||
}
|
||||
metrics.open_scope()
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -151,113 +96,3 @@ impl<M: Send + Sync + Clone + 'static> WithNamespace for LocalMetrics<M> {
|
|||
}
|
||||
}
|
||||
|
||||
/// A monotonic counter metric.
|
||||
/// Since value is only ever increased by one, no value parameter is provided,
|
||||
/// preventing programming errors.
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub struct LocalMarker<M> {
|
||||
metric: M,
|
||||
}
|
||||
|
||||
impl<M> LocalMarker<M> {
|
||||
/// Record a single event occurence.
|
||||
#[inline]
|
||||
pub fn mark(&self, scope: &mut ControlScopeFn<M>) {
|
||||
scope.write(&self.metric, 1);
|
||||
}
|
||||
}
|
||||
|
||||
/// A counter that sends values to the metrics backend
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub struct LocalCounter<M> {
|
||||
metric: M,
|
||||
}
|
||||
|
||||
impl<M> LocalCounter<M> {
|
||||
/// Record a value count.
|
||||
#[inline]
|
||||
pub fn count<V>(&self, scope: &mut ControlScopeFn<M>, count: V)
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
scope.write(&self.metric, count.to_u64().unwrap());
|
||||
}
|
||||
}
|
||||
|
||||
/// A gauge that sends values to the metrics backend
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub struct LocalGauge<M> {
|
||||
metric: M,
|
||||
}
|
||||
|
||||
impl<M: Clone> LocalGauge<M> {
|
||||
/// Record a value point for this gauge.
|
||||
#[inline]
|
||||
pub fn value<V>(&self, scope: &mut ControlScopeFn<M>, value: V)
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
scope.write(&self.metric, value.to_u64().unwrap());
|
||||
}
|
||||
}
|
||||
|
||||
/// A timer that sends values to the metrics backend
|
||||
/// Timers can record time intervals in multiple ways :
|
||||
/// - with the time! macro which wraps an expression or block with start() and stop() calls.
|
||||
/// - with the time(Fn) method which wraps a closure with start() and stop() calls.
|
||||
/// - with start() and stop() methods wrapping around the operation to time
|
||||
/// - with the interval_us() method, providing an externally determined microsecond interval
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Debug)]
|
||||
pub struct LocalTimer<M> {
|
||||
metric: M,
|
||||
}
|
||||
|
||||
impl<M: Clone> LocalTimer<M> {
|
||||
/// Record a microsecond interval for this timer
|
||||
/// Can be used in place of start()/stop() if an external time interval source is used
|
||||
#[inline]
|
||||
pub fn interval_us<V>(&self, scope: &mut ControlScopeFn<M>, interval_us: V) -> V
|
||||
where
|
||||
V: ToPrimitive,
|
||||
{
|
||||
scope.write(&self.metric, interval_us.to_u64().unwrap());
|
||||
interval_us
|
||||
}
|
||||
|
||||
/// Obtain a opaque handle to the current time.
|
||||
/// The handle is passed back to the stop() method to record a time interval.
|
||||
/// This is actually a convenience method to the TimeHandle::now()
|
||||
/// Beware, handles obtained here are not bound to this specific timer instance
|
||||
/// _for now_ but might be in the future for safety.
|
||||
/// If you require safe multi-timer handles, get them through TimeType::now()
|
||||
#[inline]
|
||||
pub fn start(&self) -> TimeHandle {
|
||||
TimeHandle::now()
|
||||
}
|
||||
|
||||
/// Record the time elapsed since the start_time handle was obtained.
|
||||
/// This call can be performed multiple times using the same handle,
|
||||
/// reporting distinct time intervals each time.
|
||||
/// Returns the microsecond interval value that was recorded.
|
||||
#[inline]
|
||||
pub fn stop(&self, scope: &mut ControlScopeFn<M>, start_time: TimeHandle) -> u64 {
|
||||
let elapsed_us = start_time.elapsed_us();
|
||||
self.interval_us(scope, elapsed_us)
|
||||
}
|
||||
|
||||
/// Record the time taken to execute the provided closure
|
||||
#[inline]
|
||||
pub fn time<F, R>(&self, scope: &mut ControlScopeFn<M>, operations: F) -> R
|
||||
where
|
||||
F: FnOnce() -> R,
|
||||
{
|
||||
let start_time = self.start();
|
||||
let value: R = operations();
|
||||
self.stop(scope, start_time);
|
||||
value
|
||||
}
|
||||
}
|
||||
|
|
114
src/multi.rs
114
src/multi.rs
|
@ -4,92 +4,39 @@ use core::*;
|
|||
use local_metrics::*;
|
||||
use app_metrics::*;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Two chains of different types can be combined in a tuple.
|
||||
/// The chains will act as one, each receiving calls in the order the appear in the tuple.
|
||||
/// For more than two types, make tuples of tuples, "Yo Dawg" style.
|
||||
impl<M1, M2> From<(LocalMetrics<M1>, LocalMetrics<M2>)> for LocalMetrics<(M1, M2)>
|
||||
where
|
||||
M1: 'static + Clone + Send + Sync,
|
||||
M2: 'static + Clone + Send + Sync,
|
||||
{
|
||||
fn from(combo: (LocalMetrics<M1>, LocalMetrics<M2>)) -> LocalMetrics<(M1, M2)> {
|
||||
let combo0 = combo.0.clone();
|
||||
let combo1 = combo.1.clone();
|
||||
|
||||
LocalMetrics::new(
|
||||
move |kind, name, rate| {
|
||||
(
|
||||
combo.0.define_metric(kind, name, rate),
|
||||
combo.1.define_metric(kind, name, rate),
|
||||
)
|
||||
},
|
||||
move |buffered| {
|
||||
let scope0 = combo0.open_scope(buffered);
|
||||
let scope1 = combo1.open_scope(buffered);
|
||||
|
||||
control_scope(move |cmd| match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
let metric: &(M1, M2) = metric;
|
||||
scope0.write(&metric.0, value);
|
||||
scope1.write(&metric.1, value);
|
||||
}
|
||||
ScopeCmd::Flush => {
|
||||
scope0.flush();
|
||||
scope1.flush();
|
||||
}
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl<M1, M2> From<(LocalMetrics<M1>, LocalMetrics<M2>)> for AppMetrics<(M1, M2)>
|
||||
where
|
||||
M1: 'static + Clone + Send + Sync,
|
||||
M2: 'static + Clone + Send + Sync,
|
||||
{
|
||||
fn from(combo: (LocalMetrics<M1>, LocalMetrics<M2>)) -> AppMetrics<(M1, M2)> {
|
||||
let chain: LocalMetrics<(M1, M2)> = combo.into();
|
||||
app_metrics(chain)
|
||||
}
|
||||
}
|
||||
let scope0 = combo.0.open_scope();
|
||||
let scope1 = combo.1.open_scope();
|
||||
|
||||
/// Multiple chains of the same type can be combined in a slice.
|
||||
/// The chains will act as one, each receiving calls in the order the appear in the slice.
|
||||
impl<'a, M> From<&'a [LocalMetrics<M>]> for LocalMetrics<Vec<M>>
|
||||
where
|
||||
M: 'static + Clone + Send + Sync,
|
||||
{
|
||||
fn from(chains: &'a [LocalMetrics<M>]) -> LocalMetrics<Vec<M>> {
|
||||
let chains = chains.to_vec();
|
||||
let chains2 = chains.clone();
|
||||
let scope0a = scope0.clone();
|
||||
let scope1a = scope1.clone();
|
||||
|
||||
LocalMetrics::new(
|
||||
move |kind, name, rate| {
|
||||
let mut metric = Vec::with_capacity(chains.len());
|
||||
for chain in &chains {
|
||||
metric.push(chain.define_metric(kind, name, rate));
|
||||
AppMetrics::new(
|
||||
Arc::new(move |kind, name, rate| (
|
||||
scope0.define_metric(kind, name, rate),
|
||||
scope1.define_metric(kind, name, rate),
|
||||
)),
|
||||
control_scope(move |cmd| { match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
let metric: &(M1, M2) = metric;
|
||||
scope0a.write(&metric.0, value);
|
||||
scope1a.write(&metric.1, value);
|
||||
}
|
||||
metric
|
||||
},
|
||||
move |buffered| {
|
||||
let mut scopes = Vec::with_capacity(chains2.len());
|
||||
for chain in &chains2 {
|
||||
scopes.push(chain.open_scope(buffered));
|
||||
ScopeCmd::Flush => {
|
||||
scope0a.flush();
|
||||
scope1a.flush();
|
||||
}
|
||||
|
||||
control_scope(move |cmd| match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
let metric: &Vec<M> = metric;
|
||||
for (i, scope) in scopes.iter().enumerate() {
|
||||
scope.write(&metric[i], value)
|
||||
}
|
||||
}
|
||||
ScopeCmd::Flush => for scope in &scopes {
|
||||
scope.flush()
|
||||
},
|
||||
})
|
||||
},
|
||||
}})
|
||||
)
|
||||
}
|
||||
}
|
||||
|
@ -99,7 +46,24 @@ where
|
|||
M: 'static + Clone + Send + Sync,
|
||||
{
|
||||
fn from(chains: &'a [LocalMetrics<M>]) -> AppMetrics<Vec<M>> {
|
||||
let chain: LocalMetrics<Vec<M>> = chains.into();
|
||||
app_metrics(chain)
|
||||
let scopes: Vec<AppMetrics<M>> = chains.iter().map(|x| x.open_scope()).collect();
|
||||
let scopes2 = scopes.clone();
|
||||
|
||||
AppMetrics::new(
|
||||
Arc::new(move |kind, name, rate| {
|
||||
scopes.iter().map(|m| m.define_metric(kind, name, rate)).collect()
|
||||
}),
|
||||
control_scope(move |cmd| { match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
let metric: &Vec<M> = metric;
|
||||
for (i, scope) in scopes2.iter().enumerate() {
|
||||
scope.write(&metric[i], value)
|
||||
}
|
||||
}
|
||||
ScopeCmd::Flush => for scope in &scopes2 {
|
||||
scope.flush()
|
||||
},
|
||||
}})
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
114
src/output.rs
114
src/output.rs
|
@ -6,30 +6,36 @@ use std::sync::RwLock;
|
|||
|
||||
/// Write metric values to stdout using `println!`.
|
||||
pub fn to_stdout() -> LocalMetrics<String> {
|
||||
LocalMetrics::new(
|
||||
metrics_context(
|
||||
|_kind, name, _rate| String::from(name),
|
||||
|buffered| {
|
||||
if !buffered {
|
||||
control_scope(|cmd| {
|
||||
if let ScopeCmd::Write(m, v) = cmd {
|
||||
println!("{}: {}", m, v)
|
||||
|| control_scope(|cmd|
|
||||
if let ScopeCmd::Write(m, v) = cmd {
|
||||
println!("{}: {}", m, v)
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/// Record metric values to stdout using `println!`.
|
||||
/// Values are buffered until #flush is called
|
||||
/// Buffered operation requires locking.
|
||||
/// If thread latency is a concern you may wish to also use #with_async_queue.
|
||||
pub fn to_buffered_stdout() -> LocalMetrics<String> {
|
||||
metrics_context(
|
||||
|_kind, name, _rate| String::from(name),
|
||||
|| {
|
||||
let buf = RwLock::new(String::new());
|
||||
control_scope(move |cmd| {
|
||||
let mut buf = buf.write().expect("Locking stdout buffer");
|
||||
match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
buf.push_str(format!("{}: {}\n", metric, value).as_ref())
|
||||
}
|
||||
})
|
||||
} else {
|
||||
let buf = RwLock::new(String::new());
|
||||
control_scope(move |cmd| {
|
||||
let mut buf = buf.write().expect("Locking stdout buffer");
|
||||
match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
buf.push_str(format!("{}: {}\n", metric, value).as_ref())
|
||||
}
|
||||
ScopeCmd::Flush => {
|
||||
println!("{}", buf);
|
||||
buf.clear();
|
||||
}
|
||||
ScopeCmd::Flush => {
|
||||
println!("{}", buf);
|
||||
buf.clear();
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
|
@ -37,39 +43,47 @@ pub fn to_stdout() -> LocalMetrics<String> {
|
|||
/// Write metric values to the standard log using `info!`.
|
||||
// TODO parameterize log level
|
||||
pub fn to_log() -> LocalMetrics<String> {
|
||||
LocalMetrics::new(
|
||||
metrics_context(
|
||||
|_kind, name, _rate| String::from(name),
|
||||
|buffered| {
|
||||
if !buffered {
|
||||
control_scope(|cmd| {
|
||||
if let ScopeCmd::Write(m, v) = cmd {
|
||||
info!("{}: {}", m, v)
|
||||
|| control_scope(|cmd|
|
||||
if let ScopeCmd::Write(m, v) = cmd {
|
||||
info!("{}: {}", m, v)
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/// Record metric values to the standard log using `info!`.
|
||||
/// Values are buffered until #flush is called
|
||||
/// Buffered operation requires locking.
|
||||
/// If thread latency is a concern you may wish to also use #with_async_queue.
|
||||
// TODO parameterize log level
|
||||
pub fn to_buffered_log() -> LocalMetrics<String> {
|
||||
metrics_context(
|
||||
|_kind, name, _rate| String::from(name),
|
||||
|| {
|
||||
let buf = RwLock::new(String::new());
|
||||
control_scope(move |cmd| {
|
||||
let mut buf = buf.write().expect("Locking string buffer");
|
||||
match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
buf.push_str(format!("{}: {}\n", metric, value).as_ref())
|
||||
}
|
||||
})
|
||||
} else {
|
||||
let buf = RwLock::new(String::new());
|
||||
control_scope(move |cmd| {
|
||||
let mut buf = buf.write().expect("Locking string buffer");
|
||||
match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
buf.push_str(format!("{}: {}\n", metric, value).as_ref())
|
||||
}
|
||||
ScopeCmd::Flush => {
|
||||
info!("{}", buf);
|
||||
buf.clear();
|
||||
}
|
||||
ScopeCmd::Flush => {
|
||||
info!("{}", buf);
|
||||
buf.clear();
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
/// Discard all metric values sent to it.
|
||||
pub fn to_void() -> LocalMetrics<()> {
|
||||
LocalMetrics::new(
|
||||
metrics_context(
|
||||
move |_kind, _name, _rate| (),
|
||||
|_buffered| control_scope(|_cmd| {}),
|
||||
|| control_scope(|_cmd| {}),
|
||||
)
|
||||
}
|
||||
|
||||
|
@ -79,23 +93,23 @@ mod test {
|
|||
|
||||
#[test]
|
||||
fn sink_print() {
|
||||
let c = super::to_stdout();
|
||||
let c = super::to_stdout().open_scope();
|
||||
let m = c.define_metric(Kind::Marker, "test", 1.0);
|
||||
c.open_scope(true).write(&m, 33);
|
||||
c.write(&m, 33);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_to_log() {
|
||||
let c = super::to_log();
|
||||
let c = super::to_log().open_scope();
|
||||
let m = c.define_metric(Kind::Marker, "test", 1.0);
|
||||
c.open_scope(true).write(&m, 33);
|
||||
c.write(&m, 33);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_to_void() {
|
||||
let c = super::to_void();
|
||||
let c = super::to_void().open_scope();
|
||||
let m = c.define_metric(Kind::Marker, "test", 1.0);
|
||||
c.open_scope(true).write(&m, 33);
|
||||
c.write(&m, 33);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -64,7 +64,7 @@ where
|
|||
E: Fn(Kind, &str, ScoreType) -> Option<(Kind, Vec<&str>, Value)> + Send + Sync + 'static,
|
||||
{
|
||||
fn publish(&self, snapshot: Vec<ScoreSnapshot>) {
|
||||
let publish_scope_fn = self.output.open_scope(false);
|
||||
let publish_scope = self.output.open_scope();
|
||||
if snapshot.is_empty() {
|
||||
// no data was collected for this period
|
||||
// TODO repeat previous frame min/max ?
|
||||
|
@ -73,8 +73,8 @@ where
|
|||
for metric in snapshot {
|
||||
for score in metric.2 {
|
||||
if let Some(ex) = (self.statistics)(metric.0, metric.1.as_ref(), score) {
|
||||
let pub_metric = self.output.define_metric(ex.0, &ex.1.concat(), 1.0);
|
||||
publish_scope_fn.write(&pub_metric, ex.2);
|
||||
let pub_metric = publish_scope.define_metric(ex.0, &ex.1.concat(), 1.0);
|
||||
publish_scope.write(&pub_metric, ex.2);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -82,7 +82,7 @@ where
|
|||
|
||||
// TODO parameterize whether to keep ad-hoc metrics after publish
|
||||
// source.cleanup();
|
||||
publish_scope_fn.flush()
|
||||
publish_scope.flush()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -34,8 +34,8 @@ impl<M: Send + Sync + 'static + Clone> WithSamplingRate for LocalMetrics<M> {
|
|||
let new_rate = sampling_rate * rate;
|
||||
metric_fn(kind, name, new_rate)
|
||||
}),
|
||||
Arc::new(move |buffered| {
|
||||
let next_scope = scope_fn(buffered);
|
||||
Arc::new(move || {
|
||||
let next_scope = scope_fn();
|
||||
control_scope(move |cmd| match cmd {
|
||||
ScopeCmd::Write(metric, value) => {
|
||||
if pcg32::accept_sample(int_sampling_rate) {
|
||||
|
|
|
@ -26,7 +26,10 @@ where
|
|||
socket.set_nonblocking(true)?;
|
||||
socket.connect(address)?;
|
||||
|
||||
Ok(LocalMetrics::new(
|
||||
// TODO buffering toggle
|
||||
let buffered = false;
|
||||
|
||||
Ok(metrics_context(
|
||||
move |kind, name, rate| {
|
||||
let mut prefix = String::with_capacity(32);
|
||||
prefix.push_str(name);
|
||||
|
@ -57,7 +60,7 @@ where
|
|||
scale,
|
||||
}
|
||||
},
|
||||
move |buffered| {
|
||||
move || {
|
||||
let buf = RwLock::new(ScopeBuffer {
|
||||
buffer: String::with_capacity(MAX_UDP_PAYLOAD),
|
||||
socket: socket.clone(),
|
||||
|
@ -156,11 +159,10 @@ mod bench {
|
|||
|
||||
#[bench]
|
||||
pub fn timer_statsd(b: &mut test::Bencher) {
|
||||
let sd = to_statsd("localhost:8125").unwrap();
|
||||
let sd = to_statsd("localhost:8125").unwrap().open_scope();
|
||||
let timer = sd.define_metric(Kind::Timer, "timer", 1000000.0);
|
||||
let scope = sd.open_scope(false);
|
||||
|
||||
b.iter(|| test::black_box(scope.write(&timer, 2000)));
|
||||
b.iter(|| test::black_box(sd.write(&timer, 2000)));
|
||||
}
|
||||
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue