rtb_telemetry/
sink.rs

1//! [`TelemetrySink`] trait and the three built-in implementations.
2
3use std::path::PathBuf;
4use std::sync::{Arc, Mutex};
5
6use async_trait::async_trait;
7use tokio::io::AsyncWriteExt;
8
9use crate::error::TelemetryError;
10use crate::event::Event;
11
12/// Backend-agnostic sink for [`Event`] values.
13#[async_trait]
14pub trait TelemetrySink: Send + Sync + 'static {
15    /// Emit a single event. Sinks SHOULD be idempotent on the event —
16    /// retries or batching live in wrapper sinks (v0.2).
17    async fn emit(&self, event: &Event) -> Result<(), TelemetryError>;
18
19    /// Flush any buffered events. Default impl is a no-op.
20    async fn flush(&self) -> Result<(), TelemetryError> {
21        Ok(())
22    }
23}
24
25// =====================================================================
26// NoopSink — drops everything silently.
27// =====================================================================
28
29/// Drops every event silently. Used when collection is disabled and
30/// as the default when a tool ships telemetry support but hasn't
31/// configured a real sink.
32#[derive(Debug, Default, Clone, Copy)]
33pub struct NoopSink;
34
35#[async_trait]
36impl TelemetrySink for NoopSink {
37    async fn emit(&self, _event: &Event) -> Result<(), TelemetryError> {
38        Ok(())
39    }
40}
41
42// =====================================================================
43// MemorySink — in-memory Vec, useful for tests.
44// =====================================================================
45
46/// In-memory sink for tests. Events are appended in emit order and
47/// inspectable via [`MemorySink::snapshot`].
48#[derive(Debug, Default, Clone)]
49pub struct MemorySink {
50    inner: Arc<Mutex<Vec<Event>>>,
51}
52
53impl MemorySink {
54    /// Construct a fresh empty sink.
55    #[must_use]
56    pub fn new() -> Self {
57        Self::default()
58    }
59
60    /// Return a clone of every recorded event.
61    #[must_use]
62    pub fn snapshot(&self) -> Vec<Event> {
63        self.inner.lock().map(|v| v.clone()).unwrap_or_default()
64    }
65
66    /// How many events have been recorded so far.
67    #[must_use]
68    pub fn len(&self) -> usize {
69        self.inner.lock().map_or(0, |v| v.len())
70    }
71
72    /// `true` when no events have been recorded.
73    #[must_use]
74    pub fn is_empty(&self) -> bool {
75        self.len() == 0
76    }
77}
78
79#[async_trait]
80impl TelemetrySink for MemorySink {
81    async fn emit(&self, event: &Event) -> Result<(), TelemetryError> {
82        if let Ok(mut v) = self.inner.lock() {
83            v.push(event.clone());
84        }
85        Ok(())
86    }
87}
88
89// =====================================================================
90// FileSink — newline-delimited JSON to disk.
91// =====================================================================
92
93/// Appends events as JSON Lines to a file. Parent directories are
94/// created on demand.
95///
96/// # Concurrency and line integrity
97///
98/// Concurrent `emit` calls on the same `FileSink` are serialised via
99/// a shared `tokio::sync::Mutex`. This is required for JSONL
100/// correctness: on POSIX, `O_APPEND` only guarantees atomicity for
101/// individual `write()` calls up to `PIPE_BUF` (4 KiB on Linux). An
102/// event whose serialised form exceeds that bound — plausible for
103/// events with many attrs — would interleave at the byte level with
104/// concurrent writers and produce malformed JSONL.
105///
106/// Cross-**process** writers (two `FileSink`s in different processes
107/// targeting the same file) remain interleaving-safe only up to
108/// `PIPE_BUF`. Don't do that; use per-process files and aggregate
109/// elsewhere.
110///
111/// Batching lives in a wrapper sink when we need it (v0.2).
112#[derive(Debug, Clone)]
113pub struct FileSink {
114    path: PathBuf,
115    // Serialises concurrent `emit` calls. Shared across `Clone`s of
116    // the same `FileSink` so multiple handles to the same path also
117    // serialise correctly.
118    gate: Arc<tokio::sync::Mutex<()>>,
119}
120
121impl FileSink {
122    /// Construct a sink targeting `path`. The file is not touched
123    /// until the first `emit`; parent directories are created on
124    /// demand at that point.
125    #[must_use]
126    pub fn new(path: impl Into<PathBuf>) -> Self {
127        Self { path: path.into(), gate: Arc::new(tokio::sync::Mutex::new(())) }
128    }
129}
130
131#[async_trait]
132impl TelemetrySink for FileSink {
133    async fn emit(&self, event: &Event) -> Result<(), TelemetryError> {
134        // Redact `args` / `err_msg` before serialisation — the file
135        // is the first out-of-process surface.
136        let redacted = event.redacted();
137        // Serialise the line outside the critical section — parsing
138        // the event is the expensive part.
139        let mut line =
140            serde_json::to_string(&redacted).map_err(|e| TelemetryError::Serde(e.to_string()))?;
141        line.push('\n');
142
143        // Hold the write gate across parent-dir creation + open +
144        // write + flush so concurrent `emit` calls never interleave
145        // bytes.
146        let _guard = self.gate.lock().await;
147
148        if let Some(parent) = self.path.parent() {
149            if !parent.as_os_str().is_empty() {
150                tokio::fs::create_dir_all(parent).await?;
151            }
152        }
153
154        let mut f =
155            tokio::fs::OpenOptions::new().create(true).append(true).open(&self.path).await?;
156        f.write_all(line.as_bytes()).await?;
157        f.flush().await?;
158        Ok(())
159    }
160}
rtb_telemetry/sink.rs

rtb_telemetry/
sink.rs
