saddle_observability/
call.rs

1use std::time::Instant;
2
3use saddle_core::{
4    ApplicationId, CallContext, ErrorKind, ModuleId, OperationId, SaddleError, ServiceId, SpanId,
5};
6use serde_json::{Value, json};
7
8use crate::{
9    DomainEvent, EventLevel, InboundTrace, Observer, logger::LogRecord, trace::select_trace_id,
10};
11
12/// Framework boundaries that automatically produce start and finish records.
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14pub enum CallKind {
15    ExternalRequest,
16    Service,
17    Database,
18    Transaction,
19}
20
21impl CallKind {
22    const fn as_str(self) -> &'static str {
23        match self {
24            Self::ExternalRequest => "external_request",
25            Self::Service => "service",
26            Self::Database => "database",
27            Self::Transaction => "transaction",
28        }
29    }
30}
31
32/// Stable completion classification for framework call logs.
33#[derive(Clone, Copy, Debug, Eq, PartialEq)]
34pub enum CallOutcome {
35    Success,
36    Failure,
37    Abandoned,
38}
39
40impl CallOutcome {
41    const fn as_str(self) -> &'static str {
42        match self {
43            Self::Success => "success",
44            Self::Failure => "failure",
45            Self::Abandoned => "abandoned",
46        }
47    }
48}
49
50/// An in-flight framework boundary.
51///
52/// Dropping it without calling [`ActiveCall::succeed`] or
53/// [`ActiveCall::fail`] emits an `abandoned` finish record, which preserves a
54/// balanced trace when a future is cancelled or unwinds.
55pub struct ActiveCall {
56    observer: Observer,
57    context: CallContext,
58    parent_span_id: Option<SpanId>,
59    kind: CallKind,
60    started_at: Instant,
61    finished: bool,
62}
63
64impl Observer {
65    /// Starts an external request, inheriting a valid trace id or creating one.
66    pub fn start_external_call(
67        &self,
68        application: impl Into<ApplicationId>,
69        module: impl Into<ModuleId>,
70        service: impl Into<ServiceId>,
71        operation: impl Into<OperationId>,
72        inbound_trace_id: Option<&str>,
73    ) -> (ActiveCall, InboundTrace) {
74        let (trace_id, inbound) = select_trace_id(inbound_trace_id, || self.new_trace_id());
75        let context = CallContext::new(
76            application.into(),
77            module.into(),
78            service.into(),
79            operation.into(),
80            trace_id,
81            self.new_span_id(),
82        );
83        (
84            ActiveCall::start(
85                self.clone(),
86                context,
87                None,
88                CallKind::ExternalRequest,
89                Some(inbound),
90            ),
91            inbound,
92        )
93    }
94
95    /// Starts a Service, DB or transaction child call on the parent's trace.
96    pub fn start_child_call(
97        &self,
98        parent: &CallContext,
99        kind: CallKind,
100        module: impl Into<ModuleId>,
101        service: impl Into<ServiceId>,
102        operation: impl Into<OperationId>,
103    ) -> ActiveCall {
104        let context = CallContext::new(
105            parent.application().clone(),
106            module.into(),
107            service.into(),
108            operation.into(),
109            parent.trace_id(),
110            self.new_span_id(),
111        );
112        ActiveCall::start(self.clone(), context, Some(parent.span_id()), kind, None)
113    }
114
115    /// Records an intentional domain event without accepting arbitrary objects.
116    pub fn record_domain_event(&self, context: &CallContext, event: DomainEvent) {
117        let mut record = LogRecord::new(event.level, "domain.event");
118        add_context(&mut record, context, None);
119        add_identity(&mut record, context);
120        record
121            .data
122            .insert("name".to_owned(), Value::String(event.name));
123        let fields = event
124            .fields
125            .into_iter()
126            .map(|(name, value)| (name, value.into_json()))
127            .collect();
128        record
129            .data
130            .insert("fields".to_owned(), Value::Object(fields));
131        self.emit(record);
132    }
133}
134
135impl ActiveCall {
136    fn start(
137        observer: Observer,
138        context: CallContext,
139        parent_span_id: Option<SpanId>,
140        kind: CallKind,
141        inbound_trace: Option<InboundTrace>,
142    ) -> Self {
143        let mut record = LogRecord::new(EventLevel::Info, "framework.call.started");
144        add_context(&mut record, &context, parent_span_id);
145        add_call_identity(&mut record, &context, kind);
146        if let Some(inbound_trace) = inbound_trace {
147            record.data.insert(
148                "trace_source".to_owned(),
149                Value::String(inbound_trace.as_str().to_owned()),
150            );
151        }
152        observer.emit(record);
153
154        Self {
155            observer,
156            context,
157            parent_span_id,
158            kind,
159            started_at: Instant::now(),
160            finished: false,
161        }
162    }
163
164    pub fn context(&self) -> &CallContext {
165        &self.context
166    }
167
168    pub fn succeed(mut self) {
169        self.finish(CallOutcome::Success, None);
170    }
171
172    pub fn fail(mut self, error: &SaddleError) {
173        self.finish(CallOutcome::Failure, Some(error));
174    }
175
176    fn finish(&mut self, outcome: CallOutcome, error: Option<&SaddleError>) {
177        if self.finished {
178            return;
179        }
180
181        let level = match outcome {
182            CallOutcome::Success => EventLevel::Info,
183            CallOutcome::Failure | CallOutcome::Abandoned => EventLevel::Error,
184        };
185        let mut record = LogRecord::new(level, "framework.call.finished");
186        add_context(&mut record, &self.context, self.parent_span_id);
187        add_call_identity(&mut record, &self.context, self.kind);
188        record.data.insert(
189            "outcome".to_owned(),
190            Value::String(outcome.as_str().to_owned()),
191        );
192        record.data.insert(
193            "elapsed_us".to_owned(),
194            json!(u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX)),
195        );
196        if let Some(error) = error {
197            record.data.insert(
198                "error_kind".to_owned(),
199                Value::String(error_kind(error.kind()).to_owned()),
200            );
201            record.data.insert(
202                "error_code".to_owned(),
203                Value::String(error.code().to_owned()),
204            );
205        }
206        self.observer.emit(record);
207        self.finished = true;
208    }
209}
210
211impl Drop for ActiveCall {
212    fn drop(&mut self) {
213        self.finish(CallOutcome::Abandoned, None);
214    }
215}
216
217fn add_context(record: &mut LogRecord, context: &CallContext, parent_span_id: Option<SpanId>) {
218    record.trace_id = Some(context.trace_id().to_string());
219    record.span_id = Some(context.span_id().to_string());
220    record.parent_span_id = parent_span_id.map(|span_id| span_id.to_string());
221}
222
223fn add_call_identity(record: &mut LogRecord, context: &CallContext, kind: CallKind) {
224    record.data.insert(
225        "call_kind".to_owned(),
226        Value::String(kind.as_str().to_owned()),
227    );
228    add_identity(record, context);
229}
230
231fn add_identity(record: &mut LogRecord, context: &CallContext) {
232    record.data.insert(
233        "application".to_owned(),
234        Value::String(context.application().to_string()),
235    );
236    record.data.insert(
237        "module".to_owned(),
238        Value::String(context.module().to_string()),
239    );
240    record.data.insert(
241        "service".to_owned(),
242        Value::String(context.service().to_string()),
243    );
244    record.data.insert(
245        "operation".to_owned(),
246        Value::String(context.operation().to_string()),
247    );
248}
249
250const fn error_kind(kind: ErrorKind) -> &'static str {
251    match kind {
252        ErrorKind::InvalidArgument => "invalid_argument",
253        ErrorKind::NotFound => "not_found",
254        ErrorKind::Conflict => "conflict",
255        ErrorKind::Business => "business",
256        ErrorKind::Unavailable => "unavailable",
257        ErrorKind::Infrastructure => "infrastructure",
258        ErrorKind::Internal => "internal",
259        _ => "unknown",
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use std::{
266        future::Future,
267        io,
268        sync::{Arc, Mutex},
269        task::{Context, Poll, Wake, Waker},
270        thread,
271    };
272
273    use saddle_core::{ErrorKind, SaddleError};
274
275    use super::*;
276    use crate::ObserverConfig;
277
278    struct ThreadWaker(thread::Thread);
279
280    impl Wake for ThreadWaker {
281        fn wake(self: Arc<Self>) {
282            self.0.unpark();
283        }
284    }
285
286    fn block_on<T>(future: impl Future<Output = T>) -> T {
287        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
288        let mut context = Context::from_waker(&waker);
289        let mut future = std::pin::pin!(future);
290        loop {
291            match future.as_mut().poll(&mut context) {
292                Poll::Ready(output) => return output,
293                Poll::Pending => thread::park(),
294            }
295        }
296    }
297
298    #[derive(Clone, Default)]
299    struct Capture(Arc<Mutex<Vec<u8>>>);
300
301    impl io::Write for Capture {
302        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
303            self.0.lock().unwrap().extend_from_slice(bytes);
304            Ok(bytes.len())
305        }
306
307        fn flush(&mut self) -> io::Result<()> {
308            Ok(())
309        }
310    }
311
312    fn records(capture: &Capture) -> Vec<Value> {
313        String::from_utf8(capture.0.lock().unwrap().clone())
314            .unwrap()
315            .lines()
316            .map(|line| serde_json::from_str(line).unwrap())
317            .collect()
318    }
319
320    #[test]
321    fn external_and_child_calls_share_trace_and_record_parentage() {
322        let capture = Capture::default();
323        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
324        let trace = "00112233445566778899aabbccddeeff";
325        let (external, source) =
326            observer.start_external_call("shop", "orders", "orders", "create", Some(trace));
327        assert_eq!(source, InboundTrace::Inherited);
328        assert_eq!(external.context().trace_id().to_string(), trace);
329
330        let child = observer.start_child_call(
331            external.context(),
332            CallKind::Service,
333            "users",
334            "users",
335            "validate",
336        );
337        assert_eq!(child.context().trace_id(), external.context().trace_id());
338        assert_ne!(child.context().span_id(), external.context().span_id());
339        child.succeed();
340        external.succeed();
341        block_on(observer.flush()).unwrap();
342
343        let records = records(&capture);
344        assert_eq!(records.len(), 4);
345        assert_eq!(records[1]["parent_span_id"], records[0]["span_id"]);
346        assert!(records.iter().all(|record| record["trace_id"] == trace));
347    }
348
349    #[test]
350    fn invalid_inbound_trace_is_replaced_and_identified_in_start_record() {
351        let capture = Capture::default();
352        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
353        let (call, source) = observer.start_external_call(
354            "shop",
355            "orders",
356            "orders",
357            "create",
358            Some("not-a-valid-trace"),
359        );
360        assert_eq!(source, InboundTrace::ReplacedInvalid);
361        assert_ne!(call.context().trace_id().as_u128(), 0);
362        call.succeed();
363        block_on(observer.flush()).unwrap();
364
365        let records = records(&capture);
366        assert_eq!(records[0]["trace_source"], "replaced_invalid");
367    }
368
369    #[test]
370    fn failure_logs_classification_but_not_sensitive_message() {
371        let capture = Capture::default();
372        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
373        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
374        let error = SaddleError::new(
375            ErrorKind::Infrastructure,
376            "db.query_failed",
377            "password=must-not-be-logged",
378        );
379        call.fail(&error);
380        block_on(observer.flush()).unwrap();
381
382        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
383        assert!(output.contains("db.query_failed"));
384        assert!(output.contains("infrastructure"));
385        assert!(!output.contains("must-not-be-logged"));
386    }
387
388    #[test]
389    fn domain_event_contains_only_explicit_scalar_fields() {
390        let capture = Capture::default();
391        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
392        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
393        observer.record_domain_event(
394            call.context(),
395            DomainEvent::new("order.created")
396                .unwrap()
397                .field("order_id", "o-1")
398                .unwrap()
399                .field("item_count", 2)
400                .unwrap(),
401        );
402        call.succeed();
403        block_on(observer.flush()).unwrap();
404
405        let records = records(&capture);
406        assert_eq!(records[1]["event"], "domain.event");
407        assert_eq!(records[1]["name"], "order.created");
408        assert_eq!(records[1]["fields"]["order_id"], "o-1");
409        assert_eq!(records[1]["fields"]["item_count"], 2);
410    }
411
412    #[test]
413    fn dropped_call_is_recorded_as_abandoned() {
414        let capture = Capture::default();
415        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
416        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
417        drop(call);
418        block_on(observer.flush()).unwrap();
419        let records = records(&capture);
420        assert_eq!(records[1]["outcome"], "abandoned");
421    }
422}
saddle_observability/call.rs

saddle_observability/
call.rs
