oxide_batch/
telemetry.rs

1//! Versioned, bounded, and non-authoritative telemetry contracts.
2//!
3//! Durable repository state remains the only correctness authority. Every
4//! sink boundary in this module is panic-isolated, queues drop rather than
5//! backpressure execution, and metric labels come from a closed catalog.
6
7use std::collections::{BTreeMap, BTreeSet, VecDeque};
8use std::fmt;
9use std::panic::{AssertUnwindSafe, catch_unwind};
10use std::sync::{Arc, Mutex};
11use std::time::Duration;
12
13use futures_util::FutureExt;
14
15use crate::{
16    ActorRef, AuthorizationClass, BatchStatus, BoxFuture, DiagnosticField, EventComponent,
17    EventSeverity, JobExecutionId, JobName, MetricLabel, OperationId, OperatorAction,
18    OperatorOutcomeClass, OperatorRejection, OperatorRequest, PurgeCounts, ReasonCode,
19    RecoveryProposal, RetentionAction, RetentionOutcome, StepName,
20};
21
22/// The stable M4 telemetry schema version.
23pub const TELEMETRY_SCHEMA_VERSION: u16 = 1;
24/// Maximum distinct label combinations retained by one metric family.
25pub const METRIC_CARDINALITY_BUDGET: usize = 200;
26/// Maximum explicitly allowed job and step names.
27pub const MAX_METRIC_NAME_ALLOWLIST: usize = 50;
28/// Reserved value used for values outside an allowlist or cardinality budget.
29pub const OTHER_LABEL_VALUE: &str = "__other__";
30/// Minimum bounded exporter queue length.
31pub const MIN_EXPORT_QUEUE_RECORDS: usize = 64;
32/// Maximum bounded exporter queue length.
33pub const MAX_EXPORT_QUEUE_RECORDS: usize = 65_536;
34/// Default bounded exporter queue length.
35pub const DEFAULT_EXPORT_QUEUE_RECORDS: usize = 1_024;
36/// Minimum throttling window for drop notifications.
37pub const MIN_DROP_REPORT_WINDOW: Duration = Duration::from_secs(1);
38/// Maximum throttling window for drop notifications.
39pub const MAX_DROP_REPORT_WINDOW: Duration = Duration::from_hours(1);
40/// Default throttling window for drop notifications.
41pub const DEFAULT_DROP_REPORT_WINDOW: Duration = Duration::from_mins(1);
42/// Default retained events returned for one incident execution.
43pub const DEFAULT_RETAINED_EVENTS_PER_EXECUTION: usize = 200;
44/// Maximum retained events returned for one incident execution.
45pub const MAX_RETAINED_EVENTS_PER_EXECUTION: usize = 200;
46/// Default total retained events across executions.
47pub const DEFAULT_RETAINED_EVENT_CAPACITY: usize = 4_096;
48
49const JOB_SPAN_FIELDS: &[&str] = &[
50    "job.name",
51    "job.instance.id",
52    "job.execution.id",
53    "job.attempt",
54    "status",
55    "failure.category",
56    "failure.id",
57];
58const STEP_SPAN_FIELDS: &[&str] = &[
59    "job.name",
60    "job.instance.id",
61    "job.execution.id",
62    "job.attempt",
63    "step.name",
64    "step.execution.id",
65    "step.attempt",
66    "status",
67    "failure.category",
68    "failure.id",
69];
70const CHUNK_SPAN_FIELDS: &[&str] = &[
71    "job.execution.id",
72    "step.execution.id",
73    "chunk.sequence",
74    "status",
75    "failure.category",
76    "failure.id",
77];
78const ITEM_SPAN_FIELDS: &[&str] = &[
79    "job.execution.id",
80    "step.execution.id",
81    "chunk.sequence",
82    "outcome",
83    "failure.category",
84    "failure.id",
85];
86const RETRY_SPAN_FIELDS: &[&str] = &[
87    "job.execution.id",
88    "step.execution.id",
89    "retry.ordinal",
90    "outcome",
91    "failure.category",
92    "failure.id",
93];
94const BACKOFF_SPAN_FIELDS: &[&str] = &[
95    "job.execution.id",
96    "step.execution.id",
97    "retry.ordinal",
98    "backoff.duration_class",
99    "outcome",
100];
101
102/// One stable span in the telemetry schema version 1 hierarchy.
103#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
104#[non_exhaustive]
105pub enum TelemetrySpanKind {
106    /// The root span for one job execution attempt.
107    JobExecution,
108    /// One step execution attempt below its job execution.
109    StepExecution,
110    /// One bounded chunk attempt below its step execution.
111    ChunkAttempt,
112    /// The read phase of a chunk attempt.
113    ItemRead,
114    /// The process phase of a chunk attempt.
115    ItemProcess,
116    /// The write phase of a chunk attempt.
117    ItemWrite,
118    /// The repository commit phase of a chunk attempt.
119    RepositoryCommit,
120    /// One retry attempt below its step execution.
121    Retry,
122    /// The bounded wait below one retry attempt.
123    Backoff,
124}
125
126impl TelemetrySpanKind {
127    /// Returns the stable schema-version-1 span name.
128    #[must_use]
129    pub const fn as_str(self) -> &'static str {
130        match self {
131            Self::JobExecution => "job.execution",
132            Self::StepExecution => "step.execution",
133            Self::ChunkAttempt => "chunk.attempt",
134            Self::ItemRead => "item.read",
135            Self::ItemProcess => "item.process",
136            Self::ItemWrite => "item.write",
137            Self::RepositoryCommit => "repository.commit",
138            Self::Retry => "retry",
139            Self::Backoff => "backoff",
140        }
141    }
142
143    /// Returns the required direct parent in the stable hierarchy.
144    #[must_use]
145    pub const fn parent(self) -> Option<Self> {
146        match self {
147            Self::JobExecution => None,
148            Self::StepExecution => Some(Self::JobExecution),
149            Self::ChunkAttempt | Self::Retry => Some(Self::StepExecution),
150            Self::ItemRead | Self::ItemProcess | Self::ItemWrite | Self::RepositoryCommit => {
151                Some(Self::ChunkAttempt)
152            }
153            Self::Backoff => Some(Self::Retry),
154        }
155    }
156
157    /// Returns the framework component represented by this span.
158    #[must_use]
159    pub const fn component(self) -> EventComponent {
160        match self {
161            Self::JobExecution => EventComponent::Job,
162            Self::StepExecution => EventComponent::Step,
163            Self::ChunkAttempt => EventComponent::Chunk,
164            Self::ItemRead | Self::ItemProcess | Self::ItemWrite => EventComponent::Item,
165            Self::RepositoryCommit => EventComponent::Repository,
166            Self::Retry | Self::Backoff => EventComponent::Retry,
167        }
168    }
169
170    /// Returns the complete reviewed field-key set for this span.
171    #[must_use]
172    pub const fn safe_field_keys(self) -> &'static [&'static str] {
173        match self {
174            Self::JobExecution => JOB_SPAN_FIELDS,
175            Self::StepExecution => STEP_SPAN_FIELDS,
176            Self::ChunkAttempt | Self::RepositoryCommit => CHUNK_SPAN_FIELDS,
177            Self::ItemRead | Self::ItemProcess | Self::ItemWrite => ITEM_SPAN_FIELDS,
178            Self::Retry => RETRY_SPAN_FIELDS,
179            Self::Backoff => BACKOFF_SPAN_FIELDS,
180        }
181    }
182}
183
184impl fmt::Display for TelemetrySpanKind {
185    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
186        formatter.write_str(self.as_str())
187    }
188}
189
190/// Stable adapter-neutral span outcome classes.
191#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
192#[non_exhaustive]
193pub enum TelemetrySpanStatus {
194    /// No terminal outcome has been assigned yet.
195    Unset,
196    /// The observed work completed successfully.
197    Ok,
198    /// The observed work failed with a known outcome.
199    Error,
200    /// The observed work stopped cooperatively.
201    Cancelled,
202    /// The commit or execution outcome is unknown.
203    Unknown,
204}
205
206impl TelemetrySpanStatus {
207    /// Maps a durable lifecycle status to its adapter-neutral span outcome.
208    #[must_use]
209    pub const fn from_batch_status(status: BatchStatus) -> Self {
210        match status {
211            BatchStatus::Starting | BatchStatus::Started | BatchStatus::Stopping => Self::Unset,
212            BatchStatus::Stopped => Self::Cancelled,
213            BatchStatus::Failed | BatchStatus::Abandoned => Self::Error,
214            BatchStatus::Completed => Self::Ok,
215            // `BatchStatus` is `#[non_exhaustive]`, so a status this build
216            // does not know reports the outcome it has: unknown.
217            // `BatchStatus::Unknown`, and any status this build does not know:
218            // `BatchStatus` is `#[non_exhaustive]`, and an unrecognized status
219            // reports the outcome it has.
220            _ => Self::Unknown,
221        }
222    }
223
224    /// Returns the stable lowercase representation.
225    #[must_use]
226    pub const fn as_str(self) -> &'static str {
227        match self {
228            Self::Unset => "unset",
229            Self::Ok => "ok",
230            Self::Error => "error",
231            Self::Cancelled => "cancelled",
232            Self::Unknown => "unknown",
233        }
234    }
235}
236
237impl From<BatchStatus> for TelemetrySpanStatus {
238    fn from(status: BatchStatus) -> Self {
239        Self::from_batch_status(status)
240    }
241}
242
243impl fmt::Display for TelemetrySpanStatus {
244    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
245        formatter.write_str(self.as_str())
246    }
247}
248
249/// The complete telemetry schema version 1 span catalog.
250pub const TELEMETRY_SPAN_CATALOG: &[TelemetrySpanKind] = &[
251    TelemetrySpanKind::JobExecution,
252    TelemetrySpanKind::StepExecution,
253    TelemetrySpanKind::ChunkAttempt,
254    TelemetrySpanKind::ItemRead,
255    TelemetrySpanKind::ItemProcess,
256    TelemetrySpanKind::ItemWrite,
257    TelemetrySpanKind::RepositoryCommit,
258    TelemetrySpanKind::Retry,
259    TelemetrySpanKind::Backoff,
260];
261
262/// Stable timing of an event relative to the decision it observes.
263#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
264#[non_exhaustive]
265pub enum EventTiming {
266    /// Emitted after the governing durable commit returns successfully.
267    AfterCommit,
268    /// Emitted after a bounded read returns successfully.
269    AfterRead,
270    /// Emitted after durable evidence is gathered and before it is returned.
271    AfterEvidence,
272    /// Emitted when a non-durable runtime boundary is observed.
273    RuntimeBoundary,
274}
275
276/// One stable event in telemetry schema version 1.
277#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
278#[non_exhaustive]
279pub enum TelemetryEventKind {
280    /// A launch request entered the guarded operator boundary.
281    LaunchRequested,
282    /// A launch became durable.
283    LaunchAccepted,
284    /// A launch guard durably rejected the request.
285    LaunchRejected,
286    /// A job or step lifecycle event.
287    JobStarting,
288    /// A job lifecycle transition.
289    JobStarted,
290    /// A job lifecycle transition.
291    JobStopping,
292    /// A job lifecycle transition.
293    JobStopped,
294    /// A job lifecycle transition.
295    JobFailed,
296    /// A job lifecycle transition.
297    JobCompleted,
298    /// A job lifecycle transition.
299    JobAbandoned,
300    /// A job commit outcome is unknown.
301    JobUnknown,
302    /// A step lifecycle transition.
303    StepStarting,
304    /// A step lifecycle transition.
305    StepStarted,
306    /// A step lifecycle transition.
307    StepStopping,
308    /// A step lifecycle transition.
309    StepStopped,
310    /// A step lifecycle transition.
311    StepFailed,
312    /// A step lifecycle transition.
313    StepCompleted,
314    /// A step commit outcome is unknown.
315    StepUnknown,
316    /// Chunk work began.
317    ChunkStarted,
318    /// A chunk transaction committed.
319    ChunkCommitted,
320    /// A chunk transaction rolled back.
321    ChunkRolledBack,
322    /// A chunk commit outcome is unknown.
323    ChunkUnknown,
324    /// A job before-listener failed or panicked.
325    JobBeforeListenerFailed,
326    /// A job after-listener failed or panicked.
327    JobAfterListenerFailed,
328    /// A step before-listener failed or panicked.
329    StepBeforeListenerFailed,
330    /// A step after-listener failed or panicked.
331    StepAfterListenerFailed,
332    /// A retry reservation committed.
333    RetryReserved,
334    /// A retry backoff began.
335    RetryBackoffStarted,
336    /// A retry backoff was cancelled.
337    RetryBackoffCancelled,
338    /// A retry budget was exhausted.
339    RetryExhausted,
340    /// A skip became durable with its accepting chunk.
341    ItemSkipped,
342    /// A known rollback committed its classification.
343    FaultRollbackCommitted,
344    /// A commit-safe no-rollback classification committed.
345    FaultNoRollbackCommitted,
346    /// A checkpoint was loaded.
347    CheckpointLoaded,
348    /// A checkpoint committed.
349    CheckpointCommitted,
350    /// A repository optimistic conflict was observed.
351    RepositoryConflict,
352    /// A repository transient failure was observed.
353    RepositoryTransientFailure,
354    /// A flow step result committed.
355    FlowStepResultCommitted,
356    /// A flow decision committed.
357    FlowDecisionCommitted,
358    /// A completed flow step was reused.
359    FlowCompletedStepReused,
360    /// A step start limit rejected another start.
361    StepStartLimitExceeded,
362    /// An operator request and effect committed.
363    OperatorRequestAccepted,
364    /// An operator rejection audit committed.
365    OperatorRequestRejected,
366    /// An operator request completed or replayed.
367    OperatorRequestCompleted,
368    /// A bounded explorer page returned.
369    ExplorerPageServed,
370    /// Shutdown was requested.
371    ShutdownRequested,
372    /// New intake was stopped.
373    ShutdownIntakeStopped,
374    /// Every owned child joined.
375    ShutdownDrainCompleted,
376    /// A shutdown deadline elapsed.
377    ShutdownDeadlineExceeded,
378    /// Durable evidence identified a stale candidate.
379    StaleDetected,
380    /// A recovery proposal was produced.
381    RecoveryProposed,
382    /// A recovery decision and lifecycle change committed.
383    RecoveryApplied,
384    /// A recovery request was durably rejected.
385    RecoveryRejected,
386    /// A retention plan was produced.
387    RetentionPlanned,
388    /// A retention mutation committed.
389    RetentionApplied,
390    /// A retention request was durably rejected.
391    RetentionRejected,
392    /// A bounded split branch began.
393    SplitBranchStarted,
394    /// A bounded split branch ended.
395    SplitBranchCompleted,
396    /// A partition plan committed.
397    PartitionPlanCommitted,
398    /// A local partition assignment committed.
399    PartitionAssigned,
400    /// A local partition result committed.
401    PartitionCompleted,
402    /// Parent partition aggregation committed.
403    PartitionAggregated,
404    /// A bounded exporter dropped a record.
405    TelemetryExportDropped,
406    /// A metadata migration began.
407    MigrationStarted,
408    /// A metadata migration completed.
409    MigrationCompleted,
410    /// A metadata migration failed.
411    MigrationFailed,
412}
413
414impl TelemetryEventKind {
415    /// Returns the stable dotted catalog name.
416    #[must_use]
417    pub const fn as_str(self) -> &'static str {
418        match self {
419            Self::LaunchRequested => "launch.requested",
420            Self::LaunchAccepted => "launch.accepted",
421            Self::LaunchRejected => "launch.rejected",
422            Self::JobStarting => "job.starting",
423            Self::JobStarted => "job.started",
424            Self::JobStopping => "job.stopping",
425            Self::JobStopped => "job.stopped",
426            Self::JobFailed => "job.failed",
427            Self::JobCompleted => "job.completed",
428            Self::JobAbandoned => "job.abandoned",
429            Self::JobUnknown => "job.unknown",
430            Self::StepStarting => "step.starting",
431            Self::StepStarted => "step.started",
432            Self::StepStopping => "step.stopping",
433            Self::StepStopped => "step.stopped",
434            Self::StepFailed => "step.failed",
435            Self::StepCompleted => "step.completed",
436            Self::StepUnknown => "step.unknown",
437            Self::ChunkStarted => "chunk.started",
438            Self::ChunkCommitted => "chunk.committed",
439            Self::ChunkRolledBack => "chunk.rolled_back",
440            Self::ChunkUnknown => "chunk.unknown",
441            Self::JobBeforeListenerFailed => "job.before_listener.failed",
442            Self::JobAfterListenerFailed => "job.after_listener.failed",
443            Self::StepBeforeListenerFailed => "step.before_listener.failed",
444            Self::StepAfterListenerFailed => "step.after_listener.failed",
445            Self::RetryReserved => "retry.reserved",
446            Self::RetryBackoffStarted => "retry.backoff_started",
447            Self::RetryBackoffCancelled => "retry.backoff_cancelled",
448            Self::RetryExhausted => "retry.exhausted",
449            Self::ItemSkipped => "item.skipped",
450            Self::FaultRollbackCommitted => "fault.rollback_committed",
451            Self::FaultNoRollbackCommitted => "fault.no_rollback_committed",
452            Self::CheckpointLoaded => "checkpoint.loaded",
453            Self::CheckpointCommitted => "checkpoint.committed",
454            Self::RepositoryConflict => "repository.conflict",
455            Self::RepositoryTransientFailure => "repository.transient_failure",
456            Self::FlowStepResultCommitted => "flow.step_result_committed",
457            Self::FlowDecisionCommitted => "flow.decision_committed",
458            Self::FlowCompletedStepReused => "flow.completed_step_reused",
459            Self::StepStartLimitExceeded => "step.start_limit_exceeded",
460            Self::OperatorRequestAccepted => "operator.request_accepted",
461            Self::OperatorRequestRejected => "operator.request_rejected",
462            Self::OperatorRequestCompleted => "operator.request_completed",
463            Self::ExplorerPageServed => "explorer.page_served",
464            Self::ShutdownRequested => "shutdown.requested",
465            Self::ShutdownIntakeStopped => "shutdown.intake_stopped",
466            Self::ShutdownDrainCompleted => "shutdown.drain_completed",
467            Self::ShutdownDeadlineExceeded => "shutdown.deadline_exceeded",
468            Self::StaleDetected => "stale.detected",
469            Self::RecoveryProposed => "recovery.proposed",
470            Self::RecoveryApplied => "recovery.applied",
471            Self::RecoveryRejected => "recovery.rejected",
472            Self::RetentionPlanned => "retention.planned",
473            Self::RetentionApplied => "retention.applied",
474            Self::RetentionRejected => "retention.rejected",
475            Self::SplitBranchStarted => "split.branch_started",
476            Self::SplitBranchCompleted => "split.branch_completed",
477            Self::PartitionPlanCommitted => "partition.plan_committed",
478            Self::PartitionAssigned => "partition.assigned",
479            Self::PartitionCompleted => "partition.completed",
480            Self::PartitionAggregated => "partition.aggregated",
481            Self::TelemetryExportDropped => "telemetry.export_dropped",
482            Self::MigrationStarted => "migration.started",
483            Self::MigrationCompleted => "migration.completed",
484            Self::MigrationFailed => "migration.failed",
485        }
486    }
487
488    /// Returns the stable severity.
489    #[must_use]
490    pub const fn severity(self) -> EventSeverity {
491        match self {
492            Self::ExplorerPageServed => EventSeverity::Debug,
493            Self::JobFailed
494            | Self::StepFailed
495            | Self::JobUnknown
496            | Self::StepUnknown
497            | Self::ChunkUnknown
498            | Self::JobBeforeListenerFailed
499            | Self::JobAfterListenerFailed
500            | Self::StepBeforeListenerFailed
501            | Self::StepAfterListenerFailed
502            | Self::RetryExhausted
503            | Self::ShutdownDeadlineExceeded
504            | Self::MigrationFailed => EventSeverity::Error,
505            Self::LaunchRejected
506            | Self::JobStopping
507            | Self::JobStopped
508            | Self::StepStopping
509            | Self::StepStopped
510            | Self::ChunkRolledBack
511            | Self::RetryBackoffCancelled
512            | Self::ItemSkipped
513            | Self::FaultRollbackCommitted
514            | Self::FaultNoRollbackCommitted
515            | Self::OperatorRequestRejected
516            | Self::RecoveryRejected
517            | Self::RetentionRejected
518            | Self::TelemetryExportDropped => EventSeverity::Warn,
519            _ => EventSeverity::Info,
520        }
521    }
522
523    /// Returns the framework component.
524    #[must_use]
525    pub const fn component(self) -> EventComponent {
526        match self {
527            Self::LaunchRequested | Self::LaunchAccepted | Self::LaunchRejected => {
528                EventComponent::Launcher
529            }
530            Self::JobStarting
531            | Self::JobStarted
532            | Self::JobStopping
533            | Self::JobStopped
534            | Self::JobFailed
535            | Self::JobCompleted
536            | Self::JobAbandoned
537            | Self::JobUnknown => EventComponent::Job,
538            Self::StepStarting
539            | Self::StepStarted
540            | Self::StepStopping
541            | Self::StepStopped
542            | Self::StepFailed
543            | Self::StepCompleted
544            | Self::StepUnknown
545            | Self::StepStartLimitExceeded => EventComponent::Step,
546            Self::ChunkStarted
547            | Self::ChunkCommitted
548            | Self::ChunkRolledBack
549            | Self::ChunkUnknown => EventComponent::Chunk,
550            Self::JobBeforeListenerFailed
551            | Self::JobAfterListenerFailed
552            | Self::StepBeforeListenerFailed
553            | Self::StepAfterListenerFailed => EventComponent::Listener,
554            Self::RetryReserved
555            | Self::RetryBackoffStarted
556            | Self::RetryBackoffCancelled
557            | Self::RetryExhausted => EventComponent::Retry,
558            Self::ItemSkipped => EventComponent::Item,
559            Self::FaultRollbackCommitted | Self::FaultNoRollbackCommitted => EventComponent::Fault,
560            Self::CheckpointLoaded | Self::CheckpointCommitted => EventComponent::Checkpoint,
561            Self::RepositoryConflict | Self::RepositoryTransientFailure => {
562                EventComponent::Repository
563            }
564            Self::FlowStepResultCommitted
565            | Self::FlowDecisionCommitted
566            | Self::FlowCompletedStepReused => EventComponent::Flow,
567            Self::OperatorRequestAccepted
568            | Self::OperatorRequestRejected
569            | Self::OperatorRequestCompleted => EventComponent::Operator,
570            Self::ExplorerPageServed => EventComponent::Explorer,
571            Self::ShutdownRequested
572            | Self::ShutdownIntakeStopped
573            | Self::ShutdownDrainCompleted
574            | Self::ShutdownDeadlineExceeded => EventComponent::Shutdown,
575            Self::StaleDetected
576            | Self::RecoveryProposed
577            | Self::RecoveryApplied
578            | Self::RecoveryRejected => EventComponent::Recovery,
579            Self::RetentionPlanned | Self::RetentionApplied | Self::RetentionRejected => {
580                EventComponent::Retention
581            }
582            Self::SplitBranchStarted | Self::SplitBranchCompleted => EventComponent::Split,
583            Self::PartitionPlanCommitted
584            | Self::PartitionAssigned
585            | Self::PartitionCompleted
586            | Self::PartitionAggregated => EventComponent::Partition,
587            Self::TelemetryExportDropped => EventComponent::Telemetry,
588            Self::MigrationStarted | Self::MigrationCompleted | Self::MigrationFailed => {
589                EventComponent::Migration
590            }
591        }
592    }
593
594    /// Returns when the event is emitted relative to its observation.
595    #[must_use]
596    pub const fn timing(self) -> EventTiming {
597        match self {
598            Self::OperatorRequestAccepted
599            | Self::OperatorRequestRejected
600            | Self::OperatorRequestCompleted
601            | Self::RecoveryApplied
602            | Self::RecoveryRejected
603            | Self::RetentionApplied
604            | Self::RetentionRejected
605            | Self::PartitionPlanCommitted
606            | Self::PartitionAssigned
607            | Self::PartitionCompleted
608            | Self::PartitionAggregated
609            | Self::LaunchAccepted
610            | Self::LaunchRejected
611            | Self::JobStarting
612            | Self::JobStarted
613            | Self::JobStopping
614            | Self::JobStopped
615            | Self::JobFailed
616            | Self::JobCompleted
617            | Self::JobAbandoned
618            | Self::JobUnknown
619            | Self::StepStarting
620            | Self::StepStarted
621            | Self::StepStopping
622            | Self::StepStopped
623            | Self::StepFailed
624            | Self::StepCompleted
625            | Self::StepUnknown
626            | Self::ChunkCommitted
627            | Self::ChunkRolledBack
628            | Self::ChunkUnknown
629            | Self::JobBeforeListenerFailed
630            | Self::JobAfterListenerFailed
631            | Self::StepBeforeListenerFailed
632            | Self::StepAfterListenerFailed
633            | Self::RetryReserved
634            | Self::RetryExhausted
635            | Self::ItemSkipped
636            | Self::FaultRollbackCommitted
637            | Self::FaultNoRollbackCommitted
638            | Self::CheckpointCommitted
639            | Self::FlowStepResultCommitted
640            | Self::FlowDecisionCommitted => EventTiming::AfterCommit,
641            Self::ExplorerPageServed => EventTiming::AfterRead,
642            Self::StaleDetected | Self::RecoveryProposed | Self::RetentionPlanned => {
643                EventTiming::AfterEvidence
644            }
645            _ => EventTiming::RuntimeBoundary,
646        }
647    }
648}
649
650impl fmt::Display for TelemetryEventKind {
651    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
652        formatter.write_str(self.as_str())
653    }
654}
655
656/// The complete telemetry schema version 1 event catalog.
657pub const TELEMETRY_EVENT_CATALOG: &[TelemetryEventKind] = &[
658    TelemetryEventKind::LaunchRequested,
659    TelemetryEventKind::LaunchAccepted,
660    TelemetryEventKind::LaunchRejected,
661    TelemetryEventKind::JobStarting,
662    TelemetryEventKind::JobStarted,
663    TelemetryEventKind::JobStopping,
664    TelemetryEventKind::JobStopped,
665    TelemetryEventKind::JobFailed,
666    TelemetryEventKind::JobCompleted,
667    TelemetryEventKind::JobAbandoned,
668    TelemetryEventKind::JobUnknown,
669    TelemetryEventKind::StepStarting,
670    TelemetryEventKind::StepStarted,
671    TelemetryEventKind::StepStopping,
672    TelemetryEventKind::StepStopped,
673    TelemetryEventKind::StepFailed,
674    TelemetryEventKind::StepCompleted,
675    TelemetryEventKind::StepUnknown,
676    TelemetryEventKind::ChunkStarted,
677    TelemetryEventKind::ChunkCommitted,
678    TelemetryEventKind::ChunkRolledBack,
679    TelemetryEventKind::ChunkUnknown,
680    TelemetryEventKind::JobBeforeListenerFailed,
681    TelemetryEventKind::JobAfterListenerFailed,
682    TelemetryEventKind::StepBeforeListenerFailed,
683    TelemetryEventKind::StepAfterListenerFailed,
684    TelemetryEventKind::RetryReserved,
685    TelemetryEventKind::RetryBackoffStarted,
686    TelemetryEventKind::RetryBackoffCancelled,
687    TelemetryEventKind::RetryExhausted,
688    TelemetryEventKind::ItemSkipped,
689    TelemetryEventKind::FaultRollbackCommitted,
690    TelemetryEventKind::FaultNoRollbackCommitted,
691    TelemetryEventKind::CheckpointLoaded,
692    TelemetryEventKind::CheckpointCommitted,
693    TelemetryEventKind::RepositoryConflict,
694    TelemetryEventKind::RepositoryTransientFailure,
695    TelemetryEventKind::FlowStepResultCommitted,
696    TelemetryEventKind::FlowDecisionCommitted,
697    TelemetryEventKind::FlowCompletedStepReused,
698    TelemetryEventKind::StepStartLimitExceeded,
699    TelemetryEventKind::OperatorRequestAccepted,
700    TelemetryEventKind::OperatorRequestRejected,
701    TelemetryEventKind::OperatorRequestCompleted,
702    TelemetryEventKind::ExplorerPageServed,
703    TelemetryEventKind::ShutdownRequested,
704    TelemetryEventKind::ShutdownIntakeStopped,
705    TelemetryEventKind::ShutdownDrainCompleted,
706    TelemetryEventKind::ShutdownDeadlineExceeded,
707    TelemetryEventKind::StaleDetected,
708    TelemetryEventKind::RecoveryProposed,
709    TelemetryEventKind::RecoveryApplied,
710    TelemetryEventKind::RecoveryRejected,
711    TelemetryEventKind::RetentionPlanned,
712    TelemetryEventKind::RetentionApplied,
713    TelemetryEventKind::RetentionRejected,
714    TelemetryEventKind::SplitBranchStarted,
715    TelemetryEventKind::SplitBranchCompleted,
716    TelemetryEventKind::PartitionPlanCommitted,
717    TelemetryEventKind::PartitionAssigned,
718    TelemetryEventKind::PartitionCompleted,
719    TelemetryEventKind::PartitionAggregated,
720    TelemetryEventKind::TelemetryExportDropped,
721    TelemetryEventKind::MigrationStarted,
722    TelemetryEventKind::MigrationCompleted,
723    TelemetryEventKind::MigrationFailed,
724];
725
726/// One versioned event containing only reviewed safe fields.
727#[derive(Clone, Debug, Eq, PartialEq)]
728pub struct TelemetryRecord {
729    kind: TelemetryEventKind,
730    fields: Vec<DiagnosticField>,
731    job_execution_id: Option<JobExecutionId>,
732}
733
734impl TelemetryRecord {
735    /// Constructs a catalog event without application-supplied fields.
736    #[must_use]
737    pub const fn catalog(kind: TelemetryEventKind) -> Self {
738        Self {
739            kind,
740            fields: Vec::new(),
741            job_execution_id: None,
742        }
743    }
744
745    pub(crate) fn operator(
746        kind: TelemetryEventKind,
747        request: &OperatorRequest,
748        outcome: Option<OperatorOutcomeClass>,
749        rejection: Option<OperatorRejection>,
750    ) -> Self {
751        let mut record = Self::catalog(kind);
752        record.fields = vec![
753            DiagnosticField::new("operator.action", request.action().as_str()),
754            DiagnosticField::new(
755                "authorization.class",
756                request.authorization_class().as_str(),
757            ),
758            DiagnosticField::new("operation.id", request.operation_id().as_str()),
759            DiagnosticField::new("actor.ref", request.actor().as_str()),
760        ];
761        if let Some(reason) = request.reason() {
762            record
763                .fields
764                .push(DiagnosticField::new("reason.code", reason.as_str()));
765        }
766        if let Some(outcome) = outcome {
767            record
768                .fields
769                .push(DiagnosticField::new("outcome.class", outcome.as_str()));
770        }
771        if let Some(rejection) = rejection {
772            record
773                .fields
774                .push(DiagnosticField::new("rejection.class", rejection.as_str()));
775        }
776        record.job_execution_id = request.job_execution_id();
777        record
778    }
779
780    pub(crate) fn recovery(kind: TelemetryEventKind, proposal: &RecoveryProposal) -> Self {
781        let execution_id = proposal.evidence().execution_id();
782        let mut record = Self::catalog(kind);
783        record.job_execution_id = Some(execution_id);
784        record.fields = vec![
785            DiagnosticField::new("job.execution.id", execution_id.to_string()),
786            DiagnosticField::new("evidence.digest_present", "true"),
787            DiagnosticField::new(
788                "inactivity.class",
789                inactivity_class(proposal.evidence().inactivity()),
790            ),
791        ];
792        record
793    }
794
795    pub(crate) fn shutdown(kind: TelemetryEventKind, drain: &'static str, unjoined: usize) -> Self {
796        let mut record = Self::catalog(kind);
797        record.fields = vec![
798            DiagnosticField::new("drain.result", drain),
799            DiagnosticField::new("unjoined.tasks", unjoined.to_string()),
800        ];
801        record
802    }
803
804    pub(crate) const fn explorer(job_execution_id: Option<JobExecutionId>) -> Self {
805        Self {
806            kind: TelemetryEventKind::ExplorerPageServed,
807            fields: Vec::new(),
808            job_execution_id,
809        }
810    }
811
812    pub(crate) fn retention(
813        kind: TelemetryEventKind,
814        action: Option<RetentionAction>,
815        outcome: Option<RetentionOutcome>,
816        counts: PurgeCounts,
817    ) -> Self {
818        let mut fields = Vec::new();
819        if let Some(action) = action {
820            fields.push(DiagnosticField::new("retention.action", action.as_str()));
821        }
822        if let Some(outcome) = outcome {
823            fields.push(DiagnosticField::new("outcome.class", outcome.as_str()));
824        }
825        fields.extend([
826            DiagnosticField::new(
827                "deleted.flow_decisions",
828                counts.flow_decisions().to_string(),
829            ),
830            DiagnosticField::new(
831                "deleted.recovery_decisions",
832                counts.recovery_decisions().to_string(),
833            ),
834            DiagnosticField::new(
835                "deleted.operator_requests",
836                counts.operator_requests().to_string(),
837            ),
838            DiagnosticField::new(
839                "deleted.step_partitions",
840                counts.step_partitions().to_string(),
841            ),
842            DiagnosticField::new(
843                "deleted.step_executions",
844                counts.step_executions().to_string(),
845            ),
846            DiagnosticField::new(
847                "deleted.job_executions",
848                counts.job_executions().to_string(),
849            ),
850            DiagnosticField::new("deleted.job_instances", counts.job_instances().to_string()),
851        ]);
852        Self {
853            kind,
854            fields,
855            job_execution_id: None,
856        }
857    }
858
859    /// Returns the schema version carried by this event.
860    #[must_use]
861    pub const fn schema_version(&self) -> u16 {
862        TELEMETRY_SCHEMA_VERSION
863    }
864
865    /// Returns the stable catalog kind.
866    #[must_use]
867    pub const fn kind(&self) -> TelemetryEventKind {
868        self.kind
869    }
870
871    /// Returns the named execution retained by an incident buffer, when any.
872    #[must_use]
873    pub const fn job_execution_id(&self) -> Option<JobExecutionId> {
874        self.job_execution_id
875    }
876
877    /// Borrows reviewed structured fields.
878    #[must_use]
879    pub fn fields(&self) -> &[DiagnosticField] {
880        &self.fields
881    }
882}
883
884fn inactivity_class(duration: Duration) -> &'static str {
885    match duration.as_secs() {
886        0..=59 => "lt_1m",
887        60..=899 => "1m_to_15m",
888        900..=3_599 => "15m_to_1h",
889        3_600..=86_399 => "1h_to_24h",
890        _ => "gte_24h",
891    }
892}
893
894/// Receives versioned observational events.
895pub trait TelemetryEventSink: Send + Sync {
896    /// Emits one reviewed record.
897    fn emit(&self, event: &TelemetryRecord);
898}
899
900/// Invalid finite incident-buffer configuration.
901#[derive(Clone, Copy, Debug, Eq, PartialEq)]
902pub struct IncidentBufferConfigurationError;
903
904impl fmt::Display for IncidentBufferConfigurationError {
905    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
906        formatter.write_str("incident event bounds must be nonzero and per-execution at most 200")
907    }
908}
909
910impl std::error::Error for IncidentBufferConfigurationError {}
911
912#[derive(Debug)]
913struct IncidentState {
914    records: VecDeque<TelemetryRecord>,
915}
916
917/// A process-local, finite incident event buffer.
918///
919/// The buffer is diagnostic only and loses its contents on crash. Durable
920/// metadata remains authoritative.
921#[derive(Debug)]
922pub struct IncidentEventBuffer {
923    per_execution: usize,
924    total: usize,
925    state: Mutex<IncidentState>,
926}
927
928impl IncidentEventBuffer {
929    /// Validates and constructs one finite buffer.
930    ///
931    /// # Errors
932    ///
933    /// Rejects zero bounds, a per-execution bound above 200, or a total bound
934    /// smaller than the per-execution bound.
935    pub fn new(
936        per_execution: usize,
937        total: usize,
938    ) -> Result<Self, IncidentBufferConfigurationError> {
939        if per_execution == 0
940            || per_execution > MAX_RETAINED_EVENTS_PER_EXECUTION
941            || total < per_execution
942        {
943            return Err(IncidentBufferConfigurationError);
944        }
945        Ok(Self {
946            per_execution,
947            total,
948            state: Mutex::new(IncidentState {
949                records: VecDeque::with_capacity(total),
950            }),
951        })
952    }
953
954    /// Returns at most the configured newest records for one execution.
955    #[must_use]
956    pub fn events_for(&self, execution_id: JobExecutionId) -> Vec<TelemetryRecord> {
957        let state = self
958            .state
959            .lock()
960            .unwrap_or_else(std::sync::PoisonError::into_inner);
961        let mut selected = state
962            .records
963            .iter()
964            .rev()
965            .filter(|record| record.job_execution_id() == Some(execution_id))
966            .take(self.per_execution)
967            .cloned()
968            .collect::<Vec<_>>();
969        selected.reverse();
970        selected
971    }
972}
973
974impl Default for IncidentEventBuffer {
975    fn default() -> Self {
976        Self {
977            per_execution: DEFAULT_RETAINED_EVENTS_PER_EXECUTION,
978            total: DEFAULT_RETAINED_EVENT_CAPACITY,
979            state: Mutex::new(IncidentState {
980                records: VecDeque::with_capacity(DEFAULT_RETAINED_EVENT_CAPACITY),
981            }),
982        }
983    }
984}
985
986impl TelemetryEventSink for IncidentEventBuffer {
987    fn emit(&self, event: &TelemetryRecord) {
988        let mut state = self
989            .state
990            .lock()
991            .unwrap_or_else(std::sync::PoisonError::into_inner);
992        if state.records.len() == self.total {
993            state.records.pop_front();
994        }
995        state.records.push_back(event.clone());
996    }
997}
998
999/// Emits an event while isolating a sink panic from execution correctness.
1000pub(crate) fn emit_safely(sink: Option<&Arc<dyn TelemetryEventSink>>, event: &TelemetryRecord) {
1001    if let Some(sink) = sink {
1002        let _ = catch_unwind(AssertUnwindSafe(|| sink.emit(event)));
1003    }
1004}
1005
1006/// A complete set of typed dimensions accepted by the metric catalog.
1007#[derive(Clone, Debug, Default, Eq, PartialEq)]
1008pub struct MetricDimensions {
1009    event: Option<TelemetryEventKind>,
1010    status: Option<BatchStatus>,
1011    action: Option<OperatorAction>,
1012    authorization: Option<AuthorizationClass>,
1013    outcome: Option<OperatorOutcomeClass>,
1014    job_name: Option<JobName>,
1015    step_name: Option<StepName>,
1016}
1017
1018impl MetricDimensions {
1019    /// Adds one bounded event-name dimension.
1020    #[must_use]
1021    pub const fn with_event(mut self, value: TelemetryEventKind) -> Self {
1022        self.event = Some(value);
1023        self
1024    }
1025
1026    /// Adds one lifecycle-status dimension.
1027    #[must_use]
1028    pub const fn with_status(mut self, value: BatchStatus) -> Self {
1029        self.status = Some(value);
1030        self
1031    }
1032
1033    /// Adds one operator-action dimension.
1034    #[must_use]
1035    pub const fn with_action(mut self, value: OperatorAction) -> Self {
1036        self.action = Some(value);
1037        self
1038    }
1039
1040    /// Adds one authorization-class dimension.
1041    #[must_use]
1042    pub const fn with_authorization(mut self, value: AuthorizationClass) -> Self {
1043        self.authorization = Some(value);
1044        self
1045    }
1046
1047    /// Adds one operator-outcome dimension.
1048    #[must_use]
1049    pub const fn with_outcome(mut self, value: OperatorOutcomeClass) -> Self {
1050        self.outcome = Some(value);
1051        self
1052    }
1053
1054    /// Adds a validated job name, subject to the configured allowlist.
1055    #[must_use]
1056    pub fn with_job_name(mut self, value: JobName) -> Self {
1057        self.job_name = Some(value);
1058        self
1059    }
1060
1061    /// Adds a validated step name, subject to the configured allowlist.
1062    #[must_use]
1063    pub fn with_step_name(mut self, value: StepName) -> Self {
1064        self.step_name = Some(value);
1065        self
1066    }
1067}
1068
1069/// A stable metric family and its complete allowed label set.
1070#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
1071#[non_exhaustive]
1072pub enum MetricFamily {
1073    /// Currently active executions.
1074    ActiveExecutions,
1075    /// Completed execution outcomes.
1076    CompletedExecutions,
1077    /// Execution duration distribution.
1078    ExecutionDuration,
1079    /// Read, process, write, filter, skip, retry, commit, and rollback counts.
1080    ItemCount,
1081    /// Repository operation duration distribution.
1082    RepositoryOperationDuration,
1083    /// Repository optimistic conflicts.
1084    RepositoryConflicts,
1085    /// Repository failures.
1086    RepositoryErrors,
1087    /// Bounded queue depth.
1088    QueueDepth,
1089    /// Configured concurrency budget.
1090    ConfiguredConcurrency,
1091    /// Currently active concurrency.
1092    ActiveConcurrency,
1093    /// Guarded operator request outcomes.
1094    OperatorRequests,
1095    /// Versioned execution event counter.
1096    ExecutionEvents,
1097    /// Recovery outcomes.
1098    RecoveryOutcomes,
1099    /// Shutdown outcomes.
1100    ShutdownOutcomes,
1101    /// Bounded exporter drops.
1102    ExportDropped,
1103}
1104
1105impl MetricFamily {
1106    /// Returns the stable family name.
1107    #[must_use]
1108    pub const fn as_str(self) -> &'static str {
1109        match self {
1110            Self::ActiveExecutions => "oxide_batch_active_executions",
1111            Self::CompletedExecutions => "oxide_batch_completed_executions_total",
1112            Self::ExecutionDuration => "oxide_batch_execution_duration_seconds",
1113            Self::ItemCount => "oxide_batch_item_operations_total",
1114            Self::RepositoryOperationDuration => {
1115                "oxide_batch_repository_operation_duration_seconds"
1116            }
1117            Self::RepositoryConflicts => "oxide_batch_repository_conflicts_total",
1118            Self::RepositoryErrors => "oxide_batch_repository_errors_total",
1119            Self::QueueDepth => "oxide_batch_queue_depth_records",
1120            Self::ConfiguredConcurrency => "oxide_batch_concurrency_configured_workers",
1121            Self::ActiveConcurrency => "oxide_batch_concurrency_active_workers",
1122            Self::OperatorRequests => "oxide_batch_operator_requests_total",
1123            Self::ExecutionEvents => "oxide_batch_execution_events_total",
1124            Self::RecoveryOutcomes => "oxide_batch_recovery_outcomes_total",
1125            Self::ShutdownOutcomes => "oxide_batch_shutdown_outcomes_total",
1126            Self::ExportDropped => "oxide_batch_telemetry_export_dropped_total",
1127        }
1128    }
1129
1130    /// Returns the versioned unit.
1131    #[must_use]
1132    pub const fn unit(self) -> MetricUnit {
1133        match self {
1134            Self::ExecutionDuration | Self::RepositoryOperationDuration => MetricUnit::Seconds,
1135            Self::ItemCount => MetricUnit::Items,
1136            Self::QueueDepth => MetricUnit::Records,
1137            Self::ConfiguredConcurrency | Self::ActiveConcurrency => MetricUnit::Workers,
1138            Self::RepositoryConflicts
1139            | Self::RepositoryErrors
1140            | Self::OperatorRequests
1141            | Self::ExecutionEvents
1142            | Self::RecoveryOutcomes
1143            | Self::ShutdownOutcomes
1144            | Self::ExportDropped => MetricUnit::Events,
1145            Self::ActiveExecutions | Self::CompletedExecutions => MetricUnit::Executions,
1146        }
1147    }
1148
1149    /// Returns the complete permitted label-key set for this family.
1150    #[must_use]
1151    pub const fn label_keys(self) -> &'static [&'static str] {
1152        match self {
1153            Self::ActiveExecutions | Self::CompletedExecutions | Self::ExecutionDuration => {
1154                &["status", "job", "step"]
1155            }
1156            Self::ItemCount
1157            | Self::RepositoryOperationDuration
1158            | Self::RepositoryConflicts
1159            | Self::RepositoryErrors
1160            | Self::QueueDepth
1161            | Self::ConfiguredConcurrency
1162            | Self::ActiveConcurrency => &["event"],
1163            Self::OperatorRequests => &["action", "authorization", "outcome"],
1164            Self::ExecutionEvents => &["event", "status", "job", "step"],
1165            Self::RecoveryOutcomes => &["outcome", "action"],
1166            Self::ShutdownOutcomes => &["status"],
1167            Self::ExportDropped => &["reason"],
1168        }
1169    }
1170}
1171
1172/// Stable measurement unit of one metric family.
1173#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
1174#[non_exhaustive]
1175pub enum MetricUnit {
1176    /// Execution count.
1177    Executions,
1178    /// Event or operation count.
1179    Events,
1180    /// Duration in seconds.
1181    Seconds,
1182    /// Item count.
1183    Items,
1184    /// Queue record count.
1185    Records,
1186    /// Worker count.
1187    Workers,
1188}
1189
1190/// A metric observation after allowlist and cardinality enforcement.
1191#[derive(Clone, Debug, Eq, PartialEq)]
1192pub struct MetricObservation {
1193    family: MetricFamily,
1194    labels: Vec<MetricLabel>,
1195    overflowed: bool,
1196}
1197
1198impl MetricObservation {
1199    /// Returns the stable family.
1200    #[must_use]
1201    pub const fn family(&self) -> MetricFamily {
1202        self.family
1203    }
1204
1205    /// Borrows the complete bounded label set.
1206    #[must_use]
1207    pub fn labels(&self) -> &[MetricLabel] {
1208        &self.labels
1209    }
1210
1211    /// Returns whether a new combination was mapped to the reserved series.
1212    #[must_use]
1213    pub const fn overflowed(&self) -> bool {
1214        self.overflowed
1215    }
1216}
1217
1218/// Invalid metric name-labelling configuration.
1219#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1220pub struct MetricConfigurationError;
1221
1222impl fmt::Display for MetricConfigurationError {
1223    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1224        formatter.write_str("metric name allowlists may contain at most 50 names")
1225    }
1226}
1227
1228impl std::error::Error for MetricConfigurationError {}
1229
1230/// Enforces the per-family label-cardinality and name-allowlist budgets.
1231#[derive(Debug, Default)]
1232pub struct MetricCardinalityGuard {
1233    job_names: BTreeSet<JobName>,
1234    step_names: BTreeSet<StepName>,
1235    observed: BTreeMap<MetricFamily, BTreeSet<String>>,
1236    dropped: BTreeMap<MetricFamily, u64>,
1237}
1238
1239impl MetricCardinalityGuard {
1240    /// Configures explicit job and step name allowlists.
1241    ///
1242    /// # Errors
1243    ///
1244    /// Rejects either allowlist when it contains more than 50 names.
1245    pub fn new(
1246        job_names: impl IntoIterator<Item = JobName>,
1247        step_names: impl IntoIterator<Item = StepName>,
1248    ) -> Result<Self, MetricConfigurationError> {
1249        let job_names: BTreeSet<_> = job_names.into_iter().collect();
1250        let step_names: BTreeSet<_> = step_names.into_iter().collect();
1251        if job_names.len() > MAX_METRIC_NAME_ALLOWLIST
1252            || step_names.len() > MAX_METRIC_NAME_ALLOWLIST
1253        {
1254            return Err(MetricConfigurationError);
1255        }
1256        Ok(Self {
1257            job_names,
1258            step_names,
1259            observed: BTreeMap::new(),
1260            dropped: BTreeMap::new(),
1261        })
1262    }
1263
1264    /// Applies the declared family label set and finite series budget.
1265    #[must_use]
1266    pub fn observe(
1267        &mut self,
1268        family: MetricFamily,
1269        dimensions: &MetricDimensions,
1270    ) -> MetricObservation {
1271        let mut labels = family_labels(family, dimensions, &self.job_names, &self.step_names);
1272        let key = label_key(&labels);
1273        let observed = self.observed.entry(family).or_default();
1274        let overflowed = !observed.contains(&key)
1275            && observed.len() >= METRIC_CARDINALITY_BUDGET.saturating_sub(1);
1276        if overflowed {
1277            for label in &mut labels {
1278                label.replace_value(OTHER_LABEL_VALUE);
1279            }
1280            *self.dropped.entry(family).or_default() += 1;
1281            observed.insert(label_key(&labels));
1282        } else {
1283            observed.insert(key);
1284        }
1285        MetricObservation {
1286            family,
1287            labels,
1288            overflowed,
1289        }
1290    }
1291
1292    /// Returns the count of combinations mapped to the reserved series.
1293    #[must_use]
1294    pub fn dropped_cardinality(&self, family: MetricFamily) -> u64 {
1295        self.dropped.get(&family).copied().unwrap_or(0)
1296    }
1297
1298    /// Returns the currently retained series count for one family.
1299    #[must_use]
1300    pub fn series_count(&self, family: MetricFamily) -> usize {
1301        self.observed.get(&family).map_or(0, BTreeSet::len)
1302    }
1303}
1304
1305fn family_labels(
1306    family: MetricFamily,
1307    dimensions: &MetricDimensions,
1308    job_names: &BTreeSet<JobName>,
1309    step_names: &BTreeSet<StepName>,
1310) -> Vec<MetricLabel> {
1311    let mut labels = Vec::new();
1312    match family {
1313        MetricFamily::ActiveExecutions
1314        | MetricFamily::CompletedExecutions
1315        | MetricFamily::ExecutionDuration => {
1316            labels.push(MetricLabel::new(
1317                "status",
1318                dimensions
1319                    .status
1320                    .map_or_else(|| "none".to_owned(), |status| status.to_string()),
1321            ));
1322            labels.push(MetricLabel::new(
1323                "job",
1324                allowed_job_name(dimensions.job_name.as_ref(), job_names),
1325            ));
1326            labels.push(MetricLabel::new(
1327                "step",
1328                allowed_step_name(dimensions.step_name.as_ref(), step_names),
1329            ));
1330        }
1331        MetricFamily::ItemCount
1332        | MetricFamily::RepositoryOperationDuration
1333        | MetricFamily::RepositoryConflicts
1334        | MetricFamily::RepositoryErrors
1335        | MetricFamily::QueueDepth
1336        | MetricFamily::ConfiguredConcurrency
1337        | MetricFamily::ActiveConcurrency => labels.push(MetricLabel::new(
1338            "event",
1339            dimensions
1340                .event
1341                .map_or("unknown", TelemetryEventKind::as_str),
1342        )),
1343        MetricFamily::OperatorRequests => {
1344            labels.push(MetricLabel::new(
1345                "action",
1346                dimensions.action.map_or("unknown", OperatorAction::as_str),
1347            ));
1348            labels.push(MetricLabel::new(
1349                "authorization",
1350                dimensions
1351                    .authorization
1352                    .map_or("unknown", AuthorizationClass::as_str),
1353            ));
1354            labels.push(MetricLabel::new(
1355                "outcome",
1356                dimensions
1357                    .outcome
1358                    .map_or("unknown", OperatorOutcomeClass::as_str),
1359            ));
1360        }
1361        MetricFamily::ExecutionEvents => {
1362            labels.push(MetricLabel::new(
1363                "event",
1364                dimensions
1365                    .event
1366                    .map_or("unknown", TelemetryEventKind::as_str),
1367            ));
1368            labels.push(MetricLabel::new(
1369                "status",
1370                dimensions
1371                    .status
1372                    .map_or_else(|| "none".to_owned(), |status| status.to_string()),
1373            ));
1374            labels.push(MetricLabel::new(
1375                "job",
1376                allowed_job_name(dimensions.job_name.as_ref(), job_names),
1377            ));
1378            labels.push(MetricLabel::new(
1379                "step",
1380                allowed_step_name(dimensions.step_name.as_ref(), step_names),
1381            ));
1382        }
1383        MetricFamily::RecoveryOutcomes => {
1384            labels.push(MetricLabel::new(
1385                "outcome",
1386                dimensions
1387                    .outcome
1388                    .map_or("unknown", OperatorOutcomeClass::as_str),
1389            ));
1390            labels.push(MetricLabel::new(
1391                "action",
1392                dimensions.action.map_or("RECOVER", OperatorAction::as_str),
1393            ));
1394        }
1395        MetricFamily::ShutdownOutcomes => labels.push(MetricLabel::new(
1396            "status",
1397            dimensions
1398                .status
1399                .map_or_else(|| "none".to_owned(), |status| status.to_string()),
1400        )),
1401        MetricFamily::ExportDropped => labels.push(MetricLabel::new("reason", "queue_full")),
1402    }
1403    labels
1404}
1405
1406fn allowed_job_name<'a>(name: Option<&'a JobName>, allowlist: &'a BTreeSet<JobName>) -> &'a str {
1407    match name {
1408        Some(name) if allowlist.contains(name) => name.as_str(),
1409        _ => OTHER_LABEL_VALUE,
1410    }
1411}
1412
1413fn allowed_step_name<'a>(name: Option<&'a StepName>, allowlist: &'a BTreeSet<StepName>) -> &'a str {
1414    match name {
1415        Some(name) if allowlist.contains(name) => name.as_str(),
1416        _ => OTHER_LABEL_VALUE,
1417    }
1418}
1419
1420fn label_key(labels: &[MetricLabel]) -> String {
1421    labels
1422        .iter()
1423        .map(ToString::to_string)
1424        .collect::<Vec<_>>()
1425        .join("\u{1f}")
1426}
1427
1428/// A validated finite exporter queue bound.
1429#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1430pub struct ExportQueueBound(usize);
1431
1432impl ExportQueueBound {
1433    /// Validates `64..=65536` records.
1434    ///
1435    /// # Errors
1436    ///
1437    /// Returns [`ExporterConfigurationError`] outside the accepted range.
1438    pub const fn new(value: usize) -> Result<Self, ExporterConfigurationError> {
1439        if value < MIN_EXPORT_QUEUE_RECORDS || value > MAX_EXPORT_QUEUE_RECORDS {
1440            return Err(ExporterConfigurationError::QueueBound);
1441        }
1442        Ok(Self(value))
1443    }
1444
1445    /// Returns the accepted record count.
1446    #[must_use]
1447    pub const fn get(self) -> usize {
1448        self.0
1449    }
1450}
1451
1452impl Default for ExportQueueBound {
1453    fn default() -> Self {
1454        Self(DEFAULT_EXPORT_QUEUE_RECORDS)
1455    }
1456}
1457
1458/// A validated throttling window for exporter-drop reporting.
1459#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1460pub struct DropReportWindow(Duration);
1461
1462impl DropReportWindow {
1463    /// Validates `1 s..=1 h`.
1464    ///
1465    /// # Errors
1466    ///
1467    /// Returns [`ExporterConfigurationError`] outside the accepted range.
1468    pub const fn new(value: Duration) -> Result<Self, ExporterConfigurationError> {
1469        if value.as_millis() < MIN_DROP_REPORT_WINDOW.as_millis()
1470            || value.as_millis() > MAX_DROP_REPORT_WINDOW.as_millis()
1471        {
1472            return Err(ExporterConfigurationError::DropReportWindow);
1473        }
1474        Ok(Self(value))
1475    }
1476
1477    /// Returns the accepted window.
1478    #[must_use]
1479    pub const fn get(self) -> Duration {
1480        self.0
1481    }
1482}
1483
1484impl Default for DropReportWindow {
1485    fn default() -> Self {
1486        Self(DEFAULT_DROP_REPORT_WINDOW)
1487    }
1488}
1489
1490/// Invalid bounded exporter configuration.
1491#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1492#[non_exhaustive]
1493pub enum ExporterConfigurationError {
1494    /// Queue count was outside `64..=65536`.
1495    QueueBound,
1496    /// Drop report window was outside `1 s..=1 h`.
1497    DropReportWindow,
1498}
1499
1500impl fmt::Display for ExporterConfigurationError {
1501    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1502        match self {
1503            Self::QueueBound => formatter.write_str("export queue must hold 64 to 65536 records"),
1504            Self::DropReportWindow => {
1505                formatter.write_str("drop report window must be between 1 second and 1 hour")
1506            }
1507        }
1508    }
1509}
1510
1511impl std::error::Error for ExporterConfigurationError {}
1512
1513#[derive(Debug)]
1514struct QueueState {
1515    records: VecDeque<TelemetryRecord>,
1516    dropped: u64,
1517    last_drop_report: Option<Duration>,
1518}
1519
1520/// Result of one non-blocking enqueue attempt.
1521#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1522#[non_exhaustive]
1523pub enum EnqueueResult {
1524    /// The record entered the queue.
1525    Accepted,
1526    /// The newest record was dropped because the queue was full.
1527    Dropped {
1528        /// Whether a throttled `telemetry.export_dropped` observation is due.
1529        report_due: bool,
1530    },
1531}
1532
1533/// Cloneable producer for one bounded exporter queue.
1534#[derive(Clone, Debug)]
1535pub struct TelemetryQueue {
1536    bound: ExportQueueBound,
1537    report_window: DropReportWindow,
1538    state: Arc<Mutex<QueueState>>,
1539}
1540
1541impl TelemetryQueue {
1542    /// Constructs an empty bounded queue.
1543    #[must_use]
1544    pub fn new(bound: ExportQueueBound, report_window: DropReportWindow) -> Self {
1545        Self {
1546            bound,
1547            report_window,
1548            state: Arc::new(Mutex::new(QueueState {
1549                records: VecDeque::with_capacity(bound.get()),
1550                dropped: 0,
1551                last_drop_report: None,
1552            })),
1553        }
1554    }
1555
1556    /// Enqueues without waiting; a full queue drops this newest record.
1557    #[must_use]
1558    pub fn enqueue(&self, record: TelemetryRecord, now: Duration) -> EnqueueResult {
1559        let mut state = self
1560            .state
1561            .lock()
1562            .unwrap_or_else(std::sync::PoisonError::into_inner);
1563        if state.records.len() < self.bound.get() {
1564            state.records.push_back(record);
1565            return EnqueueResult::Accepted;
1566        }
1567        state.dropped = state.dropped.saturating_add(1);
1568        let report_due = state.last_drop_report.is_none_or(|last| {
1569            now.checked_sub(last)
1570                .is_some_and(|elapsed| elapsed >= self.report_window.get())
1571        });
1572        if report_due {
1573            state.last_drop_report = Some(now);
1574        }
1575        EnqueueResult::Dropped { report_due }
1576    }
1577
1578    /// Returns the current queued count.
1579    #[must_use]
1580    pub fn len(&self) -> usize {
1581        self.state
1582            .lock()
1583            .unwrap_or_else(std::sync::PoisonError::into_inner)
1584            .records
1585            .len()
1586    }
1587
1588    /// Returns whether the queue is empty.
1589    #[must_use]
1590    pub fn is_empty(&self) -> bool {
1591        self.len() == 0
1592    }
1593
1594    /// Returns the cumulative dropped-newest count.
1595    #[must_use]
1596    pub fn dropped(&self) -> u64 {
1597        self.state
1598            .lock()
1599            .unwrap_or_else(std::sync::PoisonError::into_inner)
1600            .dropped
1601    }
1602
1603    fn pop(&self) -> Option<TelemetryRecord> {
1604        self.state
1605            .lock()
1606            .unwrap_or_else(std::sync::PoisonError::into_inner)
1607            .records
1608            .pop_front()
1609    }
1610}
1611
1612/// Adapter-owned asynchronous export boundary.
1613pub trait TelemetryExportSink: Send + Sync {
1614    /// Exports one reviewed record without exposing SDK types to the facade.
1615    fn export<'a>(&'a self, record: &'a TelemetryRecord) -> BoxFuture<'a, Result<(), ExportError>>;
1616}
1617
1618/// A value-redacted exporter failure.
1619#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1620pub struct ExportError;
1621
1622impl fmt::Display for ExportError {
1623    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1624        formatter.write_str("telemetry export failed")
1625    }
1626}
1627
1628impl std::error::Error for ExportError {}
1629
1630/// Flush result that never changes batch correctness.
1631#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1632pub struct ExportFlushReport {
1633    exported: u64,
1634    failed: u64,
1635    dropped: u64,
1636}
1637
1638impl ExportFlushReport {
1639    /// Returns successfully exported records.
1640    #[must_use]
1641    pub const fn exported(self) -> u64 {
1642        self.exported
1643    }
1644
1645    /// Returns records isolated after an exporter error or panic.
1646    #[must_use]
1647    pub const fn failed(self) -> u64 {
1648        self.failed
1649    }
1650
1651    /// Returns records dropped before export.
1652    #[must_use]
1653    pub const fn dropped(self) -> u64 {
1654        self.dropped
1655    }
1656}
1657
1658/// Drains one queue from an application-owned task.
1659pub struct TelemetryExporter<S> {
1660    queue: TelemetryQueue,
1661    sink: S,
1662}
1663
1664impl<S: TelemetryExportSink> TelemetryExporter<S> {
1665    /// Binds a queue to an adapter without spawning any task.
1666    #[must_use]
1667    pub const fn new(queue: TelemetryQueue, sink: S) -> Self {
1668        Self { queue, sink }
1669    }
1670
1671    /// Drains every currently queued record and isolates adapter failures.
1672    pub async fn flush(&self) -> ExportFlushReport {
1673        let mut exported = 0_u64;
1674        let mut failed = 0_u64;
1675        while let Some(record) = self.queue.pop() {
1676            let result = AssertUnwindSafe(self.sink.export(&record))
1677                .catch_unwind()
1678                .await;
1679            match result {
1680                Ok(Ok(())) => exported = exported.saturating_add(1),
1681                Ok(Err(_)) | Err(_) => failed = failed.saturating_add(1),
1682            }
1683        }
1684        ExportFlushReport {
1685            exported,
1686            failed,
1687            dropped: self.queue.dropped(),
1688        }
1689    }
1690}
1691
1692// These imports are deliberately exercised in public field constructors. Keep
1693// them visible to rustdoc as the closed safe-field vocabulary grows.
1694const _: fn(&ActorRef, &OperationId, &ReasonCode, PurgeCounts) = |_, _, _, _| {};
oxide_batch/telemetry.rs

oxide_batch/
telemetry.rs
