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

saddle_observability/
call.rs
