1use std::{error::Error, fmt, time::Instant};
2
3use saddle_core::{CallContext, ErrorKind, SaddleError};
4use serde_json::{Value, json};
5
6use crate::{EventLevel, Observer, OutputStage, logger::LogRecord};
7
8const MAX_IDENTITY_BYTES: usize = 256;
9
10#[derive(Clone, Debug, Eq, PartialEq)]
11pub enum ChainFieldError {
12 Empty,
13 TooLong,
14 ControlCharacter,
15 UnsafeAuthority,
16 ZeroAttempt,
17}
18
19impl fmt::Display for ChainFieldError {
20 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
21 f.write_str(match self {
22 Self::Empty => "observability identity must not be empty",
23 Self::TooLong => "observability identity exceeds 256 bytes",
24 Self::ControlCharacter => "observability identity contains a control character",
25 Self::UnsafeAuthority => "outbound authority must be a credential-free target label",
26 Self::ZeroAttempt => "attempt must be greater than zero",
27 })
28 }
29}
30
31impl Error for ChainFieldError {}
32
33macro_rules! safe_identity {
34 ($name:ident) => {
35 #[derive(Clone, Debug, Eq, PartialEq)]
36 pub struct $name(String);
37
38 impl $name {
39 pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
40 validate_identity(value.into()).map(Self)
41 }
42
43 pub fn as_str(&self) -> &str {
44 &self.0
45 }
46 }
47 };
48}
49
50safe_identity!(RequestIdentity);
51safe_identity!(RouteIdentity);
52safe_identity!(BottleneckIdentity);
53safe_identity!(RejectReason);
54
55#[derive(Clone, Debug, Eq, PartialEq)]
56pub struct OutboundAuthority(String);
57
58impl OutboundAuthority {
59 pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
60 let value = validate_identity(value.into())?;
61 if value.contains(['@', '?', '#', '/', '\\']) || value.contains("://") {
62 return Err(ChainFieldError::UnsafeAuthority);
63 }
64 Ok(Self(value))
65 }
66}
67
68fn validate_identity(value: String) -> Result<String, ChainFieldError> {
69 if value.is_empty() {
70 return Err(ChainFieldError::Empty);
71 }
72 if value.len() > MAX_IDENTITY_BYTES {
73 return Err(ChainFieldError::TooLong);
74 }
75 if value.chars().any(char::is_control) {
76 return Err(ChainFieldError::ControlCharacter);
77 }
78 Ok(value)
79}
80
81#[derive(Clone, Copy, Debug, Eq, PartialEq)]
82pub enum Stage {
83 Ingress,
84 Admission,
85 Handler,
86 Database,
87 ProfuseContract,
88 Response,
89 ResourceFinalization,
90}
91
92impl Stage {
93 const fn as_str(self) -> &'static str {
94 match self {
95 Self::Ingress => "ingress",
96 Self::Admission => "admission",
97 Self::Handler => "handler",
98 Self::Database => "database",
99 Self::ProfuseContract => "profuse_contract",
100 Self::Response => "response",
101 Self::ResourceFinalization => "resource_finalization",
102 }
103 }
104}
105
106#[derive(Clone, Copy, Debug, Eq, PartialEq)]
107pub enum StageOutcome {
108 Success,
109 Failure,
110 Rejected,
111 Cancelled,
112}
113
114impl StageOutcome {
115 const fn as_str(self) -> &'static str {
116 match self {
117 Self::Success => "success",
118 Self::Failure => "failure",
119 Self::Rejected => "rejected",
120 Self::Cancelled => "cancelled",
121 }
122 }
123}
124
125#[derive(Clone, Debug, Eq, PartialEq)]
126pub struct EventContext {
127 request: RequestIdentity,
128 route: RouteIdentity,
129 attempt: u32,
130}
131
132impl EventContext {
133 pub fn new(
134 request: RequestIdentity,
135 route: RouteIdentity,
136 attempt: u32,
137 ) -> Result<Self, ChainFieldError> {
138 if attempt == 0 {
139 return Err(ChainFieldError::ZeroAttempt);
140 }
141 Ok(Self {
142 request,
143 route,
144 attempt,
145 })
146 }
147}
148
149pub struct ActiveStage {
150 observer: Observer,
151 context: CallContext,
152 parent_rpc_id: String,
153 stage: Stage,
154 request: RequestIdentity,
155 route: RouteIdentity,
156 attempt: u32,
157 started_at: Instant,
158 finished: bool,
159}
160
161impl Observer {
162 pub fn record_lifecycle_timeout(
163 &self,
164 application: &str,
165 stage: LifecycleTimeoutStage,
166 elapsed_ms: u64,
167 ) {
168 let Ok((call, _)) = self.start_external_call_checked(
169 application,
170 "runtime",
171 "lifecycle",
172 stage.as_str(),
173 None,
174 ) else {
175 return;
176 };
177 let mut record = base_record(
178 call.context(),
179 EventLevel::Error,
180 "framework.lifecycle.timeout",
181 "lifecycle",
182 );
183 record
184 .data
185 .insert("timeout_stage".into(), Value::String(stage.as_str().into()));
186 record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
187 record
188 .data
189 .insert("outcome".into(), Value::String("timeout".into()));
190 self.emit(record);
191 call.fail(&SaddleError::new(
192 ErrorKind::Infrastructure,
193 "runtime.lifecycle_timeout",
194 "managed lifecycle deadline elapsed",
195 ));
196 }
197
198 pub fn start_stage(
199 &self,
200 parent: &CallContext,
201 stage: Stage,
202 event_context: EventContext,
203 ) -> ActiveStage {
204 let context = CallContext::new(
205 parent.application().clone(),
206 parent.module().clone(),
207 parent.service().clone(),
208 parent.operation().clone(),
209 parent.trace_id(),
210 self.new_span_id(),
211 )
212 .with_trace_correlation_id(parent.trace_correlation_id().clone());
213 let active = ActiveStage {
214 observer: self.clone(),
215 context,
216 parent_rpc_id: parent.span_id().to_string(),
217 stage,
218 request: event_context.request,
219 route: event_context.route,
220 attempt: event_context.attempt,
221 started_at: Instant::now(),
222 finished: false,
223 };
224 active.emit_started();
225 active
226 }
227
228 pub fn record_capacity(
229 &self,
230 context: &CallContext,
231 event_context: &EventContext,
232 value: CapacityObservation,
233 ) {
234 self.inner
235 .metrics
236 .capacity(value.dimension, value.used, value.reject_reason.is_some());
237 let mut record = base_record(context, EventLevel::Info, "framework.capacity", "admission");
238 add_event_context(&mut record, event_context);
239 if let Some(dimension) = value.dimension {
240 record.data.insert(
241 "capacity_dimension".into(),
242 Value::String(dimension.as_str().into()),
243 );
244 }
245 record.data.insert("budget".into(), json!(value.budget));
246 record.data.insert("limit".into(), json!(value.limit));
247 record.data.insert("used".into(), json!(value.used));
248 record
249 .data
250 .insert("elapsed_ms".into(), json!(value.elapsed_ms));
251 record
252 .data
253 .insert("bottleneck".into(), Value::String(value.bottleneck.0));
254 record.data.insert(
255 "outcome".into(),
256 Value::String(
257 if value.reject_reason.is_some() {
258 "rejected"
259 } else {
260 "accepted"
261 }
262 .into(),
263 ),
264 );
265 if let Some(reason) = value.reject_reason {
266 record
267 .data
268 .insert("reject_reason".into(), Value::String(reason.0));
269 }
270 self.emit(record);
271 }
272
273 pub fn record_database_disposition(
274 &self,
275 context: &CallContext,
276 event_context: &EventContext,
277 disposition: DatabaseDisposition,
278 ) {
279 self.inner.metrics.database(disposition);
280 let mut record = base_record(
281 context,
282 EventLevel::Info,
283 "framework.database.disposition",
284 "database",
285 );
286 add_event_context(&mut record, event_context);
287 record.data.insert(
288 "db_disposition".into(),
289 Value::String(disposition.as_str().into()),
290 );
291 record
292 .data
293 .insert("outcome".into(), Value::String(disposition.as_str().into()));
294 self.emit(record);
295 }
296
297 pub fn record_resource_finalization(
298 &self,
299 context: &CallContext,
300 event_context: &EventContext,
301 disposition: DatabaseDisposition,
302 elapsed_ms: u64,
303 ) {
304 let mut record = base_record(
305 context,
306 EventLevel::Info,
307 "framework.resource.finalized",
308 "resource_finalization",
309 );
310 add_event_context(&mut record, event_context);
311 record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
312 record
313 .data
314 .insert("credit".into(), Value::String("released".into()));
315 record.data.insert(
316 "db_disposition".into(),
317 Value::String(disposition.as_str().into()),
318 );
319 record
320 .data
321 .insert("outcome".into(), Value::String("success".into()));
322 self.emit(record);
323 }
324
325 pub fn record_outbound(
326 &self,
327 context: &CallContext,
328 event_context: &EventContext,
329 observation: OutboundObservation,
330 ) {
331 self.inner.metrics.outbound(observation.result);
332 let mut record = base_record(
333 context,
334 EventLevel::Info,
335 "framework.outbound",
336 "profuse_contract",
337 );
338 add_event_context(&mut record, event_context);
339 record
340 .data
341 .insert("zone".into(), Value::String(observation.zone.0));
342 record
343 .data
344 .insert("authority".into(), Value::String(observation.authority.0));
345 record.data.insert(
346 "outbound_result".into(),
347 Value::String(observation.result.as_str().into()),
348 );
349 record.data.insert(
350 "outcome".into(),
351 Value::String(observation.result.as_str().into()),
352 );
353 self.emit(record);
354 }
355
356 pub fn record_lifecycle(
357 &self,
358 context: &CallContext,
359 event_context: &EventContext,
360 state: LifecycleState,
361 health: Health,
362 ) {
363 self.inner.metrics.lifecycle(state, health);
364 let mut record = base_record(
365 context,
366 if health == Health::Healthy {
367 EventLevel::Info
368 } else {
369 EventLevel::Error
370 },
371 "framework.lifecycle",
372 "resource_finalization",
373 );
374 add_event_context(&mut record, event_context);
375 record
376 .data
377 .insert("lifecycle".into(), Value::String(state.as_str().into()));
378 record
379 .data
380 .insert("health".into(), Value::String(health.as_str().into()));
381 record
382 .data
383 .insert("outcome".into(), Value::String(health.as_str().into()));
384 self.emit(record);
385 }
386
387 pub fn record_logger_health(
388 &self,
389 context: &CallContext,
390 event_context: &EventContext,
391 dropped: u64,
392 failure: Option<OutputStage>,
393 ) {
394 self.inner.metrics.logger_output(failure.is_some());
395 let mut record = base_record(
396 context,
397 if failure.is_some() || dropped > 0 {
398 EventLevel::Error
399 } else {
400 EventLevel::Info
401 },
402 "framework.logger.health",
403 "logger",
404 );
405 add_event_context(&mut record, event_context);
406 record.data.insert("logger_dropped".into(), json!(dropped));
407 record.data.insert(
408 "logger_health".into(),
409 Value::String(
410 if failure.is_some() {
411 "output_failed"
412 } else {
413 "healthy"
414 }
415 .into(),
416 ),
417 );
418 record.data.insert(
419 "outcome".into(),
420 Value::String(
421 if failure.is_some() {
422 "failure"
423 } else {
424 "success"
425 }
426 .into(),
427 ),
428 );
429 if let Some(stage) = failure {
430 record.data.insert(
431 "output_failed".into(),
432 Value::String(format!("{stage:?}").to_ascii_lowercase()),
433 );
434 }
435 self.emit(record);
436 }
437}
438
439#[derive(Clone, Copy, Debug, Eq, PartialEq)]
440pub enum LifecycleTimeoutStage {
441 ComponentStart,
442 RequestDrain,
443 ComponentShutdown,
444 PostDriverFinalization,
445}
446
447impl LifecycleTimeoutStage {
448 const fn as_str(self) -> &'static str {
449 match self {
450 Self::ComponentStart => "component_start",
451 Self::RequestDrain => "request_drain",
452 Self::ComponentShutdown => "component_shutdown",
453 Self::PostDriverFinalization => "post_driver_finalization",
454 }
455 }
456}
457
458impl ActiveStage {
459 pub fn context(&self) -> &CallContext {
460 &self.context
461 }
462 pub fn succeed(mut self) {
463 self.finish(StageOutcome::Success, None);
464 }
465 pub fn reject(mut self, error: &SaddleError) {
466 self.finish(StageOutcome::Rejected, Some(error));
467 }
468 pub fn fail(mut self, error: &SaddleError) {
469 self.finish(StageOutcome::Failure, Some(error));
470 }
471
472 fn emit_started(&self) {
473 let mut record = self.record(EventLevel::Info, "framework.stage.started");
474 record
475 .data
476 .insert("outcome".into(), Value::String("started".into()));
477 self.observer.emit(record);
478 }
479
480 fn finish(&mut self, outcome: StageOutcome, error: Option<&SaddleError>) {
481 if self.finished {
482 return;
483 }
484 let mut record = self.record(
485 if outcome == StageOutcome::Success {
486 EventLevel::Info
487 } else {
488 EventLevel::Error
489 },
490 "framework.stage.finished",
491 );
492 record
493 .data
494 .insert("outcome".into(), Value::String(outcome.as_str().into()));
495 let elapsed_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
496 record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
497 if let Some(error) = error {
498 record
499 .data
500 .insert("error_code".into(), Value::String(error.code().to_owned()));
501 record.data.insert(
502 "error_kind".into(),
503 Value::String(error_kind(error.kind()).into()),
504 );
505 }
506 self.observer
507 .inner
508 .metrics
509 .stage_finished(self.stage, outcome, elapsed_ms);
510 self.observer.emit(record);
511 self.finished = true;
512 }
513
514 fn record(&self, level: EventLevel, event: &'static str) -> LogRecord {
515 let mut record = base_record(&self.context, level, event, self.stage.as_str());
516 record.parent = Some(self.parent_rpc_id.clone());
517 record.parent_span_id = Some(self.parent_rpc_id.clone());
518 record.data.insert(
519 "request_identity".into(),
520 Value::String(self.request.0.clone()),
521 );
522 record
523 .data
524 .insert("route".into(), Value::String(self.route.0.clone()));
525 record.data.insert("attempt".into(), json!(self.attempt));
526 record.data.insert("elapsed_ms".into(), json!(0));
527 record
528 .data
529 .insert("error_code".into(), Value::String("none".into()));
530 record
531 }
532}
533
534impl Drop for ActiveStage {
535 fn drop(&mut self) {
536 self.finish(StageOutcome::Cancelled, None);
537 }
538}
539
540pub struct CapacityObservation {
541 dimension: Option<CapacityDimension>,
542 budget: u64,
543 limit: u64,
544 used: u64,
545 bottleneck: BottleneckIdentity,
546 reject_reason: Option<RejectReason>,
547 elapsed_ms: u64,
548}
549
550impl CapacityObservation {
551 pub fn accepted(budget: u64, limit: u64, used: u64, bottleneck: BottleneckIdentity) -> Self {
552 Self {
553 dimension: None,
554 budget,
555 limit,
556 used,
557 bottleneck,
558 reject_reason: None,
559 elapsed_ms: 0,
560 }
561 }
562 pub fn rejected(
563 budget: u64,
564 limit: u64,
565 used: u64,
566 bottleneck: BottleneckIdentity,
567 reason: RejectReason,
568 ) -> Self {
569 Self {
570 dimension: None,
571 budget,
572 limit,
573 used,
574 bottleneck,
575 reject_reason: Some(reason),
576 elapsed_ms: 0,
577 }
578 }
579
580 pub fn accepted_dimension(
581 dimension: CapacityDimension,
582 budget: u64,
583 limit: u64,
584 used: u64,
585 bottleneck: BottleneckIdentity,
586 elapsed_ms: u64,
587 ) -> Self {
588 Self {
589 dimension: Some(dimension),
590 budget,
591 limit,
592 used,
593 bottleneck,
594 reject_reason: None,
595 elapsed_ms,
596 }
597 }
598
599 pub fn rejected_dimension(
600 dimension: CapacityDimension,
601 budget: u64,
602 limit: u64,
603 used: u64,
604 bottleneck: BottleneckIdentity,
605 reason: RejectReason,
606 elapsed_ms: u64,
607 ) -> Self {
608 Self {
609 dimension: Some(dimension),
610 budget,
611 limit,
612 used,
613 bottleneck,
614 reject_reason: Some(reason),
615 elapsed_ms,
616 }
617 }
618}
619
620#[derive(Clone, Copy, Debug, Eq, PartialEq)]
621pub enum CapacityDimension {
622 Cpu,
623 Memory,
624 Database,
625 ProfuseContract,
626}
627
628impl CapacityDimension {
629 const fn as_str(self) -> &'static str {
630 match self {
631 Self::Cpu => "cpu",
632 Self::Memory => "memory",
633 Self::Database => "database",
634 Self::ProfuseContract => "profuse_contract",
635 }
636 }
637}
638
639#[derive(Clone, Copy, Debug, Eq, PartialEq)]
640pub enum DatabaseDisposition {
641 NotUsed,
642 Returned,
643 Discarded,
644}
645impl DatabaseDisposition {
646 const fn as_str(self) -> &'static str {
647 match self {
648 Self::NotUsed => "not_used",
649 Self::Returned => "returned",
650 Self::Discarded => "discarded",
651 }
652 }
653}
654
655pub struct OutboundObservation {
656 zone: RouteIdentity,
657 authority: OutboundAuthority,
658 result: OutboundResult,
659}
660impl OutboundObservation {
661 pub fn new(zone: RouteIdentity, authority: OutboundAuthority, result: OutboundResult) -> Self {
662 Self {
663 zone,
664 authority,
665 result,
666 }
667 }
668}
669
670#[derive(Clone, Copy, Debug, Eq, PartialEq)]
671pub enum OutboundResult {
672 Success,
673 Failure,
674 Rejected,
675 Timeout,
676}
677impl OutboundResult {
678 const fn as_str(self) -> &'static str {
679 match self {
680 Self::Success => "success",
681 Self::Failure => "failure",
682 Self::Rejected => "rejected",
683 Self::Timeout => "timeout",
684 }
685 }
686}
687
688#[derive(Clone, Copy, Debug, Eq, PartialEq)]
689pub enum LifecycleState {
690 Starting,
691 Running,
692 Draining,
693 Stopped,
694}
695impl LifecycleState {
696 const fn as_str(self) -> &'static str {
697 match self {
698 Self::Starting => "starting",
699 Self::Running => "running",
700 Self::Draining => "draining",
701 Self::Stopped => "stopped",
702 }
703 }
704}
705
706#[derive(Clone, Copy, Debug, Eq, PartialEq)]
707pub enum Health {
708 Healthy,
709 Degraded,
710 Failed,
711}
712impl Health {
713 const fn as_str(self) -> &'static str {
714 match self {
715 Self::Healthy => "healthy",
716 Self::Degraded => "degraded",
717 Self::Failed => "failed",
718 }
719 }
720}
721
722fn base_record(
723 context: &CallContext,
724 level: EventLevel,
725 event: &'static str,
726 stage: &'static str,
727) -> LogRecord {
728 let mut record = LogRecord::new(level, event);
729 record.trace_id = Some(context.trace_correlation_id().to_string());
730 record.span = Some(stage.into());
731 record.span_id = Some(context.span_id().to_string());
732 record
733 .data
734 .insert("timestamp".into(), json!(record.timestamp_unix_ms));
735 record
736 .data
737 .insert("stage".into(), Value::String(stage.into()));
738 record.data.insert(
739 "rpc_id".into(),
740 Value::String(context.span_id().to_string()),
741 );
742 record.data.insert(
743 "request".into(),
744 Value::String(context.operation().to_string()),
745 );
746 record
747}
748
749fn add_event_context(record: &mut LogRecord, context: &EventContext) {
750 record.data.insert(
751 "request_identity".into(),
752 Value::String(context.request.0.clone()),
753 );
754 record
755 .data
756 .insert("route".into(), Value::String(context.route.0.clone()));
757 record.data.insert("attempt".into(), json!(context.attempt));
758 record.data.insert("elapsed_ms".into(), json!(0));
759 record
760 .data
761 .insert("error_code".into(), Value::String("none".into()));
762}
763
764const fn error_kind(kind: ErrorKind) -> &'static str {
765 match kind {
766 ErrorKind::InvalidArgument => "invalid_argument",
767 ErrorKind::NotFound => "not_found",
768 ErrorKind::Conflict => "conflict",
769 ErrorKind::Business => "business",
770 ErrorKind::Unavailable => "unavailable",
771 ErrorKind::Infrastructure => "infrastructure",
772 ErrorKind::Internal => "internal",
773 _ => "unknown",
774 }
775}
776
777#[cfg(test)]
778mod tests {
779 use std::{
780 future::Future,
781 io,
782 sync::{Arc, Mutex},
783 task::{Context, Poll, Wake, Waker},
784 thread,
785 };
786
787 use super::*;
788 use crate::ObserverConfig;
789
790 #[derive(Clone, Default)]
791 struct Capture(Arc<Mutex<Vec<u8>>>);
792
793 impl io::Write for Capture {
794 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
795 self.0.lock().unwrap().extend_from_slice(bytes);
796 Ok(bytes.len())
797 }
798 fn flush(&mut self) -> io::Result<()> {
799 Ok(())
800 }
801 }
802
803 struct ThreadWaker(thread::Thread);
804 impl Wake for ThreadWaker {
805 fn wake(self: Arc<Self>) {
806 self.0.unpark();
807 }
808 }
809
810 fn block_on<T>(future: impl Future<Output = T>) -> T {
811 let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
812 let mut context = Context::from_waker(&waker);
813 let mut future = std::pin::pin!(future);
814 loop {
815 match future.as_mut().poll(&mut context) {
816 Poll::Ready(output) => return output,
817 Poll::Pending => thread::park(),
818 }
819 }
820 }
821
822 fn event_context(attempt: u32) -> EventContext {
823 EventContext::new(
824 RequestIdentity::new("request-7").unwrap(),
825 RouteIdentity::new("orders.create").unwrap(),
826 attempt,
827 )
828 .unwrap()
829 }
830
831 fn records(capture: &Capture) -> Vec<Value> {
832 String::from_utf8(capture.0.lock().unwrap().clone())
833 .unwrap()
834 .lines()
835 .map(|line| serde_json::from_str(line).unwrap())
836 .collect()
837 }
838
839 #[test]
840 fn lifecycle_timeout_is_typed_and_contains_no_application_payload() {
841 let capture = Capture::default();
842 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
843 observer.record_lifecycle_timeout(
844 "profusegw",
845 LifecycleTimeoutStage::ComponentShutdown,
846 30_000,
847 );
848 block_on(observer.flush()).unwrap();
849
850 let records = records(&capture);
851 let timeout = records
852 .iter()
853 .find(|record| record["event"] == "framework.lifecycle.timeout")
854 .unwrap();
855 assert_eq!(timeout["timeout_stage"], "component_shutdown");
856 assert_eq!(timeout["elapsed_ms"], 30_000);
857 assert_eq!(timeout["outcome"], "timeout");
858 assert!(timeout.get("payload").is_none());
859 }
860
861 #[test]
862 fn stage_chain_is_correlated_ordered_and_has_one_terminal() {
863 let capture = Capture::default();
864 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
865 let (root, _) = observer.start_external_call(
866 "shop",
867 "entry",
868 "orders",
869 "create",
870 Some("00112233445566778899aabbccddeeff"),
871 );
872 let stages = [
873 Stage::Ingress,
874 Stage::Admission,
875 Stage::Handler,
876 Stage::Database,
877 Stage::ProfuseContract,
878 Stage::Response,
879 Stage::ResourceFinalization,
880 ];
881 for (index, stage) in stages.into_iter().enumerate() {
882 observer
883 .start_stage(root.context(), stage, event_context((index + 1) as u32))
884 .succeed();
885 }
886 drop(observer.start_stage(root.context(), Stage::Handler, event_context(8)));
887 root.succeed();
888 block_on(observer.flush()).unwrap();
889
890 let records = records(&capture);
891 let stage_records: Vec<_> = records
892 .iter()
893 .filter(|value| {
894 value["event"]
895 .as_str()
896 .unwrap()
897 .starts_with("framework.stage.")
898 })
899 .collect();
900 assert_eq!(stage_records.len(), 16);
901 for pair in stage_records.chunks_exact(2) {
902 assert_eq!(pair[0]["event"], "framework.stage.started");
903 assert_eq!(pair[1]["event"], "framework.stage.finished");
904 assert_eq!(pair[0]["rpc_id"], pair[1]["rpc_id"]);
905 assert_eq!(pair[0]["trace_id"], "00112233445566778899aabbccddeeff");
906 for field in [
907 "timestamp_unix_ms",
908 "timestamp",
909 "level",
910 "event",
911 "stage",
912 "trace_id",
913 "rpc_id",
914 "request_identity",
915 "route",
916 "attempt",
917 "outcome",
918 "error_code",
919 ] {
920 assert!(pair[0].get(field).is_some(), "missing {field}");
921 }
922 assert!(pair[1].get("elapsed_ms").is_some());
923 }
924 assert_eq!(stage_records.last().unwrap()["outcome"], "cancelled");
925 }
926
927 #[test]
928 fn closed_observations_have_common_fields_and_no_sensitive_payload() {
929 let capture = Capture::default();
930 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
931 let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
932 let common = event_context(1);
933 observer.record_capacity(
934 root.context(),
935 &common,
936 CapacityObservation::rejected(
937 100,
938 80,
939 80,
940 BottleneckIdentity::new("request_slots").unwrap(),
941 RejectReason::new("limit_reached").unwrap(),
942 ),
943 );
944 observer.record_database_disposition(
945 root.context(),
946 &common,
947 DatabaseDisposition::Returned,
948 );
949 observer.record_outbound(
950 root.context(),
951 &common,
952 OutboundObservation::new(
953 RouteIdentity::new("cn-hz-a").unwrap(),
954 OutboundAuthority::new("inventory-service").unwrap(),
955 OutboundResult::Success,
956 ),
957 );
958 observer.record_lifecycle(
959 root.context(),
960 &common,
961 LifecycleState::Draining,
962 Health::Healthy,
963 );
964 observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
965 root.succeed();
966 let _ = block_on(observer.flush());
967
968 let records: Vec<_> = records(&capture)
969 .into_iter()
970 .filter(|value| {
971 matches!(
972 value["event"].as_str(),
973 Some(
974 "framework.capacity"
975 | "framework.database.disposition"
976 | "framework.outbound"
977 | "framework.lifecycle"
978 | "framework.logger.health"
979 )
980 )
981 })
982 .collect();
983 assert_eq!(records.len(), 5);
984 for record in records {
985 for field in [
986 "timestamp_unix_ms",
987 "timestamp",
988 "level",
989 "event",
990 "stage",
991 "trace_id",
992 "rpc_id",
993 "request_identity",
994 "route",
995 "attempt",
996 "elapsed_ms",
997 "outcome",
998 "error_code",
999 ] {
1000 assert!(record.get(field).is_some(), "missing {field}");
1001 }
1002 let encoded = serde_json::to_string(&record).unwrap();
1003 for forbidden in [
1004 "request_body",
1005 "response_body",
1006 "cookie",
1007 "session",
1008 "token",
1009 "password",
1010 "connection_string",
1011 "db_value",
1012 ] {
1013 assert!(!encoded.contains(forbidden));
1014 }
1015 }
1016 }
1017
1018 #[test]
1019 fn fixed_metrics_snapshot_tracks_closed_low_cardinality_events() {
1020 let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
1021 let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1022 let common = event_context(1);
1023 observer
1024 .start_stage(root.context(), Stage::Response, event_context(1))
1025 .succeed();
1026 observer.record_capacity(
1027 root.context(),
1028 &common,
1029 CapacityObservation::rejected_dimension(
1030 CapacityDimension::Database,
1031 100,
1032 80,
1033 80,
1034 BottleneckIdentity::new("database_slots").unwrap(),
1035 RejectReason::new("limit_reached").unwrap(),
1036 2,
1037 ),
1038 );
1039 observer.record_database_disposition(
1040 root.context(),
1041 &common,
1042 DatabaseDisposition::Discarded,
1043 );
1044 observer.record_outbound(
1045 root.context(),
1046 &common,
1047 OutboundObservation::new(
1048 RouteIdentity::new("cn-hz-a").unwrap(),
1049 OutboundAuthority::new("inventory-service").unwrap(),
1050 OutboundResult::Timeout,
1051 ),
1052 );
1053 observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
1054 observer.record_lifecycle(
1055 root.context(),
1056 &common,
1057 LifecycleState::Running,
1058 Health::Healthy,
1059 );
1060
1061 let snapshot = observer.metrics_snapshot();
1062 assert_eq!(snapshot.requests(StageOutcome::Success), 1);
1063 assert_eq!(
1064 snapshot.stage_latency(Stage::Response).iter().sum::<u64>(),
1065 1
1066 );
1067 assert_eq!(
1068 snapshot.capacity_rejected(Some(CapacityDimension::Database)),
1069 1
1070 );
1071 assert_eq!(
1072 snapshot.capacity_used(Some(CapacityDimension::Database)),
1073 80
1074 );
1075 assert_eq!(snapshot.database(DatabaseDisposition::Discarded), 1);
1076 assert_eq!(snapshot.outbound(OutboundResult::Timeout), 1);
1077 assert_eq!(snapshot.logger_dropped(), 0);
1078 assert_eq!(snapshot.logger_output_failed(), 1);
1079 assert_eq!(snapshot.lifecycle(), LifecycleState::Running);
1080 assert_eq!(snapshot.health(), Health::Healthy);
1081 assert!(snapshot.ready());
1082 root.succeed();
1083 }
1084
1085 #[test]
1086 fn typed_capacity_and_resource_terminal_preserve_closed_schema() {
1087 let capture = Capture::default();
1088 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1089 let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1090 let common = event_context(1);
1091 observer.record_capacity(
1092 root.context(),
1093 &common,
1094 CapacityObservation::rejected_dimension(
1095 CapacityDimension::Database,
1096 4,
1097 2,
1098 2,
1099 BottleneckIdentity::new("database").unwrap(),
1100 RejectReason::new("at_limit").unwrap(),
1101 17,
1102 ),
1103 );
1104 observer.record_resource_finalization(
1105 root.context(),
1106 &common,
1107 DatabaseDisposition::NotUsed,
1108 23,
1109 );
1110 root.succeed();
1111 block_on(observer.flush()).unwrap();
1112
1113 let records = records(&capture);
1114 let capacity = records
1115 .iter()
1116 .find(|record| record["event"] == "framework.capacity")
1117 .unwrap();
1118 assert_eq!(capacity["capacity_dimension"], "database");
1119 assert_eq!(capacity["budget"], 4);
1120 assert_eq!(capacity["limit"], 2);
1121 assert_eq!(capacity["used"], 2);
1122 assert_eq!(capacity["bottleneck"], "database");
1123 assert_eq!(capacity["reject_reason"], "at_limit");
1124 assert_eq!(capacity["elapsed_ms"], 17);
1125
1126 let terminal = records
1127 .iter()
1128 .find(|record| record["event"] == "framework.resource.finalized")
1129 .unwrap();
1130 assert_eq!(terminal["stage"], "resource_finalization");
1131 assert_eq!(terminal["credit"], "released");
1132 assert_eq!(terminal["db_disposition"], "not_used");
1133 assert_eq!(terminal["elapsed_ms"], 23);
1134 assert_eq!(terminal["outcome"], "success");
1135 assert_eq!(capacity["trace_id"], terminal["trace_id"]);
1136 assert_eq!(capacity["rpc_id"], terminal["rpc_id"]);
1137 assert_eq!(capacity["request_identity"], terminal["request_identity"]);
1138 assert_eq!(capacity["route"], terminal["route"]);
1139 assert_eq!(capacity["attempt"], terminal["attempt"]);
1140 }
1141
1142 #[test]
1143 fn unsafe_or_unbounded_identifiers_and_zero_attempt_are_rejected() {
1144 assert_eq!(RequestIdentity::new(""), Err(ChainFieldError::Empty));
1145 assert_eq!(
1146 RouteIdentity::new("x".repeat(257)),
1147 Err(ChainFieldError::TooLong)
1148 );
1149 assert_eq!(
1150 RejectReason::new("bad\nreason"),
1151 Err(ChainFieldError::ControlCharacter)
1152 );
1153 for unsafe_value in ["https://user:secret@host/path", "host/path", "host?token=x"] {
1154 assert_eq!(
1155 OutboundAuthority::new(unsafe_value),
1156 Err(ChainFieldError::UnsafeAuthority)
1157 );
1158 }
1159 assert_eq!(
1160 EventContext::new(
1161 RequestIdentity::new("r").unwrap(),
1162 RouteIdentity::new("route").unwrap(),
1163 0
1164 ),
1165 Err(ChainFieldError::ZeroAttempt),
1166 );
1167 }
1168}