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    /// Endpoint has been paused by the user.
352    ///
353    /// A paused output endpoint discards the output it receives instead of
354    /// sending it to its sink.
355    #[serde(default)]
356    pub paused: bool,
357}
358
359/// Global controller metrics.
360#[derive(Debug, Default, Serialize, Deserialize, ToSchema)]
361#[schema(as = GlobalControllerMetrics)]
362pub struct ExternalGlobalControllerMetrics {
363    /// State of the pipeline: running, paused, or terminating.
364    pub state: PipelineState,
365    /// The pipeline has been resumed from a checkpoint and is currently bootstrapping new and modified views.
366    pub bootstrap_in_progress: bool,
367    /// Status of the current transaction.
368    pub transaction_status: TransactionStatus,
369    /// ID of the current transaction or 0 if no transaction is in progress.
370    #[schema(value_type = i64)]
371    pub transaction_id: TransactionId,
372    /// Elapsed time in milliseconds, according to `transaction_status`:
373    ///
374    /// - [TransactionStatus::TransactionInProgress]: Time that this transaction
375    ///   has been in progress.
376    ///
377    /// - [TransactionStatus::CommitInProgress]: Time that this transaction has
378    ///   been committing.
379    pub transaction_msecs: Option<u64>,
380    /// Number of records in this transaction, according to
381    /// `transaction_status`:
382    ///
383    /// - [TransactionStatus::TransactionInProgress]: Number of records added so
384    ///   far.  More records might be added.
385    ///
386    /// - [TransactionStatus::CommitInProgress]: Final number of records.
387    pub transaction_records: Option<u64>,
388    /// Progress of the current transaction commit, if one is in progress.
389    pub commit_progress: Option<CommitProgressSummary>,
390    /// Entities that initiated the current transaction.
391    #[schema(value_type = TransactionInitiators)]
392    pub transaction_initiators: ExternalTransactionInitiators,
393    /// Phase of a concurrent bootstrap, or `Inactive` if none is in progress.
394    pub concurrent_bootstrap_phase: ConcurrentBootstrapPhase,
395    /// Progress of the concurrent bootstrap's transaction commit, if a commit is
396    /// in progress: the backfill transaction during `ConcurrentBootstrapping` and
397    /// the synchronization transaction during `Synchronizing`.
398    pub concurrent_bootstrap_progress: Option<CommitProgressSummary>,
399    /// Resident set size of the pipeline process, in bytes.
400    pub rss_bytes: u64,
401    /// Memory pressure.
402    pub memory_pressure: MemoryPressure,
403    /// Memory pressure epoch.
404    pub memory_pressure_epoch: u64,
405    /// CPU time used by the pipeline across all threads, in milliseconds.
406    pub cpu_msecs: u64,
407    /// Time since the pipeline process started, including time that the
408    /// pipeline was running or paused.
409    ///
410    /// This is the elapsed time since `start_time`.
411    pub uptime_msecs: u64,
412    /// Time at which the pipeline process started, in seconds since the epoch.
413    #[serde(with = "chrono::serde::ts_seconds")]
414    #[schema(value_type = u64)]
415    pub start_time: DateTime<Utc>,
416    /// Uniquely identifies the pipeline process that started at start_time.
417    pub incarnation_uuid: Uuid,
418    /// Time at which the pipeline process from which we resumed started, in seconds since the epoch.
419    #[serde(with = "chrono::serde::ts_seconds")]
420    #[schema(value_type = u64)]
421    pub initial_start_time: DateTime<Utc>,
422    /// Current storage usage in bytes.
423    pub storage_bytes: u64,
424    /// Storage usage integrated over time, in megabytes * seconds.
425    pub storage_mb_secs: u64,
426    /// Time elapsed while the pipeline is executing a step, multiplied by the number of threads, in milliseconds.
427    pub runtime_elapsed_msecs: u64,
428    /// Total number of records currently buffered by all endpoints.
429    pub buffered_input_records: u64,
430    /// Total number of bytes currently buffered by all endpoints.
431    pub buffered_input_bytes: u64,
432    /// Total number of records received from all endpoints.
433    pub total_input_records: u64,
434    /// Total number of bytes received from all endpoints.
435    pub total_input_bytes: u64,
436    /// Total number of input records processed by the DBSP engine.
437    pub total_processed_records: u64,
438    /// Total bytes of input records processed by the DBSP engine.
439    pub total_processed_bytes: u64,
440    /// Total number of input records processed to completion.
441    pub total_completed_records: u64,
442    /// If the pipeline is stalled because one or more output connectors' output
443    /// buffers are full, this is the number of milliseconds that the current
444    /// stall has lasted.
445    ///
446    /// If this is nonzero, then the output connectors causing the stall can be
447    /// identified by noticing `ExternalOutputEndpointMetrics::queued_records`
448    /// is greater than or equal to `ConnectorConfig::max_queued_records`.
449    ///
450    /// In the ordinary case, the pipeline is not stalled, and this value is 0.
451    pub output_stall_msecs: u64,
452    /// Number of steps that have been initiated.
453    ///
454    /// # Interpretation
455    ///
456    /// This is a count, not a step number.  If `total_initiated_steps` is 0, no
457    /// steps have been initiated.  If `total_initiated_steps > 0`, then step
458    /// `total_initiated_steps - 1` has been started and all steps previous to
459    /// that have been completely processed by the circuit.
460    #[schema(value_type = u64)]
461    pub total_initiated_steps: Step,
462    /// Number of steps whose input records have been processed to completion.
463    ///
464    /// A record is processed to completion if it has been processed by the DBSP engine and
465    /// all outputs derived from it have been processed by all output connectors.
466    ///
467    /// # Interpretation
468    ///
469    /// This is a count, not a step number.  If `total_completed_steps` is 0, no
470    /// steps have been processed to completion.  If `total_completed_steps >
471    /// 0`, then the last step whose input records have been processed to
472    /// completion is `total_completed_steps - 1`. A record that was ingested
473    /// when `total_initiated_steps` was `n` is fully processed when
474    /// `total_completed_steps >= n`.
475    #[schema(value_type = u64)]
476    pub total_completed_steps: Step,
477    /// True if the pipeline has processed all input data to completion.
478    pub pipeline_complete: bool,
479}
480
481/// Complete pipeline statistics returned by the `/stats` endpoint.
482///
483/// This schema definition matches the serialized JSON structure from
484/// `adapters::controller::ControllerStatus`. The actual implementation with
485/// atomics and mutexes lives in the adapters crate, which uses ExternalControllerStatus to
486/// register this OpenAPI schema, making it available to pipeline-manager
487/// without requiring a direct dependency on the adapters crate.
488#[derive(Debug, Deserialize, Serialize, ToSchema, Default)]
489#[schema(as = ControllerStatus)]
490pub struct ExternalControllerStatus {
491    /// Global controller metrics.
492    #[schema(value_type = GlobalControllerMetrics)]
493    pub global_metrics: ExternalGlobalControllerMetrics,
494    /// Reason why the pipeline cannot be suspended or checkpointed (if any).
495    pub suspend_error: Option<SuspendError>,
496    /// Current checkpoint activity (idle, delayed, or in-progress).
497    /// `None` when the pipeline binary predates checkpoint activity tracking.
498    pub checkpoint_activity: Option<CheckpointActivity>,
499    /// If the pipeline fundamentally cannot checkpoint (e.g. storage is not
500    /// configured, or an input endpoint does not support suspend), the reasons
501    /// are listed here.  Unlike a checkpoint failure, this means *no*
502    /// checkpoint can succeed until the pipeline configuration changes.
503    pub permanent_checkpoint_errors: Option<Vec<PermanentSuspendError>>,
504    /// Input endpoint configs and metrics.
505    #[schema(value_type = Vec<InputEndpointStatus>)]
506    pub inputs: Vec<ExternalInputEndpointStatus>,
507    /// Output endpoint configs and metrics.
508    #[schema(value_type = Vec<OutputEndpointStatus>)]
509    pub outputs: Vec<ExternalOutputEndpointStatus>,
510}
511
512fn serialize_timestamp_micros<S>(
513    timestamp: &DateTime<Utc>,
514    serializer: S,
515) -> Result<S::Ok, S::Error>
516where
517    S: serde::Serializer,
518{
519    serializer.serialize_str(&timestamp.to_rfc3339_opts(SecondsFormat::Micros, true))
520}
521
522#[cfg(test)]
523mod tests {
524    use super::ConnectorError;
525    use chrono::{DateTime, Utc};
526
527    #[test]
528    fn connector_error_timestamp_serializes_with_microsecond_precision() {
529        let error = ConnectorError {
530            timestamp: DateTime::parse_from_rfc3339("2026-03-08T05:26:42.442438448Z")
531                .unwrap()
532                .with_timezone(&Utc),
533            index: 1,
534            tag: None,
535            message: "boom".to_string(),
536        };
537
538        let json = serde_json::to_string(&error).unwrap();
539        assert!(json.contains(r#""timestamp":"2026-03-08T05:26:42.442438Z""#));
540    }
541
542    #[test]
543    fn output_metrics_batch_records_written_serializes() {
544        use super::ExternalOutputEndpointMetrics;
545
546        let metrics = ExternalOutputEndpointMetrics {
547            batch_records_written: Some(42),
548            ..Default::default()
549        };
550        let json = serde_json::to_string(&metrics).unwrap();
551        assert!(json.contains(r#""batch_records_written":42"#));
552
553        let deserialized: ExternalOutputEndpointMetrics = serde_json::from_str(&json).unwrap();
554        assert_eq!(deserialized.batch_records_written, Some(42));
555    }
556
557    #[test]
558    fn output_metrics_batch_records_written_defaults_to_none() {
559        use super::ExternalOutputEndpointMetrics;
560
561        // Old JSON without batch_records_written defaults to None.
562        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}"#;
563        let metrics: ExternalOutputEndpointMetrics = serde_json::from_str(json).unwrap();
564        assert_eq!(metrics.batch_records_written, None);
565    }
566}
feldera_types/adapter_stats.rs

feldera_types/
adapter_stats.rs
