feldera_types/
adapter_stats.rs

1use bytemuck::NoUninit;
2use chrono::{DateTime, SecondsFormat, Utc};
3use serde::{Deserialize, Serialize};
4use serde_json::Value as JsonValue;
5use std::collections::BTreeMap;
6use utoipa::ToSchema;
7use uuid::Uuid;
8
9use crate::{
10    checkpoint::CheckpointActivity,
11    coordination::Step,
12    memory_pressure::MemoryPressure,
13    suspend::{PermanentSuspendError, SuspendError},
14    transaction::{CommitProgressSummary, ConcurrentBootstrapPhase, TransactionId},
15};
16
17/// Pipeline state.
18#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
19#[serde(rename_all = "PascalCase")]
20pub enum PipelineState {
21    /// All input endpoints are paused (or are in the process of being paused).
22    #[default]
23    Paused,
24    /// Controller is running.
25    Running,
26    /// Controller is being terminated.
27    Terminated,
28}
29
30// Metrics for an input endpoint.
31///
32/// Serializes to match the subset of fields needed for error tracking.
33#[derive(Default, Serialize, Deserialize, Debug, Clone)]
34pub struct InputEndpointErrorMetrics {
35    pub endpoint_name: String,
36    #[serde(default)]
37    pub num_transport_errors: u64,
38    #[serde(default)]
39    pub num_parse_errors: u64,
40}
41
42/// Metrics for an output endpoint.
43///
44/// Serializes to match the subset of fields needed for error tracking.
45#[derive(Default, Serialize, Deserialize, Debug, Clone)]
46pub struct OutputEndpointErrorMetrics {
47    pub endpoint_name: String,
48    #[serde(default)]
49    pub num_encode_errors: u64,
50    #[serde(default)]
51    pub num_transport_errors: u64,
52}
53
54/// Endpoint statistics containing metrics.
55///
56/// Wraps metrics in a structure that matches the full stats API shape.
57#[derive(Serialize, Deserialize, Debug, Clone)]
58pub struct EndpointErrorStats<T> {
59    #[serde(default)]
60    pub metrics: T,
61}
62
63/// Schema definition for endpoint config that only includes the stream field.
64#[derive(Debug, Deserialize, Serialize, ToSchema)]
65pub struct ShortEndpointConfig {
66    /// The name of the stream.
67    pub stream: String,
68}
69
70/// Pipeline error statistics response from the runtime.
71///
72/// Lightweight response containing only error counts from all endpoints.
73#[derive(Serialize, Deserialize, Debug, Clone)]
74pub struct PipelineStatsErrorsResponse {
75    #[serde(default)]
76    pub inputs: Vec<EndpointErrorStats<InputEndpointErrorMetrics>>,
77    #[serde(default)]
78    pub outputs: Vec<EndpointErrorStats<OutputEndpointErrorMetrics>>,
79}
80
81// OpenAPI schema definitions for controller statistics
82// These match the serialized JSON structure from the adapters crate
83
84/// Transaction status summarized as a single value.
85#[derive(
86    Debug, Default, Copy, PartialEq, Eq, Clone, NoUninit, Serialize, Deserialize, ToSchema,
87)]
88#[repr(u8)]
89pub enum TransactionStatus {
90    #[default]
91    NoTransaction,
92    TransactionInProgress,
93    CommitInProgress,
94}
95
96/// Transaction phase.
97#[derive(Clone, Copy, PartialEq, Eq, Debug, Deserialize, Serialize, ToSchema)]
98#[serde(rename_all = "PascalCase")]
99#[schema(as = TransactionPhase)]
100pub enum ExternalTransactionPhase {
101    /// Transaction is in progress.
102    Started,
103    /// Transaction has been committed.
104    Committed,
105}
106
107/// Connector transaction phase with debugging label.
108#[derive(Clone, PartialEq, Eq, Debug, Deserialize, Serialize, ToSchema)]
109#[schema(as = ConnectorTransactionPhase)]
110pub struct ExternalConnectorTransactionPhase {
111    /// Current phase of the transaction.
112    #[schema(value_type = TransactionPhase)]
113    pub phase: ExternalTransactionPhase,
114    /// Optional label for debugging.
115    pub label: Option<String>,
116}
117
118/// Information about entities that initiated the current transaction.
119#[derive(Clone, Default, Debug, Deserialize, Serialize, ToSchema)]
120#[schema(as = TransactionInitiators)]
121pub struct ExternalTransactionInitiators {
122    /// ID assigned to the transaction (None if no transaction is in progress).
123    #[schema(value_type = Option<i64>)]
124    pub transaction_id: Option<TransactionId>,
125    /// Transaction phase initiated by the API.
126    #[schema(value_type = Option<TransactionPhase>)]
127    pub initiated_by_api: Option<ExternalTransactionPhase>,
128    /// Transaction phases initiated by connectors, indexed by endpoint name.
129    #[schema(value_type = BTreeMap<String, ConnectorTransactionPhase>)]
130    pub initiated_by_connectors: BTreeMap<String, ExternalConnectorTransactionPhase>,
131}
132
133/// A watermark that has been fully processed by the pipeline.
134#[derive(Clone, Debug, Deserialize, Serialize, ToSchema)]
135pub struct CompletedWatermark {
136    /// Metadata that describes the position in the input stream (e.g., Kafka partition/offset pairs).
137    #[schema(value_type = Object)]
138    pub metadata: JsonValue,
139    /// Timestamp when the data was ingested from the wire.
140    #[serde(serialize_with = "serialize_timestamp_micros")]
141    pub ingested_at: DateTime<Utc>,
142    /// Timestamp when the data was processed by the circuit.
143    #[serde(serialize_with = "serialize_timestamp_micros")]
144    pub processed_at: DateTime<Utc>,
145    /// Timestamp when all outputs produced from this input have been pushed to all output endpoints.
146    #[serde(serialize_with = "serialize_timestamp_micros")]
147    pub completed_at: DateTime<Utc>,
148}
149
150#[derive(Debug, Default, Deserialize, Serialize, ToSchema, Clone)]
151pub enum ConnectorHealthStatus {
152    #[default]
153    Healthy,
154    Unhealthy,
155}
156
157#[derive(Debug, Default, Deserialize, Serialize, ToSchema, Clone)]
158pub struct ConnectorHealth {
159    pub status: ConnectorHealthStatus,
160    pub description: Option<String>,
161}
162
163impl ConnectorHealth {
164    pub fn healthy() -> Self {
165        Self {
166            status: ConnectorHealthStatus::Healthy,
167            description: None,
168        }
169    }
170    pub fn unhealthy(description: &str) -> Self {
171        Self {
172            status: ConnectorHealthStatus::Unhealthy,
173            description: Some(description.to_string()),
174        }
175    }
176}
177
178#[derive(Debug, Default, Deserialize, Serialize, ToSchema, Clone, PartialEq, Eq)]
179pub struct ConnectorError {
180    /// Timestamp when the error occurred, serialized as RFC3339 with microseconds.
181    #[serde(serialize_with = "serialize_timestamp_micros")]
182    pub timestamp: DateTime<Utc>,
183
184    /// Sequence number of the error.
185    ///
186    /// The client can use this field to detect gaps in the error list reported
187    /// by the pipeline. When the connector reports a large number of errors, the
188    /// pipeline will only preserve and report the most recent errors of each kind.
189    pub index: u64,
190
191    /// Optional tag for the error.
192    ///
193    /// The tag is used to group errors by their type.
194    pub tag: Option<String>,
195
196    /// Error message.
197    pub message: String,
198}
199
200/// Performance metrics for an input endpoint.
201#[derive(Debug, Default, Deserialize, Serialize, ToSchema)]
202#[schema(as = InputEndpointMetrics)]
203pub struct ExternalInputEndpointMetrics {
204    /// Total bytes pushed to the endpoint since it was created.
205    pub total_bytes: u64,
206    /// Total records pushed to the endpoint since it was created.
207    pub total_records: u64,
208    /// Number of records currently buffered by the endpoint (not yet consumed by the circuit).
209    pub buffered_records: u64,
210    /// Number of bytes currently buffered by the endpoint (not yet consumed by the circuit).
211    pub buffered_bytes: u64,
212    /// Number of transport errors.
213    pub num_transport_errors: u64,
214    /// Number of parse errors.
215    pub num_parse_errors: u64,
216    /// True if end-of-input has been signaled.
217    pub end_of_input: bool,
218    /// 99th percentile processing latency (from ingesting a batch to finishing processing it) in microseconds.
219    ///
220    /// The time from ingesting a batch of records off the wire
221    /// to the circuit finishing processing them,
222    /// covering parsing, queuing, and the circuit step.
223    ///
224    /// Does not account for completion latency.
225    ///
226    /// Taken over the endpoint's sliding histogram, which holds the 10,000 most
227    /// recent samples spanning at most 10 minutes, whichever bound is reached
228    /// first. One sample is recorded per completed batch. An endpoint that stops
229    /// ingesting keeps reporting its last known latency instead of dropping to
230    /// `None`.
231    ///
232    /// `None` until the endpoint records its first sample.
233    #[serde(default, skip_serializing_if = "Option::is_none")]
234    pub processing_latency_p99_micros: Option<u64>,
235}
236
237/// Input endpoint status information.
238#[derive(Debug, Serialize, Deserialize, ToSchema)]
239#[schema(as = InputEndpointStatus)]
240pub struct ExternalInputEndpointStatus {
241    /// Endpoint name.
242    pub endpoint_name: String,
243    /// Endpoint configuration.
244    pub config: ShortEndpointConfig,
245    /// Performance metrics.
246    #[schema(value_type = InputEndpointMetrics)]
247    pub metrics: ExternalInputEndpointMetrics,
248    /// The first fatal error that occurred at the endpoint.
249    pub fatal_error: Option<String>,
250    /// Recent parse errors on this endpoint.
251    #[serde(default, skip_serializing_if = "Option::is_none")]
252    pub parse_errors: Option<Vec<ConnectorError>>,
253    /// Recent transport errors on this endpoint.
254    #[serde(default, skip_serializing_if = "Option::is_none")]
255    pub transport_errors: Option<Vec<ConnectorError>>,
256    /// Health status of the connector.
257    #[serde(default)]
258    pub health: Option<ConnectorHealth>,
259    /// Endpoint has been paused by the user.
260    pub paused: bool,
261    /// Endpoint is currently a barrier to checkpointing and suspend.
262    pub barrier: bool,
263    /// The latest completed watermark.
264    #[schema(value_type = Option<CompletedWatermark>)]
265    pub completed_frontier: Option<CompletedWatermark>,
266}
267
268/// Performance metrics for an output endpoint.
269#[derive(Debug, Default, Deserialize, Serialize, ToSchema, PartialEq, Eq, PartialOrd, Ord)]
270#[schema(as = OutputEndpointMetrics)]
271pub struct ExternalOutputEndpointMetrics {
272    /// Records sent on the underlying transport.
273    pub transmitted_records: u64,
274    /// Bytes sent on the underlying transport.
275    pub transmitted_bytes: u64,
276    /// Number of queued records.
277    pub queued_records: u64,
278    /// Number of queued batches.
279    pub queued_batches: u64,
280    /// Number of records pushed to the output buffer.
281    pub buffered_records: u64,
282    /// Number of batches in the buffer.
283    pub buffered_batches: u64,
284    /// Number of encoding errors.
285    pub num_encode_errors: u64,
286    /// Number of transport errors.
287    pub num_transport_errors: u64,
288    /// The number of input records processed by the circuit.
289    ///
290    /// This metric tracks the end-to-end progress of the pipeline: the output
291    /// of this endpoint is equal to the output of the circuit after
292    /// processing `total_processed_input_records` records.
293    ///
294    /// The counter never runs ahead of the endpoint's output. It advances to a
295    /// value `N` only once the endpoint has processed every batch derived from
296    /// the first `N` records received by the pipeline, which means transmitting
297    /// the batch, or discarding it while silent bootstrapping suppresses the
298    /// endpoint's output.
299    ///
300    /// In a multihost pipeline, this count reflects only the input records
301    /// processed on the same host as the output endpoint, which is not usually
302    /// meaningful.
303    pub total_processed_input_records: u64,
304    /// The number of steps whose input records have been processed by the
305    /// endpoint.
306    ///
307    /// This is meaningful in a multihost pipeline because steps are
308    /// synchronized across all of the hosts.
309    ///
310    /// # Interpretation
311    ///
312    /// This is a count, not a step number.  If `total_processed_steps` is 0, no
313    /// steps have been processed to completion.  If `total_processed_steps >
314    /// 0`, then the last step whose input records have been processed to
315    /// completion is `total_processed_steps - 1`. A record that was ingested in
316    /// step `n` is fully processed when `total_processed_steps > n`.
317    #[schema(value_type = u64)]
318    pub total_processed_steps: Step,
319    /// Extra memory in use beyond that used for queuing records.
320    pub memory: u64,
321    /// Number of records written so far while the connector is processing a
322    /// batch of updates.  Resets to 0 after the batch is committed.
323    ///
324    /// `None` when the connector does not support batch-progress reporting.
325    #[serde(default, skip_serializing_if = "Option::is_none")]
326    pub batch_records_written: Option<u64>,
327}
328
329/// Output endpoint status information.
330#[derive(Debug, Deserialize, Serialize, ToSchema)]
331#[schema(as = OutputEndpointStatus)]
332pub struct ExternalOutputEndpointStatus {
333    /// Endpoint name.
334    pub endpoint_name: String,
335    /// Endpoint configuration.
336    pub config: ShortEndpointConfig,
337    /// Performance metrics.
338    #[schema(value_type = OutputEndpointMetrics)]
339    pub metrics: ExternalOutputEndpointMetrics,
340    /// The first fatal error that occurred at the endpoint.
341    pub fatal_error: Option<String>,
342    /// Recent encoding errors on this endpoint.
343    #[serde(default, skip_serializing_if = "Option::is_none")]
344    pub encode_errors: Option<Vec<ConnectorError>>,
345    /// Recent transport errors on this endpoint.
346    #[serde(default, skip_serializing_if = "Option::is_none")]
347    pub transport_errors: Option<Vec<ConnectorError>>,
348    /// Health status of the connector.
349    #[serde(default)]
350    pub health: Option<ConnectorHealth>,
351}
352
353/// Global controller metrics.
354#[derive(Debug, Default, Serialize, Deserialize, ToSchema)]
355#[schema(as = GlobalControllerMetrics)]
356pub struct ExternalGlobalControllerMetrics {
357    /// State of the pipeline: running, paused, or terminating.
358    pub state: PipelineState,
359    /// The pipeline has been resumed from a checkpoint and is currently bootstrapping new and modified views.
360    pub bootstrap_in_progress: bool,
361    /// Status of the current transaction.
362    pub transaction_status: TransactionStatus,
363    /// ID of the current transaction or 0 if no transaction is in progress.
364    #[schema(value_type = i64)]
365    pub transaction_id: TransactionId,
366    /// Elapsed time in milliseconds, according to `transaction_status`:
367    ///
368    /// - [TransactionStatus::TransactionInProgress]: Time that this transaction
369    ///   has been in progress.
370    ///
371    /// - [TransactionStatus::CommitInProgress]: Time that this transaction has
372    ///   been committing.
373    pub transaction_msecs: Option<u64>,
374    /// Number of records in this transaction, according to
375    /// `transaction_status`:
376    ///
377    /// - [TransactionStatus::TransactionInProgress]: Number of records added so
378    ///   far.  More records might be added.
379    ///
380    /// - [TransactionStatus::CommitInProgress]: Final number of records.
381    pub transaction_records: Option<u64>,
382    /// Progress of the current transaction commit, if one is in progress.
383    pub commit_progress: Option<CommitProgressSummary>,
384    /// Entities that initiated the current transaction.
385    #[schema(value_type = TransactionInitiators)]
386    pub transaction_initiators: ExternalTransactionInitiators,
387    /// Phase of a concurrent bootstrap, or `Inactive` if none is in progress.
388    pub concurrent_bootstrap_phase: ConcurrentBootstrapPhase,
389    /// Progress of the concurrent bootstrap's transaction commit, if a commit is
390    /// in progress: the backfill transaction during `ConcurrentBootstrapping` and
391    /// the synchronization transaction during `Synchronizing`.
392    pub concurrent_bootstrap_progress: Option<CommitProgressSummary>,
393    /// Resident set size of the pipeline process, in bytes.
394    pub rss_bytes: u64,
395    /// Memory pressure.
396    pub memory_pressure: MemoryPressure,
397    /// Memory pressure epoch.
398    pub memory_pressure_epoch: u64,
399    /// CPU time used by the pipeline across all threads, in milliseconds.
400    pub cpu_msecs: u64,
401    /// Time since the pipeline process started, including time that the
402    /// pipeline was running or paused.
403    ///
404    /// This is the elapsed time since `start_time`.
405    pub uptime_msecs: u64,
406    /// Time at which the pipeline process started, in seconds since the epoch.
407    #[serde(with = "chrono::serde::ts_seconds")]
408    #[schema(value_type = u64)]
409    pub start_time: DateTime<Utc>,
410    /// Uniquely identifies the pipeline process that started at start_time.
411    pub incarnation_uuid: Uuid,
412    /// Time at which the pipeline process from which we resumed started, in seconds since the epoch.
413    #[serde(with = "chrono::serde::ts_seconds")]
414    #[schema(value_type = u64)]
415    pub initial_start_time: DateTime<Utc>,
416    /// Current storage usage in bytes.
417    pub storage_bytes: u64,
418    /// Storage usage integrated over time, in megabytes * seconds.
419    pub storage_mb_secs: u64,
420    /// Time elapsed while the pipeline is executing a step, multiplied by the number of threads, in milliseconds.
421    pub runtime_elapsed_msecs: u64,
422    /// Total number of records currently buffered by all endpoints.
423    pub buffered_input_records: u64,
424    /// Total number of bytes currently buffered by all endpoints.
425    pub buffered_input_bytes: u64,
426    /// Total number of records received from all endpoints.
427    pub total_input_records: u64,
428    /// Total number of bytes received from all endpoints.
429    pub total_input_bytes: u64,
430    /// Total number of input records processed by the DBSP engine.
431    pub total_processed_records: u64,
432    /// Total bytes of input records processed by the DBSP engine.
433    pub total_processed_bytes: u64,
434    /// Total number of input records processed to completion.
435    pub total_completed_records: u64,
436    /// If the pipeline is stalled because one or more output connectors' output
437    /// buffers are full, this is the number of milliseconds that the current
438    /// stall has lasted.
439    ///
440    /// If this is nonzero, then the output connectors causing the stall can be
441    /// identified by noticing `ExternalOutputEndpointMetrics::queued_records`
442    /// is greater than or equal to `ConnectorConfig::max_queued_records`.
443    ///
444    /// In the ordinary case, the pipeline is not stalled, and this value is 0.
445    pub output_stall_msecs: u64,
446    /// Number of steps that have been initiated.
447    ///
448    /// # Interpretation
449    ///
450    /// This is a count, not a step number.  If `total_initiated_steps` is 0, no
451    /// steps have been initiated.  If `total_initiated_steps > 0`, then step
452    /// `total_initiated_steps - 1` has been started and all steps previous to
453    /// that have been completely processed by the circuit.
454    #[schema(value_type = u64)]
455    pub total_initiated_steps: Step,
456    /// Number of steps whose input records have been processed to completion.
457    ///
458    /// A record is processed to completion if it has been processed by the DBSP engine and
459    /// all outputs derived from it have been processed by all output connectors.
460    ///
461    /// # Interpretation
462    ///
463    /// This is a count, not a step number.  If `total_completed_steps` is 0, no
464    /// steps have been processed to completion.  If `total_completed_steps >
465    /// 0`, then the last step whose input records have been processed to
466    /// completion is `total_completed_steps - 1`. A record that was ingested
467    /// when `total_initiated_steps` was `n` is fully processed when
468    /// `total_completed_steps >= n`.
469    #[schema(value_type = u64)]
470    pub total_completed_steps: Step,
471    /// True if the pipeline has processed all input data to completion.
472    pub pipeline_complete: bool,
473}
474
475/// Complete pipeline statistics returned by the `/stats` endpoint.
476///
477/// This schema definition matches the serialized JSON structure from
478/// `adapters::controller::ControllerStatus`. The actual implementation with
479/// atomics and mutexes lives in the adapters crate, which uses ExternalControllerStatus to
480/// register this OpenAPI schema, making it available to pipeline-manager
481/// without requiring a direct dependency on the adapters crate.
482#[derive(Debug, Deserialize, Serialize, ToSchema, Default)]
483#[schema(as = ControllerStatus)]
484pub struct ExternalControllerStatus {
485    /// Global controller metrics.
486    #[schema(value_type = GlobalControllerMetrics)]
487    pub global_metrics: ExternalGlobalControllerMetrics,
488    /// Reason why the pipeline cannot be suspended or checkpointed (if any).
489    pub suspend_error: Option<SuspendError>,
490    /// Current checkpoint activity (idle, delayed, or in-progress).
491    /// `None` when the pipeline binary predates checkpoint activity tracking.
492    pub checkpoint_activity: Option<CheckpointActivity>,
493    /// If the pipeline fundamentally cannot checkpoint (e.g. storage is not
494    /// configured, or an input endpoint does not support suspend), the reasons
495    /// are listed here.  Unlike a checkpoint failure, this means *no*
496    /// checkpoint can succeed until the pipeline configuration changes.
497    pub permanent_checkpoint_errors: Option<Vec<PermanentSuspendError>>,
498    /// Input endpoint configs and metrics.
499    #[schema(value_type = Vec<InputEndpointStatus>)]
500    pub inputs: Vec<ExternalInputEndpointStatus>,
501    /// Output endpoint configs and metrics.
502    #[schema(value_type = Vec<OutputEndpointStatus>)]
503    pub outputs: Vec<ExternalOutputEndpointStatus>,
504}
505
506fn serialize_timestamp_micros<S>(
507    timestamp: &DateTime<Utc>,
508    serializer: S,
509) -> Result<S::Ok, S::Error>
510where
511    S: serde::Serializer,
512{
513    serializer.serialize_str(&timestamp.to_rfc3339_opts(SecondsFormat::Micros, true))
514}
515
516#[cfg(test)]
517mod tests {
518    use super::ConnectorError;
519    use chrono::{DateTime, Utc};
520
521    #[test]
522    fn connector_error_timestamp_serializes_with_microsecond_precision() {
523        let error = ConnectorError {
524            timestamp: DateTime::parse_from_rfc3339("2026-03-08T05:26:42.442438448Z")
525                .unwrap()
526                .with_timezone(&Utc),
527            index: 1,
528            tag: None,
529            message: "boom".to_string(),
530        };
531
532        let json = serde_json::to_string(&error).unwrap();
533        assert!(json.contains(r#""timestamp":"2026-03-08T05:26:42.442438Z""#));
534    }
535
536    #[test]
537    fn output_metrics_batch_records_written_serializes() {
538        use super::ExternalOutputEndpointMetrics;
539
540        let metrics = ExternalOutputEndpointMetrics {
541            batch_records_written: Some(42),
542            ..Default::default()
543        };
544        let json = serde_json::to_string(&metrics).unwrap();
545        assert!(json.contains(r#""batch_records_written":42"#));
546
547        let deserialized: ExternalOutputEndpointMetrics = serde_json::from_str(&json).unwrap();
548        assert_eq!(deserialized.batch_records_written, Some(42));
549    }
550
551    #[test]
552    fn output_metrics_batch_records_written_defaults_to_none() {
553        use super::ExternalOutputEndpointMetrics;
554
555        // Old JSON without batch_records_written defaults to None.
556        let json = r#"{"transmitted_records":0,"transmitted_bytes":0,"queued_records":0,"queued_batches":0,"buffered_records":0,"buffered_batches":0,"num_encode_errors":0,"num_transport_errors":0,"total_processed_input_records":0,"total_processed_steps":0,"memory":0}"#;
557        let metrics: ExternalOutputEndpointMetrics = serde_json::from_str(json).unwrap();
558        assert_eq!(metrics.batch_records_written, None);
559    }
560}
feldera_types/adapter_stats.rs

feldera_types/
adapter_stats.rs
