dwara_core/extensions/
analytics.rs

1//! Analytics sink extension point.
2//!
3//! # Contract: [`AnalyticsSink`]
4//!
5//! **Purpose:** accept request/flow events emitted by the dataplane. The
6//! sink is fire-and-forget from the caller's perspective.
7//!
8//! **Semantics:** `record` appends one event and must not block the caller
9//! beyond a bounded enqueue. Implementations bound themselves (ring buffer,
10//! batch, drop-oldest) — an analytics overload must never stall the
11//! dataplane. Ordering is best-effort; events may be coalesced or dropped
12//! under pressure, and `record` returning `Ok` means "accepted", not
13//! "durably persisted". Events must not contain secret material.
14//!
15//! **Failure model:** only enqueue failures (closed channel, full local
16//! buffer) produce [`ExtensionsError::Backend`]; call sites should log and
17//! continue, never fail the request. No retries.
18//!
19//! **Editions:** OSS ships [`InMemoryAnalyticsSink`] (bounded ring buffer)
20//! and the DW-043 embedded store (`analytics::EmbeddedAnalytics`, SQLite
21//! rollups), which implements this contract on top of the richer
22//! per-request fields below. Ent planned: federated sink + warehouse
23//! export (DW-095). The raw-record firehose (DW-121) deliberately does
24//! NOT implement this contract: it streams the access record itself
25//! (with `request_id` and the redacted path this event shape omits)
26//! through its own sink seam — see `events::stream`.
27
28use std::collections::VecDeque;
29use std::sync::Mutex;
30
31use async_trait::async_trait;
32
33use super::ExtensionsError;
34
35/// A single analytics event.
36///
37/// Deliberately minimal and extensible: new fields are additive. `kind`
38/// selects the event type (e.g. `request`); optional route/consumer
39/// identifiers carry the common correlation keys, and `attributes` holds
40/// everything else as string pairs to avoid a schema lock-in at M1. The
41/// DW-043 request-completion fields (listener, method, duration,
42/// outcomes) were added additively so the embedded store consumes this
43/// type directly; `attributes` carries custom analytics dimensions.
44#[derive(Debug, Clone, PartialEq)]
45#[non_exhaustive]
46pub struct Event {
47    /// Event type discriminator, e.g. `request`.
48    pub kind: String,
49    /// Unix epoch milliseconds.
50    pub timestamp_ms: u64,
51    /// Route name, when the event is tied to a matched route.
52    pub route: Option<String>,
53    /// Consumer name, when authenticated.
54    pub consumer: Option<String>,
55    /// Upstream endpoint the request was forwarded to, if any.
56    pub endpoint: Option<String>,
57    /// Upstream response status code, if a response was produced.
58    pub status: Option<u16>,
59    /// Listener that accepted the request (DW-043).
60    pub listener: Option<String>,
61    /// HTTP method (DW-043).
62    pub method: Option<String>,
63    /// End-to-end request duration in milliseconds (DW-043).
64    pub duration_ms: Option<f64>,
65    /// Upstream attempts made (DW-043).
66    pub attempts: Option<u32>,
67    /// The request was rejected by a rate limit (DW-043).
68    pub rate_limited: bool,
69    /// The request failed through an open breaker (DW-043).
70    pub broken: bool,
71    /// The request was shed by admission control (DW-043).
72    pub shed: bool,
73    /// The edge instance ID this event originated from (SCALE-07,
74    /// #186). `None` for single-process (OSS) deployments; set by
75    /// the controller's `EmbeddedCollector` when forwarding federated
76    /// events from edges to the aggregate store.
77    pub edge_id: Option<String>,
78    /// Additional string key/value data (custom dimensions ride here).
79    pub attributes: Vec<(String, String)>,
80}
81
82impl Event {
83    /// Build a `request` event stamped with the current time.
84    pub fn request_now() -> Self {
85        Self {
86            kind: "request".to_owned(),
87            timestamp_ms: std::time::SystemTime::now()
88                .duration_since(std::time::UNIX_EPOCH)
89                .map(|d| d.as_millis() as u64)
90                .unwrap_or(0),
91            route: None,
92            consumer: None,
93            endpoint: None,
94            status: None,
95            listener: None,
96            method: None,
97            duration_ms: None,
98            attempts: None,
99            rate_limited: false,
100            broken: false,
101            shed: false,
102            edge_id: None,
103            attributes: Vec::new(),
104        }
105    }
106}
107
108/// Swappable analytics event sink.
109#[async_trait]
110pub trait AnalyticsSink: Send + Sync {
111    /// Accept one event for eventual processing.
112    async fn record(&self, event: Event) -> Result<(), ExtensionsError>;
113}
114
115/// Bounded in-memory ring-buffer sink (OSS skeleton).
116///
117/// Keeps the most recent `capacity` events; `record` on a full buffer drops
118/// the oldest entry. DW-021 replaces the consumer side, not this contract.
119#[derive(Debug)]
120pub struct InMemoryAnalyticsSink {
121    events: Mutex<VecDeque<Event>>,
122    capacity: usize,
123}
124
125impl InMemoryAnalyticsSink {
126    /// New sink retaining the most recent `capacity` events.
127    pub fn new(capacity: usize) -> Self {
128        Self {
129            events: Mutex::new(VecDeque::with_capacity(capacity)),
130            capacity,
131        }
132    }
133
134    /// Snapshot the currently retained events, oldest first.
135    pub fn events(&self) -> Vec<Event> {
136        self.events
137            .lock()
138            .expect("analytics state poisoned")
139            .clone()
140            .into()
141    }
142}
143
144#[async_trait]
145impl AnalyticsSink for InMemoryAnalyticsSink {
146    async fn record(&self, event: Event) -> Result<(), ExtensionsError> {
147        let mut events = self.events.lock().expect("analytics state poisoned");
148        if events.len() == self.capacity {
149            events.pop_front();
150        }
151        events.push_back(event);
152        Ok(())
153    }
154}
dwara_core/extensions/analytics.rs

dwara_core/extensions/
analytics.rs
