agentguard_telemetry/sinks/
jsonl.rs

1//! JSONL sink — appends each event as one JSON line.
2
3use crate::sink::{Sink, SinkError, SinkEvent};
4use async_trait::async_trait;
5use std::fs::{File, OpenOptions};
6use std::io::Write;
7use std::path::{Path, PathBuf};
8use std::sync::Mutex;
9
10/// Append-only JSONL sink.
11///
12/// Thread-safe: a single `Arc<JsonlSink>` can be shared across threads.
13pub struct JsonlSink {
14    path: PathBuf,
15    inner: Mutex<Option<File>>,
16}
17
18impl JsonlSink {
19    pub fn open(path: impl Into<PathBuf>) -> std::io::Result<Self> {
20        let path = path.into();
21        if let Some(parent) = path.parent() {
22            std::fs::create_dir_all(parent)?;
23        }
24        let f = OpenOptions::new().create(true).append(true).open(&path)?;
25        Ok(Self {
26            path,
27            inner: Mutex::new(Some(f)),
28        })
29    }
30
31    pub fn path(&self) -> &Path {
32        &self.path
33    }
34}
35
36#[async_trait]
37impl Sink for JsonlSink {
38    fn name(&self) -> &str {
39        "jsonl"
40    }
41
42    async fn emit(&self, event: &SinkEvent) -> Result<(), SinkError> {
43        let line = serde_json::to_string(event)?;
44        // ponytail: see StdoutSink — recover from a poisoned
45        // mutex instead of panicking the worker.
46        let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
47        if let Some(f) = guard.as_mut() {
48            writeln!(f, "{}", line)?;
49            f.flush()?;
50        }
51        Ok(())
52    }
53
54    async fn shutdown(&self) -> Result<(), SinkError> {
55        let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
56        *guard = None;
57        Ok(())
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64    use crate::sink::SinkEvent;
65    use tempfile::tempdir;
66
67    #[tokio::test]
68    async fn jsonl_sink_writes_one_line_per_event() {
69        let dir = tempdir().unwrap();
70        let path = dir.path().join("events.jsonl");
71        let sink = JsonlSink::open(&path).unwrap();
72
73        for i in 0..3 {
74            let event = SinkEvent::decision(
75                if i % 2 == 0 { "allow" } else { "deny" },
76                "alice",
77                "send_email",
78                "alice@acme",
79                vec![format!("policy{i}")],
80                vec![],
81                1000 + i as u64,
82            );
83            sink.emit(&event).await.unwrap();
84        }
85        sink.flush().await.unwrap();
86
87        let contents = std::fs::read_to_string(&path).unwrap();
88        let lines: Vec<&str> = contents.lines().collect();
89        assert_eq!(lines.len(), 3);
90        for (i, line) in lines.iter().enumerate() {
91            let parsed: serde_json::Value = serde_json::from_str(line).unwrap();
92            assert_eq!(parsed["kind"]["principal"], "alice");
93            assert_eq!(parsed["kind"]["action"], "send_email");
94            assert_eq!(
95                parsed["kind"]["effect"],
96                if i % 2 == 0 { "allow" } else { "deny" }
97            );
98        }
99    }
100}
agentguard_telemetry/sinks/jsonl.rs

agentguard_telemetry/sinks/
jsonl.rs
