parasyte
commented
4 years ago While I would generally prefer a natively async crate for metrics, I don't necessarily think that it's a showstopper.
The synchronous client can be run in its own thread, consuming from an IPC channel synchronously. The async side would then send metrics over the IPC channel asynchronously. Picking a good IPC implementation is the only challenge; for apps built on Tokio 0.1, there is parity-tokio-ipc, and for Tokio 0.2 you can go straight to tokio-uds (no Windows support).
The better option is of course async all the way down. And best without tying the crate directly to any specific runtime (like Tokio). Going over the code, it doesn't look like this would be too much trouble.
First all of the client traits need async versions. E.g.:
pub trait AsyncGauged: Gauged {
/// Record a gauge value with the given key
async fn gauge(&self, key: &str, value: u64) -> MetricResult<Gauge> {
self.gauge_with_tags(key, value)
.try_send_async()
.await?
}
}
/// Async trait that encompasses all other async traits for sending metrics.
pub trait AsyncMetricClient: AsyncCounted + AsyncTimed + AsyncGauged + AsyncMetered + AsyncHistogrammed + AsyncSetted {}And now MetricBuilder needs a new async method:
impl<'m, 'c, T> MetricBuilder<'m, 'c, T>
where
T: Metric + From<String>,
{
// ... snip ...
/// Asynchronously send a metric using the client that created this builder.
pub async fn try_send_async() -> MetricResult<T> {
match self.repr {
BuilderRepr::Error(err, _) => Err(err),
BuilderRepr::Success(ref formatter, client) => {
let metric: T = formatter.build();
client.send_metric_async(&metric).await?;
Ok(metric)
}
}
}
// ... same for async version of the `send` method ...
}We extend MetricBackend with the new send_metric_async method, and implement it on a new async client:
pub trait MetricBackend {
// ... snip ...
/// Asynchronously send a full formed `Metric` implementation via the underlying `MetricSink`
async fn send_metric_async<M>(&self, metric: &M) -> MetricResult<()>
where
M: Metric;
}
impl MetricBackend for AsyncStatsdClient {
// ... snip ...
async fn send_metric_async<M>(&self, metric: &M) -> MetricResult<()>
where
M: Metric,
{
let metric_string = metric.as_metric_str();
self.sink.emit_async(metric_string).await?;
Ok(())
}
}Finally, we need an async version of the MetricsSink trait, and new async implementers like AsyncBufferedUdpMetricSink and AsyncQueueingMetricSink:
pub trait AsyncMetricSink {
// ... snip ...
/// Asynchronously send the Statsd metric using this sink and return the number of bytes
/// written or an I/O error.
async fn emit_async(&self, metric: &str) -> io::Result<usize>;
}And that should do it! Client usage would look like this:
use cadence::async_prelude::*;
use cadence::{AsyncBufferedUdpMetricSink, AsyncQueuingMetricSink, AsyncStatsdClient, DEFAULT_PORT};
use tokio::net::UdpSocket
#[tokio::main]
fn main() {
let sock = UdpSocket::bind("0.0.0.0:0").await.unwrap();
let buffered = AsyncBufferedUdpMetricSink::from(("localhost", DEFAULT_PORT), sock).unwrap();
let queued = AsyncQueuingMetricSink::from(buffered);
let metrics = AsyncStatsdClient::from_sink("example.prefix", queued);
metrics.count("example.counter", 1).await.unwrap();
metrics.gauge("example.gauge", 5).await.unwrap();
metrics.time("example.timer", 32).await.unwrap();
metrics.histogram("example.histogram", 22).await.unwrap();
metrics.meter("example.meter", 8).await.unwrap();
}I'm not in love with the naming. There is also a lot of bifurcation in this proposal. But that's kind of a known issue in general. It may be worth splitting the async-specific parts into their own crate, but that could be difficult with the existing method visibility. Another popular option in these cases is a feature flag to switch the API from sync to async (without renaming traits and methods).
If this sounds good to you, I can possibly start on this work next week or early January. I will probably need this sooner than later, and I wouldn't be comfortable putting the IPC workaround into production. :)
56quarters
tustvold
Lots of Rust servers / apps are being created based on Tokio which uses
Future(and the associated lib) as one its key abstractions. What would it take to expose something like theMetricClientAPI in a way that either uses Futures or the Tokio event loop?