agentguard_telemetry/
sink.rs

1//! Sink trait and event payload.
2
3use async_trait::async_trait;
4use chrono::{DateTime, Utc};
5use serde::{Deserialize, Serialize};
6use thiserror::Error;
7use uuid::Uuid;
8
9/// Errors raised by [`Sink::emit`].
10#[derive(Debug, Error)]
11pub enum SinkError {
12    #[error("io error: {0}")]
13    Io(#[from] std::io::Error),
14    #[error("json error: {0}")]
15    Json(#[from] serde_json::Error),
16    #[error("send error: {0}")]
17    Send(String),
18    #[error("other: {0}")]
19    Other(String),
20}
21
22/// Telemetry event emitted by agentguard.
23#[derive(Debug, Clone, Serialize, Deserialize)]
24pub struct SinkEvent {
25    /// Unique event id (UUID v4).
26    pub id: Uuid,
27    /// Event timestamp.
28    pub timestamp: DateTime<Utc>,
29    /// Discriminator.
30    pub kind: SinkEventKind,
31}
32
33/// What the event represents.
34#[derive(Debug, Clone, Serialize, Deserialize)]
35#[serde(tag = "type", rename_all = "snake_case")]
36pub enum SinkEventKind {
37    /// Authorization decision (allow or deny).
38    Decision {
39        effect: String,
40        principal: String,
41        action: String,
42        resource: String,
43        policies: Vec<String>,
44        reasons: Vec<String>,
45        duration_micros: u64,
46        #[serde(default, skip_serializing_if = "Option::is_none")]
47        trace_id: Option<String>,
48        #[serde(default, skip_serializing_if = "Option::is_none")]
49        span_id: Option<String>,
50        #[serde(default, skip_serializing_if = "Option::is_none")]
51        tenant_id: Option<String>,
52        #[serde(default, skip_serializing_if = "Option::is_none")]
53        subject_id: Option<String>,
54        #[serde(default)]
55        cached: bool,
56    },
57    /// Delegation token minted.
58    DelegationMint {
59        issuer: String,
60        subject: String,
61        actions: Vec<String>,
62        ttl_seconds: i64,
63    },
64    /// Delegation token verified.
65    DelegationVerify {
66        success: bool,
67        issuer: Option<String>,
68        subject: Option<String>,
69    },
70    /// Policy bundle loaded or reloaded.
71    PolicyReload {
72        version: String,
73        policy_count: usize,
74        source: String,
75    },
76    /// Decision cache hit/miss.
77    CacheLookup {
78        hit: bool,
79        principal: String,
80        action: String,
81    },
82    /// Authorization failed because the PDP could not be reached.
83    PdpError { error: String, fallback: String },
84}
85
86impl SinkEvent {
87    pub fn decision(
88        effect: impl Into<String>,
89        principal: impl Into<String>,
90        action: impl Into<String>,
91        resource: impl Into<String>,
92        policies: Vec<String>,
93        reasons: Vec<String>,
94        duration_micros: u64,
95    ) -> Self {
96        Self {
97            id: Uuid::new_v4(),
98            timestamp: Utc::now(),
99            kind: SinkEventKind::Decision {
100                effect: effect.into(),
101                principal: principal.into(),
102                action: action.into(),
103                resource: resource.into(),
104                policies,
105                reasons,
106                duration_micros,
107                trace_id: None,
108                span_id: None,
109                tenant_id: None,
110                subject_id: None,
111                cached: false,
112            },
113        }
114    }
115
116    pub fn with_trace(mut self, trace_id: String, span_id: String) -> Self {
117        if let SinkEventKind::Decision {
118            trace_id: ref mut t,
119            span_id: ref mut s,
120            ..
121        } = self.kind
122        {
123            *t = Some(trace_id);
124            *s = Some(span_id);
125        }
126        self
127    }
128
129    pub fn with_tenant(mut self, tenant_id: String, subject_id: Option<String>) -> Self {
130        if let SinkEventKind::Decision {
131            tenant_id: ref mut t,
132            subject_id: ref mut s,
133            ..
134        } = self.kind
135        {
136            *t = Some(tenant_id);
137            *s = subject_id;
138        }
139        self
140    }
141
142    pub fn mark_cached(mut self) -> Self {
143        if let SinkEventKind::Decision {
144            cached: ref mut c, ..
145        } = self.kind
146        {
147            *c = true;
148        }
149        self
150    }
151}
152
153/// Sink for telemetry events.
154///
155/// Implementations should be cheap to clone (typically `Arc<Sink>`) and
156/// must be safe to call from multiple threads concurrently.
157///
158/// # Why `async_trait`
159/// Rust's native `async fn` in traits doesn't support `dyn Sink`
160/// without `BoxFuture` wrappers, and adding the bound requires
161/// lifetime gymnastics at every call site. The `async_trait` macro
162/// gives object safety for the cost of one `Box<dyn Future>` per
163/// call. The Sink trait is consumed in only a handful of hot paths;
164/// if/when those migrate to generics we can drop `async_trait`.
165#[async_trait]
166pub trait Sink: Send + Sync {
167    /// Stable identifier for this sink (e.g. `"jsonl"`, `"stdout"`, `"otlp"`).
168    fn name(&self) -> &str;
169
170    /// Emit one event. Implementations should be non-blocking when possible.
171    async fn emit(&self, event: &SinkEvent) -> Result<(), SinkError>;
172
173    /// Flush any buffered events. Default is no-op.
174    async fn flush(&self) -> Result<(), SinkError> {
175        Ok(())
176    }
177
178    /// Release resources (close file handles, flush OTLP exporter, etc).
179    /// Default is no-op.
180    async fn shutdown(&self) -> Result<(), SinkError> {
181        Ok(())
182    }
183}
184
185#[cfg(test)]
186mod tests {
187    use super::{SinkEvent, SinkEventKind};
188
189    #[test]
190    fn decision_event_builders_attach_trace_tenant_and_cache_metadata() {
191        let event = SinkEvent::decision("allow", "agent", "read", "repo", vec![], vec![], 12)
192            .with_trace("trace-id".into(), "span-id".into())
193            .with_tenant("tenant-a".into(), Some("subject-a".into()))
194            .mark_cached();
195        let SinkEventKind::Decision {
196            trace_id,
197            span_id,
198            tenant_id,
199            subject_id,
200            cached,
201            ..
202        } = event.kind
203        else {
204            panic!("decision constructor returned another event kind");
205        };
206        assert_eq!(trace_id.as_deref(), Some("trace-id"));
207        assert_eq!(span_id.as_deref(), Some("span-id"));
208        assert_eq!(tenant_id.as_deref(), Some("tenant-a"));
209        assert_eq!(subject_id.as_deref(), Some("subject-a"));
210        assert!(cached);
211    }
212
213    #[test]
214    fn decision_metadata_builders_leave_other_event_kinds_unchanged() {
215        let event = SinkEvent {
216            id: uuid::Uuid::new_v4(),
217            timestamp: chrono::Utc::now(),
218            kind: SinkEventKind::PdpError {
219                error: "unavailable".into(),
220                fallback: "deny".into(),
221            },
222        }
223        .with_trace("trace-id".into(), "span-id".into())
224        .with_tenant("tenant-a".into(), Some("subject-a".into()))
225        .mark_cached();
226
227        assert!(matches!(event.kind, SinkEventKind::PdpError { .. }));
228    }
229}
agentguard_telemetry/sink.rs

agentguard_telemetry/
sink.rs
