Write a custom sink¶

The trait is two methods, one of which has a default. Implementing it is the supported way to reach a destination the built-ins do not cover, to buffer, or to send to more than one place.

Implement the trait¶

use async_trait::async_trait;
use rtb_telemetry::{Event, TelemetryError, TelemetrySink};

pub struct StdoutSink;

#[async_trait]
impl TelemetrySink for StdoutSink {
    async fn emit(&self, event: &Event) -> Result<(), TelemetryError> {
        let line = serde_json::to_string(&event.redacted())
            .map_err(|e| TelemetryError::Serde(e.to_string()))?;
        println!("{line}");
        Ok(())
    }
}

You need async-trait in your own Cargo.toml — the trait is declared with #[async_trait], so implementations must carry the attribute too.

emit takes &self, so any state your sink needs is behind a Mutex, a channel or an atomic. The trait requires Send + Sync + 'static, which your type will satisfy automatically if its fields do.

Call redacted() if the event leaves the process¶

Nothing calls it for you. event.redacted() returns a clone with args and err_msg scrubbed; write that, not the original, whenever the destination is a file, a socket or anything else outside the program. A sink that only keeps events in memory for tests does not need it — which is exactly why MemorySink does not do it.

Fan out to several destinations¶

A context holds one sink, so "file and HTTP" is a sink that owns both:

pub struct FanOutSink {
    sinks: Vec<Arc<dyn TelemetrySink>>,
}

#[async_trait]
impl TelemetrySink for FanOutSink {
    async fn emit(&self, event: &Event) -> Result<(), TelemetryError> {
        let mut first_err = None;
        for sink in &self.sinks {
            if let Err(err) = sink.emit(event).await {
                first_err.get_or_insert(err);
            }
        }
        first_err.map_or(Ok(()), Err)
    }

    async fn flush(&self) -> Result<(), TelemetryError> {
        for sink in &self.sinks {
            sink.flush().await?;
        }
        Ok(())
    }
}

The interesting decision is the failure policy, and it is yours to make. The version above tries every sink and reports the first failure, so a dead HTTP endpoint does not stop the local file being written. Returning on the first error instead would be a different, equally defensible choice — but decide it deliberately rather than inheriting it from a ?.

Buffer, if one-request-per-event is too expensive¶

Nothing in the crate batches for you. A buffering sink collects events in emit, flushes on a size or time threshold, and does the real work in flush — which is the method TelemetryContext::flush reaches. Two things to get right: the buffer needs to be Send + Sync, and something has to call flush before the process exits, or the buffer dies with it. There is no Drop hook that flushes for you.

Override flush only if you buffer¶

The default flush returns Ok(()). Leave it alone for a sink that writes synchronously in emit — an override that does nothing is one more thing to read.