appcore_ops/
observation.rs

1//! Bounded, tool-independent runtime observation events.
2
3use appcore_core::redact_text_with_limit;
4use parking_lot::{Mutex, RwLock};
5use serde::{Deserialize, Serialize};
6use std::collections::{BTreeMap, VecDeque};
7use std::sync::Arc;
8
9/// Maximum retained attributes per observation.
10pub const MAX_OBSERVATION_ATTRIBUTES: usize = 32;
11/// Maximum UTF-8 bytes retained in an observation name.
12pub const MAX_OBSERVATION_NAME_BYTES: usize = 128;
13/// Maximum UTF-8 bytes retained in an observation attribute key.
14pub const MAX_OBSERVATION_KEY_BYTES: usize = 64;
15/// Maximum UTF-8 bytes retained in an observation attribute value.
16pub const MAX_OBSERVATION_VALUE_BYTES: usize = 1_024;
17/// Maximum UTF-8 bytes retained in a trace identifier.
18pub const MAX_OBSERVATION_TRACE_BYTES: usize = 256;
19
20/// Runtime subsystem that produced an observation.
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum ObservationKind {
24    /// Process lifecycle.
25    Lifecycle,
26    /// Runtime or deployment configuration.
27    Configuration,
28    /// Health evaluation.
29    Health,
30    /// Authentication, authorization or secret boundary.
31    Security,
32    /// Storage operation.
33    Storage,
34    /// Control-plane operation.
35    ControlPlane,
36    /// Direct peer RPC operation.
37    PeerRpc,
38    /// Scheduler operation.
39    Scheduler,
40    /// Synchronization operation.
41    Sync,
42    /// Audit operation.
43    Audit,
44    /// Diagnostic operation.
45    Diagnostic,
46}
47
48/// Severity of one runtime observation.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum ObservationSeverity {
52    /// Developer diagnostic fact.
53    Debug,
54    /// Normal operational fact.
55    Info,
56    /// Recoverable or degraded condition.
57    Warning,
58    /// Failed operation.
59    Error,
60}
61
62/// Generic runtime fact suitable for logs, diagnostics, metrics and audit sinks.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64pub struct ObservationEvent {
65    /// Runtime subsystem category.
66    pub kind: ObservationKind,
67    /// Observation severity.
68    pub severity: ObservationSeverity,
69    /// Stable observation name.
70    pub name: String,
71    /// Timestamp in Unix milliseconds.
72    pub timestamp_ms: u64,
73    /// Optional trace identity.
74    pub trace_id: Option<String>,
75    /// Bounded non-sensitive dimensions.
76    pub attributes: BTreeMap<String, String>,
77}
78
79impl ObservationEvent {
80    /// Creates an event without attributes.
81    pub fn new(
82        kind: ObservationKind,
83        severity: ObservationSeverity,
84        name: impl Into<String>,
85        timestamp_ms: u64,
86    ) -> Self {
87        Self {
88            kind,
89            severity,
90            name: name.into(),
91            timestamp_ms,
92            trace_id: None,
93            attributes: BTreeMap::new(),
94        }
95    }
96
97    /// Attaches a trace identity.
98    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
99        self.trace_id = Some(trace_id.into());
100        self
101    }
102
103    /// Adds one attribute. Sensitive keys and values are redacted immediately.
104    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
105        let key = redact_text_with_limit(&key.into(), MAX_OBSERVATION_KEY_BYTES);
106        if self.attributes.len() >= MAX_OBSERVATION_ATTRIBUTES
107            && !self.attributes.contains_key(&key)
108        {
109            return self;
110        }
111        let value = if is_sensitive_key(&key) {
112            "[REDACTED]".to_string()
113        } else {
114            redact_text_with_limit(&value.into(), MAX_OBSERVATION_VALUE_BYTES)
115        };
116        self.attributes.insert(key, value);
117        self
118    }
119
120    pub(crate) fn redacted(mut self) -> Self {
121        self.name = redact_text_with_limit(&self.name, MAX_OBSERVATION_NAME_BYTES);
122        self.trace_id = self
123            .trace_id
124            .map(|value| redact_text_with_limit(&value, MAX_OBSERVATION_TRACE_BYTES));
125        while self.attributes.len() > MAX_OBSERVATION_ATTRIBUTES {
126            let Some(key) = self.attributes.keys().next_back().cloned() else {
127                break;
128            };
129            self.attributes.remove(&key);
130        }
131        for (key, value) in &mut self.attributes {
132            *value = if is_sensitive_key(key) {
133                "[REDACTED]".to_string()
134            } else {
135                redact_text_with_limit(value, MAX_OBSERVATION_VALUE_BYTES)
136            };
137        }
138        self
139    }
140}
141
142/// Destination for generic observation events.
143pub trait ObservationSink: Send + Sync {
144    /// Emits one event without blocking on external tooling.
145    fn emit(&self, event: ObservationEvent);
146}
147
148/// Bounded in-memory sink used by diagnostics and embedded runtimes.
149#[derive(Clone)]
150pub struct InMemoryObservationSink {
151    capacity: usize,
152    events: Arc<Mutex<VecDeque<ObservationEvent>>>,
153    drains: Arc<RwLock<Vec<Arc<dyn ObservationSink>>>>,
154}
155
156impl std::fmt::Debug for InMemoryObservationSink {
157    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
158        formatter
159            .debug_struct("InMemoryObservationSink")
160            .field("capacity", &self.capacity)
161            .field("event_count", &self.len())
162            .field("drain_count", &self.drains.read().len())
163            .finish()
164    }
165}
166
167impl InMemoryObservationSink {
168    /// Creates a sink that retains the newest `capacity` events.
169    pub fn new(capacity: usize) -> Self {
170        Self {
171            capacity: capacity.max(1),
172            events: Arc::new(Mutex::new(VecDeque::with_capacity(capacity.max(1)))),
173            drains: Arc::new(RwLock::new(Vec::new())),
174        }
175    }
176
177    /// Adds an operational drain that receives future redacted events.
178    pub fn add_drain(&self, drain: Arc<dyn ObservationSink>) {
179        self.drains.write().push(drain);
180    }
181
182    /// Returns the number of attached operational drains.
183    pub fn drain_count(&self) -> usize {
184        self.drains.read().len()
185    }
186
187    /// Returns a stable snapshot from oldest to newest.
188    pub fn snapshot(&self) -> Vec<ObservationEvent> {
189        self.events.lock().iter().cloned().collect()
190    }
191
192    /// Returns the number of retained events.
193    pub fn len(&self) -> usize {
194        self.events.lock().len()
195    }
196
197    /// Reports whether no events are retained.
198    pub fn is_empty(&self) -> bool {
199        self.events.lock().is_empty()
200    }
201}
202
203impl Default for InMemoryObservationSink {
204    fn default() -> Self {
205        Self::new(1_024)
206    }
207}
208
209impl ObservationSink for InMemoryObservationSink {
210    fn emit(&self, event: ObservationEvent) {
211        let event = event.redacted();
212        let mut events = self.events.lock();
213        if events.len() == self.capacity {
214            let _ = events.pop_front();
215        }
216        events.push_back(event.clone());
217        drop(events);
218        let drains = self.drains.read().clone();
219        for drain in drains {
220            drain.emit(event.clone());
221        }
222    }
223}
224
225fn is_sensitive_key(key: &str) -> bool {
226    let normalized = key.to_ascii_lowercase();
227    ["secret", "password", "token", "credential", "private_key"]
228        .iter()
229        .any(|fragment| normalized.contains(fragment))
230}
231
232#[cfg(test)]
233mod tests {
234    use super::*;
235
236    #[test]
237    fn bounded_sink_discards_oldest_event() {
238        let sink = InMemoryObservationSink::new(2);
239        for index in 0..3 {
240            sink.emit(ObservationEvent::new(
241                ObservationKind::Lifecycle,
242                ObservationSeverity::Info,
243                format!("runtime.event.{index}"),
244                index,
245            ));
246        }
247        let snapshot = sink.snapshot();
248        assert_eq!(snapshot.len(), 2);
249        assert_eq!(snapshot[0].name, "runtime.event.1");
250    }
251
252    #[test]
253    fn sink_redacts_sensitive_attributes() {
254        let sink = InMemoryObservationSink::new(2);
255        sink.emit(
256            ObservationEvent::new(
257                ObservationKind::Security,
258                ObservationSeverity::Warning,
259                "security.rejected",
260                1,
261            )
262            .with_attribute("access_token", "raw-secret"),
263        );
264        assert_eq!(sink.snapshot()[0].attributes["access_token"], "[REDACTED]");
265    }
266
267    #[test]
268    fn sink_bounds_names_values_and_attribute_count() {
269        let mut event = ObservationEvent::new(
270            ObservationKind::Diagnostic,
271            ObservationSeverity::Info,
272            "n".repeat(1_000),
273            1,
274        );
275        for index in 0..100 {
276            event = event.with_attribute(format!("key-{index}"), "v".repeat(2_000));
277        }
278        let sink = InMemoryObservationSink::new(1);
279        sink.emit(event);
280        let event = &sink.snapshot()[0];
281
282        assert!(event.name.len() <= MAX_OBSERVATION_NAME_BYTES);
283        assert_eq!(event.attributes.len(), MAX_OBSERVATION_ATTRIBUTES);
284        assert!(event
285            .attributes
286            .values()
287            .all(|value| value.len() <= MAX_OBSERVATION_VALUE_BYTES));
288    }
289}
appcore_ops/observation.rs

appcore_ops/
observation.rs
