saddle_observability/
call.rs

1use std::time::Instant;
2
3use saddle_core::{
4    ApplicationId, CallContext, ErrorKind, ModuleId, OperationId, SaddleError, ServiceId, SpanId,
5    TraceCorrelationId,
6};
7use serde_json::{Value, json};
8
9use crate::{
10    DomainEvent, EventLevel, InboundTrace, Observer, TraceIdError,
11    logger::LogRecord,
12    trace::{select_entry_trace_id, select_trace_id},
13};
14
15/// Framework boundaries that automatically produce start and finish records.
16#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub enum CallKind {
18    Decode,
19    Admission,
20    RuntimeDispatch,
21    Handler,
22    ExternalRequest,
23    Service,
24    Database,
25    Transaction,
26    ExternalDecision,
27    ExternalConnect,
28    ExternalUnary,
29    ExternalCall,
30    Finalizer,
31    Response,
32}
33
34impl CallKind {
35    const fn as_str(self) -> &'static str {
36        match self {
37            Self::Decode => "decode",
38            Self::Admission => "admission",
39            Self::RuntimeDispatch => "runtime_dispatch",
40            Self::Handler => "handler",
41            Self::ExternalRequest => "external_request",
42            Self::Service => "service",
43            Self::Database => "database",
44            Self::Transaction => "transaction",
45            Self::ExternalDecision => "external_decision",
46            Self::ExternalConnect => "external_connect",
47            Self::ExternalUnary => "external_unary",
48            Self::ExternalCall => "external_call",
49            Self::Finalizer => "finalizer",
50            Self::Response => "response",
51        }
52    }
53}
54
55/// Stable completion classification for framework call logs.
56#[derive(Clone, Copy, Debug, Eq, PartialEq)]
57pub enum CallOutcome {
58    Success,
59    Failure,
60    Abandoned,
61}
62
63impl CallOutcome {
64    const fn as_str(self) -> &'static str {
65        match self {
66            Self::Success => "success",
67            Self::Failure => "failure",
68            Self::Abandoned => "abandoned",
69        }
70    }
71}
72
73/// An in-flight framework boundary.
74///
75/// Dropping it without calling [`ActiveCall::succeed`] or
76/// [`ActiveCall::fail`] emits an `abandoned` finish record, which preserves a
77/// balanced trace when a future is cancelled or unwinds.
78pub struct ActiveCall {
79    observer: Observer,
80    context: CallContext,
81    parent_span_id: Option<SpanId>,
82    kind: CallKind,
83    started_at: Instant,
84    finished: bool,
85}
86
87struct SelectedEntryTrace {
88    internal: saddle_core::TraceId,
89    correlation: Option<TraceCorrelationId>,
90    source: InboundTrace,
91}
92
93impl Observer {
94    /// Starts an external request using the legacy compatibility policy.
95    ///
96    /// Invalid supplied identifiers are replaced for compatibility with 0.3.0
97    /// callers. Production entry points must use
98    /// [`Observer::start_external_call_checked`] so only a missing identifier
99    /// can cause generation.
100    pub fn start_external_call(
101        &self,
102        application: impl Into<ApplicationId>,
103        module: impl Into<ModuleId>,
104        service: impl Into<ServiceId>,
105        operation: impl Into<OperationId>,
106        inbound_trace_id: Option<&str>,
107    ) -> (ActiveCall, InboundTrace) {
108        let (trace_id, inbound) = select_trace_id(inbound_trace_id, || self.new_trace_id());
109        self.start_selected_external_call(
110            application,
111            module,
112            service,
113            operation,
114            SelectedEntryTrace {
115                internal: trace_id,
116                correlation: None,
117                source: inbound,
118            },
119        )
120    }
121
122    /// Starts a production entry call. A supplied opaque protocol identifier
123    /// is inherited byte-for-byte after bounded validation; only an absent
124    /// identifier may cause generation.
125    pub fn start_external_call_checked(
126        &self,
127        application: impl Into<ApplicationId>,
128        module: impl Into<ModuleId>,
129        service: impl Into<ServiceId>,
130        operation: impl Into<OperationId>,
131        inbound_trace_id: Option<&str>,
132    ) -> Result<(ActiveCall, InboundTrace), TraceIdError> {
133        let (trace_id, trace_correlation_id, inbound) =
134            select_entry_trace_id(inbound_trace_id, || self.new_trace_id())?;
135        Ok(self.start_selected_external_call(
136            application,
137            module,
138            service,
139            operation,
140            SelectedEntryTrace {
141                internal: trace_id,
142                correlation: Some(trace_correlation_id),
143                source: inbound,
144            },
145        ))
146    }
147
148    fn start_selected_external_call(
149        &self,
150        application: impl Into<ApplicationId>,
151        module: impl Into<ModuleId>,
152        service: impl Into<ServiceId>,
153        operation: impl Into<OperationId>,
154        trace: SelectedEntryTrace,
155    ) -> (ActiveCall, InboundTrace) {
156        let mut context = CallContext::new(
157            application.into(),
158            module.into(),
159            service.into(),
160            operation.into(),
161            trace.internal,
162            self.new_span_id(),
163        );
164        if let Some(trace_correlation_id) = trace.correlation {
165            context = context.with_trace_correlation_id(trace_correlation_id);
166        }
167        (
168            ActiveCall::start(
169                self.clone(),
170                context,
171                None,
172                CallKind::ExternalRequest,
173                Some(trace.source),
174            ),
175            trace.source,
176        )
177    }
178
179    /// Starts a Service, DB or transaction child call on the parent's trace.
180    pub fn start_child_call(
181        &self,
182        parent: &CallContext,
183        kind: CallKind,
184        module: impl Into<ModuleId>,
185        service: impl Into<ServiceId>,
186        operation: impl Into<OperationId>,
187    ) -> ActiveCall {
188        let context = CallContext::new(
189            parent.application().clone(),
190            module.into(),
191            service.into(),
192            operation.into(),
193            parent.trace_id(),
194            self.new_span_id(),
195        )
196        .with_trace_correlation_id(parent.trace_correlation_id().clone());
197        ActiveCall::start(self.clone(), context, Some(parent.span_id()), kind, None)
198    }
199
200    /// Records an intentional domain event without accepting arbitrary objects.
201    pub fn record_domain_event(&self, context: &CallContext, event: DomainEvent) {
202        let mut record = LogRecord::new(event.level, "domain.event");
203        add_context(&mut record, context, None);
204        add_identity(&mut record, context);
205        record
206            .data
207            .insert("name".to_owned(), Value::String(event.name));
208        let fields = event
209            .fields
210            .into_iter()
211            .map(|(name, value)| (name, value.into_json()))
212            .collect();
213        record
214            .data
215            .insert("fields".to_owned(), Value::Object(fields));
216        self.emit(record);
217    }
218}
219
220impl ActiveCall {
221    fn start(
222        observer: Observer,
223        context: CallContext,
224        parent_span_id: Option<SpanId>,
225        kind: CallKind,
226        inbound_trace: Option<InboundTrace>,
227    ) -> Self {
228        let mut record = LogRecord::new(EventLevel::Info, "framework.call.started");
229        add_context(&mut record, &context, parent_span_id);
230        add_call_identity(&mut record, &context, kind);
231        if let Some(inbound_trace) = inbound_trace {
232            record.data.insert(
233                "trace_source".to_owned(),
234                Value::String(inbound_trace.as_str().to_owned()),
235            );
236        }
237        observer.emit(record);
238
239        Self {
240            observer,
241            context,
242            parent_span_id,
243            kind,
244            started_at: Instant::now(),
245            finished: false,
246        }
247    }
248
249    pub fn context(&self) -> &CallContext {
250        &self.context
251    }
252
253    pub fn succeed(mut self) {
254        self.finish(CallOutcome::Success, None);
255    }
256
257    pub fn fail(mut self, error: &SaddleError) {
258        self.finish(CallOutcome::Failure, Some(error));
259    }
260
261    fn finish(&mut self, outcome: CallOutcome, error: Option<&SaddleError>) {
262        if self.finished {
263            return;
264        }
265
266        let level = match outcome {
267            CallOutcome::Success => EventLevel::Info,
268            CallOutcome::Failure | CallOutcome::Abandoned => EventLevel::Error,
269        };
270        let mut record = LogRecord::new(level, "framework.call.finished");
271        add_context(&mut record, &self.context, self.parent_span_id);
272        add_call_identity(&mut record, &self.context, self.kind);
273        record.data.insert(
274            "outcome".to_owned(),
275            Value::String(outcome.as_str().to_owned()),
276        );
277        let duration = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX);
278        record.data.insert("duration".to_owned(), json!(duration));
279        record.data.insert("elapsed_us".to_owned(), json!(duration));
280        if let Some(error) = error {
281            record.data.insert(
282                "error_kind".to_owned(),
283                Value::String(error_kind(error.kind()).to_owned()),
284            );
285            record.data.insert(
286                "error_code".to_owned(),
287                Value::String(error.code().to_owned()),
288            );
289        }
290        self.observer.emit(record);
291        self.finished = true;
292    }
293}
294
295impl Drop for ActiveCall {
296    fn drop(&mut self) {
297        self.finish(CallOutcome::Abandoned, None);
298    }
299}
300
301fn add_context(record: &mut LogRecord, context: &CallContext, parent_span_id: Option<SpanId>) {
302    record.trace_id = Some(context.trace_correlation_id().to_string());
303    record.span_id = Some(context.span_id().to_string());
304    record.parent = parent_span_id.map(|span_id| span_id.to_string());
305    record.parent_span_id = parent_span_id.map(|span_id| span_id.to_string());
306    record.data.insert(
307        "rpc_id".to_owned(),
308        Value::String(context.span_id().to_string()),
309    );
310    record.data.insert(
311        "request".to_owned(),
312        Value::String(context.operation().to_string()),
313    );
314}
315
316fn add_call_identity(record: &mut LogRecord, context: &CallContext, kind: CallKind) {
317    record.span = Some(kind.as_str().to_owned());
318    record
319        .data
320        .insert("call".to_owned(), Value::String(kind.as_str().to_owned()));
321    record.data.insert(
322        "call_kind".to_owned(),
323        Value::String(kind.as_str().to_owned()),
324    );
325    add_identity(record, context);
326}
327
328fn add_identity(record: &mut LogRecord, context: &CallContext) {
329    record.data.insert(
330        "app".to_owned(),
331        Value::String(context.application().to_string()),
332    );
333    record.data.insert(
334        "application".to_owned(),
335        Value::String(context.application().to_string()),
336    );
337    record.data.insert(
338        "zone".to_owned(),
339        Value::String(context.module().to_string()),
340    );
341    record.data.insert(
342        "interface".to_owned(),
343        Value::String(context.service().to_string()),
344    );
345    record.data.insert(
346        "module".to_owned(),
347        Value::String(context.module().to_string()),
348    );
349    record.data.insert(
350        "service".to_owned(),
351        Value::String(context.service().to_string()),
352    );
353    record.data.insert(
354        "operation".to_owned(),
355        Value::String(context.operation().to_string()),
356    );
357}
358
359const fn error_kind(kind: ErrorKind) -> &'static str {
360    match kind {
361        ErrorKind::InvalidArgument => "invalid_argument",
362        ErrorKind::NotFound => "not_found",
363        ErrorKind::Conflict => "conflict",
364        ErrorKind::Business => "business",
365        ErrorKind::Unavailable => "unavailable",
366        ErrorKind::Infrastructure => "infrastructure",
367        ErrorKind::Internal => "internal",
368        _ => "unknown",
369    }
370}
371
372#[cfg(test)]
373mod tests {
374    use std::{
375        future::Future,
376        io,
377        sync::{Arc, Mutex},
378        task::{Context, Poll, Wake, Waker},
379        thread,
380    };
381
382    use saddle_core::{ErrorKind, SaddleError};
383
384    use super::*;
385    use crate::ObserverConfig;
386
387    struct ThreadWaker(thread::Thread);
388
389    impl Wake for ThreadWaker {
390        fn wake(self: Arc<Self>) {
391            self.0.unpark();
392        }
393    }
394
395    fn block_on<T>(future: impl Future<Output = T>) -> T {
396        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
397        let mut context = Context::from_waker(&waker);
398        let mut future = std::pin::pin!(future);
399        loop {
400            match future.as_mut().poll(&mut context) {
401                Poll::Ready(output) => return output,
402                Poll::Pending => thread::park(),
403            }
404        }
405    }
406
407    #[derive(Clone, Default)]
408    struct Capture(Arc<Mutex<Vec<u8>>>);
409
410    impl io::Write for Capture {
411        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
412            self.0.lock().unwrap().extend_from_slice(bytes);
413            Ok(bytes.len())
414        }
415
416        fn flush(&mut self) -> io::Result<()> {
417            Ok(())
418        }
419    }
420
421    fn records(capture: &Capture) -> Vec<Value> {
422        String::from_utf8(capture.0.lock().unwrap().clone())
423            .unwrap()
424            .lines()
425            .map(|line| serde_json::from_str(line).unwrap())
426            .collect()
427    }
428
429    #[test]
430    fn external_and_child_calls_share_trace_and_record_parentage() {
431        let capture = Capture::default();
432        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
433        let trace = "00112233445566778899aabbccddeeff";
434        let (external, source) =
435            observer.start_external_call("shop", "orders", "orders", "create", Some(trace));
436        assert_eq!(source, InboundTrace::Inherited);
437        assert_eq!(external.context().trace_id().to_string(), trace);
438
439        let child = observer.start_child_call(
440            external.context(),
441            CallKind::Service,
442            "users",
443            "users",
444            "validate",
445        );
446        assert_eq!(child.context().trace_id(), external.context().trace_id());
447        assert_ne!(child.context().span_id(), external.context().span_id());
448        child.succeed();
449        external.succeed();
450        block_on(observer.flush()).unwrap();
451
452        let records = records(&capture);
453        assert_eq!(records.len(), 4);
454        assert_eq!(records[1]["parent_span_id"], records[0]["span_id"]);
455        assert!(records.iter().all(|record| record["trace_id"] == trace));
456    }
457
458    #[test]
459    fn full_stage_chain_has_safe_schema_and_unique_external_rpc_ids() {
460        let capture = Capture::default();
461        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
462        let trace = "00112233445566778899aabbccddeeff";
463        let (root, _) =
464            observer.start_external_call("shop", "entry", "orders", "create", Some(trace));
465        let stages = [
466            CallKind::Decode,
467            CallKind::Admission,
468            CallKind::RuntimeDispatch,
469            CallKind::Handler,
470            CallKind::Database,
471            CallKind::ExternalDecision,
472            CallKind::Finalizer,
473            CallKind::Response,
474        ];
475        for stage in stages {
476            observer
477                .start_child_call(root.context(), stage, "framework", "orders", "create")
478                .succeed();
479        }
480        observer
481            .start_child_call(
482                root.context(),
483                CallKind::ExternalCall,
484                "rpc",
485                "inventory",
486                "reserve",
487            )
488            .succeed();
489        observer
490            .start_child_call(
491                root.context(),
492                CallKind::ExternalCall,
493                "rpc",
494                "payment",
495                "authorize",
496            )
497            .succeed();
498        root.succeed();
499        block_on(observer.flush()).unwrap();
500
501        let records = records(&capture);
502        let started: Vec<_> = records
503            .iter()
504            .filter(|record| record["event"] == "framework.call.started")
505            .collect();
506        assert_eq!(started.len(), 11);
507        assert!(started.iter().all(|record| record["trace_id"] == trace));
508        for record in &started {
509            for field in [
510                "trace_id",
511                "rpc_id",
512                "span",
513                "request",
514                "call",
515                "app",
516                "interface",
517                "zone",
518            ] {
519                assert!(record.get(field).is_some(), "missing {field}");
520            }
521            let serialized = serde_json::to_string(record).unwrap();
522            for forbidden in [
523                "payload",
524                "cookie",
525                "session",
526                "requestCtx",
527                "user_id",
528                "password",
529                "secret",
530                "db_value",
531            ] {
532                assert!(!serialized.contains(forbidden));
533            }
534        }
535        let external_rpc_ids: Vec<_> = started
536            .iter()
537            .filter(|record| record["span"] == "external_call")
538            .map(|record| record["rpc_id"].as_str().unwrap())
539            .collect();
540        assert_eq!(external_rpc_ids.len(), 2);
541        assert_ne!(external_rpc_ids[0], external_rpc_ids[1]);
542        assert!(started.iter().skip(1).all(|record| {
543            record["parent"] == started[0]["rpc_id"] && record["rpc_id"] != started[0]["rpc_id"]
544        }));
545        assert!(
546            records
547                .iter()
548                .filter(|record| record["event"] == "framework.call.finished")
549                .all(|record| record.get("outcome").is_some() && record.get("duration").is_some())
550        );
551    }
552
553    #[test]
554    fn invalid_inbound_trace_is_rejected_without_generating_a_replacement() {
555        let capture = Capture::default();
556        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
557        let result = observer.start_external_call_checked(
558            "shop",
559            "orders",
560            "orders",
561            "create",
562            Some("not-a-valid-trace\n"),
563        );
564        assert!(matches!(result, Err(TraceIdError::ControlCharacter)));
565        block_on(observer.flush()).unwrap();
566        assert!(records(&capture).is_empty());
567    }
568
569    #[test]
570    fn failure_logs_classification_but_not_sensitive_message() {
571        let capture = Capture::default();
572        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
573        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
574        let error = SaddleError::new(
575            ErrorKind::Infrastructure,
576            "db.query_failed",
577            "password=must-not-be-logged",
578        );
579        call.fail(&error);
580        block_on(observer.flush()).unwrap();
581
582        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
583        assert!(output.contains("db.query_failed"));
584        assert!(output.contains("infrastructure"));
585        assert!(!output.contains("must-not-be-logged"));
586    }
587
588    #[test]
589    fn domain_event_contains_only_explicit_scalar_fields() {
590        let capture = Capture::default();
591        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
592        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
593        observer.record_domain_event(
594            call.context(),
595            DomainEvent::new("order.created")
596                .unwrap()
597                .field("order_id", "o-1")
598                .unwrap()
599                .field("item_count", 2)
600                .unwrap(),
601        );
602        call.succeed();
603        block_on(observer.flush()).unwrap();
604
605        let records = records(&capture);
606        assert_eq!(records[1]["event"], "domain.event");
607        assert_eq!(records[1]["name"], "order.created");
608        assert_eq!(records[1]["fields"]["order_id"], "o-1");
609        assert_eq!(records[1]["fields"]["item_count"], 2);
610    }
611
612    #[test]
613    fn dropped_call_is_recorded_as_abandoned() {
614        let capture = Capture::default();
615        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
616        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
617        drop(call);
618        block_on(observer.flush()).unwrap();
619        let records = records(&capture);
620        assert_eq!(records[1]["outcome"], "abandoned");
621    }
622}
saddle_observability/call.rs

saddle_observability/
call.rs
