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}
219
220/// Input endpoint status information.
221#[derive(Debug, Serialize, Deserialize, ToSchema)]
222#[schema(as = InputEndpointStatus)]
223pub struct ExternalInputEndpointStatus {
224    /// Endpoint name.
225    pub endpoint_name: String,
226    /// Endpoint configuration.
227    pub config: ShortEndpointConfig,
228    /// Performance metrics.
229    #[schema(value_type = InputEndpointMetrics)]
230    pub metrics: ExternalInputEndpointMetrics,
231    /// The first fatal error that occurred at the endpoint.
232    pub fatal_error: Option<String>,
233    /// Recent parse errors on this endpoint.
234    #[serde(default, skip_serializing_if = "Option::is_none")]
235    pub parse_errors: Option<Vec<ConnectorError>>,
236    /// Recent transport errors on this endpoint.
237    #[serde(default, skip_serializing_if = "Option::is_none")]
238    pub transport_errors: Option<Vec<ConnectorError>>,
239    /// Health status of the connector.
240    #[serde(default)]
241    pub health: Option<ConnectorHealth>,
242    /// Endpoint has been paused by the user.
243    pub paused: bool,
244    /// Endpoint is currently a barrier to checkpointing and suspend.
245    pub barrier: bool,
246    /// The latest completed watermark.
247    #[schema(value_type = Option<CompletedWatermark>)]
248    pub completed_frontier: Option<CompletedWatermark>,
249}
250
251/// Performance metrics for an output endpoint.
252#[derive(Debug, Default, Deserialize, Serialize, ToSchema, PartialEq, Eq, PartialOrd, Ord)]
253#[schema(as = OutputEndpointMetrics)]
254pub struct ExternalOutputEndpointMetrics {
255    /// Records sent on the underlying transport.
256    pub transmitted_records: u64,
257    /// Bytes sent on the underlying transport.
258    pub transmitted_bytes: u64,
259    /// Number of queued records.
260    pub queued_records: u64,
261    /// Number of queued batches.
262    pub queued_batches: u64,
263    /// Number of records pushed to the output buffer.
264    pub buffered_records: u64,
265    /// Number of batches in the buffer.
266    pub buffered_batches: u64,
267    /// Number of encoding errors.
268    pub num_encode_errors: u64,
269    /// Number of transport errors.
270    pub num_transport_errors: u64,
271    /// The number of input records processed by the circuit.
272    ///
273    /// This metric tracks the end-to-end progress of the pipeline: the output
274    /// of this endpoint is equal to the output of the circuit after
275    /// processing `total_processed_input_records` records.
276    ///
277    /// The counter never runs ahead of the endpoint's output. It advances to a
278    /// value `N` only once the endpoint has processed every batch derived from
279    /// the first `N` records received by the pipeline, which means transmitting
280    /// the batch, or discarding it while silent bootstrapping suppresses the
281    /// endpoint's output.
282    ///
283    /// In a multihost pipeline, this count reflects only the input records
284    /// processed on the same host as the output endpoint, which is not usually
285    /// meaningful.
286    pub total_processed_input_records: u64,
287    /// The number of steps whose input records have been processed by the
288    /// endpoint.
289    ///
290    /// This is meaningful in a multihost pipeline because steps are
291    /// synchronized across all of the hosts.
292    ///
293    /// # Interpretation
294    ///
295    /// This is a count, not a step number.  If `total_processed_steps` is 0, no
296    /// steps have been processed to completion.  If `total_processed_steps >
297    /// 0`, then the last step whose input records have been processed to
298    /// completion is `total_processed_steps - 1`. A record that was ingested in
299    /// step `n` is fully processed when `total_processed_steps > n`.
300    #[schema(value_type = u64)]
301    pub total_processed_steps: Step,
302    /// Extra memory in use beyond that used for queuing records.
303    pub memory: u64,
304    /// Number of records written so far while the connector is processing a
305    /// batch of updates.  Resets to 0 after the batch is committed.
306    ///
307    /// `None` when the connector does not support batch-progress reporting.
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub batch_records_written: Option<u64>,
310}
311
312/// Output endpoint status information.
313#[derive(Debug, Deserialize, Serialize, ToSchema)]
314#[schema(as = OutputEndpointStatus)]
315pub struct ExternalOutputEndpointStatus {
316    /// Endpoint name.
317    pub endpoint_name: String,
318    /// Endpoint configuration.
319    pub config: ShortEndpointConfig,
320    /// Performance metrics.
321    #[schema(value_type = OutputEndpointMetrics)]
322    pub metrics: ExternalOutputEndpointMetrics,
323    /// The first fatal error that occurred at the endpoint.
324    pub fatal_error: Option<String>,
325    /// Recent encoding errors on this endpoint.
326    #[serde(default, skip_serializing_if = "Option::is_none")]
327    pub encode_errors: Option<Vec<ConnectorError>>,
328    /// Recent transport errors on this endpoint.
329    #[serde(default, skip_serializing_if = "Option::is_none")]
330    pub transport_errors: Option<Vec<ConnectorError>>,
331    /// Health status of the connector.
332    #[serde(default)]
333    pub health: Option<ConnectorHealth>,
334}
335
336/// Global controller metrics.
337#[derive(Debug, Default, Serialize, Deserialize, ToSchema)]
338#[schema(as = GlobalControllerMetrics)]
339pub struct ExternalGlobalControllerMetrics {
340    /// State of the pipeline: running, paused, or terminating.
341    pub state: PipelineState,
342    /// The pipeline has been resumed from a checkpoint and is currently bootstrapping new and modified views.
343    pub bootstrap_in_progress: bool,
344    /// Status of the current transaction.
345    pub transaction_status: TransactionStatus,
346    /// ID of the current transaction or 0 if no transaction is in progress.
347    #[schema(value_type = i64)]
348    pub transaction_id: TransactionId,
349    /// Elapsed time in milliseconds, according to `transaction_status`:
350    ///
351    /// - [TransactionStatus::TransactionInProgress]: Time that this transaction
352    ///   has been in progress.
353    ///
354    /// - [TransactionStatus::CommitInProgress]: Time that this transaction has
355    ///   been committing.
356    pub transaction_msecs: Option<u64>,
357    /// Number of records in this transaction, according to
358    /// `transaction_status`:
359    ///
360    /// - [TransactionStatus::TransactionInProgress]: Number of records added so
361    ///   far.  More records might be added.
362    ///
363    /// - [TransactionStatus::CommitInProgress]: Final number of records.
364    pub transaction_records: Option<u64>,
365    /// Progress of the current transaction commit, if one is in progress.
366    pub commit_progress: Option<CommitProgressSummary>,
367    /// Entities that initiated the current transaction.
368    #[schema(value_type = TransactionInitiators)]
369    pub transaction_initiators: ExternalTransactionInitiators,
370    /// Phase of a concurrent bootstrap, or `Inactive` if none is in progress.
371    pub concurrent_bootstrap_phase: ConcurrentBootstrapPhase,
372    /// Progress of the concurrent bootstrap's transaction commit, if a commit is
373    /// in progress: the backfill transaction during `ConcurrentBootstrapping` and
374    /// the synchronization transaction during `Synchronizing`.
375    pub concurrent_bootstrap_progress: Option<CommitProgressSummary>,
376    /// Resident set size of the pipeline process, in bytes.
377    pub rss_bytes: u64,
378    /// Memory pressure.
379    pub memory_pressure: MemoryPressure,
380    /// Memory pressure epoch.
381    pub memory_pressure_epoch: u64,
382    /// CPU time used by the pipeline across all threads, in milliseconds.
383    pub cpu_msecs: u64,
384    /// Time since the pipeline process started, including time that the
385    /// pipeline was running or paused.
386    ///
387    /// This is the elapsed time since `start_time`.
388    pub uptime_msecs: u64,
389    /// Time at which the pipeline process started, in seconds since the epoch.
390    #[serde(with = "chrono::serde::ts_seconds")]
391    #[schema(value_type = u64)]
392    pub start_time: DateTime<Utc>,
393    /// Uniquely identifies the pipeline process that started at start_time.
394    pub incarnation_uuid: Uuid,
395    /// Time at which the pipeline process from which we resumed started, in seconds since the epoch.
396    #[serde(with = "chrono::serde::ts_seconds")]
397    #[schema(value_type = u64)]
398    pub initial_start_time: DateTime<Utc>,
399    /// Current storage usage in bytes.
400    pub storage_bytes: u64,
401    /// Storage usage integrated over time, in megabytes * seconds.
402    pub storage_mb_secs: u64,
403    /// Time elapsed while the pipeline is executing a step, multiplied by the number of threads, in milliseconds.
404    pub runtime_elapsed_msecs: u64,
405    /// Total number of records currently buffered by all endpoints.
406    pub buffered_input_records: u64,
407    /// Total number of bytes currently buffered by all endpoints.
408    pub buffered_input_bytes: u64,
409    /// Total number of records received from all endpoints.
410    pub total_input_records: u64,
411    /// Total number of bytes received from all endpoints.
412    pub total_input_bytes: u64,
413    /// Total number of input records processed by the DBSP engine.
414    pub total_processed_records: u64,
415    /// Total bytes of input records processed by the DBSP engine.
416    pub total_processed_bytes: u64,
417    /// Total number of input records processed to completion.
418    pub total_completed_records: u64,
419    /// If the pipeline is stalled because one or more output connectors' output
420    /// buffers are full, this is the number of milliseconds that the current
421    /// stall has lasted.
422    ///
423    /// If this is nonzero, then the output connectors causing the stall can be
424    /// identified by noticing `ExternalOutputEndpointMetrics::queued_records`
425    /// is greater than or equal to `ConnectorConfig::max_queued_records`.
426    ///
427    /// In the ordinary case, the pipeline is not stalled, and this value is 0.
428    pub output_stall_msecs: u64,
429    /// Number of steps that have been initiated.
430    ///
431    /// # Interpretation
432    ///
433    /// This is a count, not a step number.  If `total_initiated_steps` is 0, no
434    /// steps have been initiated.  If `total_initiated_steps > 0`, then step
435    /// `total_initiated_steps - 1` has been started and all steps previous to
436    /// that have been completely processed by the circuit.
437    #[schema(value_type = u64)]
438    pub total_initiated_steps: Step,
439    /// Number of steps whose input records have been processed to completion.
440    ///
441    /// A record is processed to completion if it has been processed by the DBSP engine and
442    /// all outputs derived from it have been processed by all output connectors.
443    ///
444    /// # Interpretation
445    ///
446    /// This is a count, not a step number.  If `total_completed_steps` is 0, no
447    /// steps have been processed to completion.  If `total_completed_steps >
448    /// 0`, then the last step whose input records have been processed to
449    /// completion is `total_completed_steps - 1`. A record that was ingested
450    /// when `total_initiated_steps` was `n` is fully processed when
451    /// `total_completed_steps >= n`.
452    #[schema(value_type = u64)]
453    pub total_completed_steps: Step,
454    /// True if the pipeline has processed all input data to completion.
455    pub pipeline_complete: bool,
456}
457
458/// Complete pipeline statistics returned by the `/stats` endpoint.
459///
460/// This schema definition matches the serialized JSON structure from
461/// `adapters::controller::ControllerStatus`. The actual implementation with
462/// atomics and mutexes lives in the adapters crate, which uses ExternalControllerStatus to
463/// register this OpenAPI schema, making it available to pipeline-manager
464/// without requiring a direct dependency on the adapters crate.
465#[derive(Debug, Deserialize, Serialize, ToSchema, Default)]
466#[schema(as = ControllerStatus)]
467pub struct ExternalControllerStatus {
468    /// Global controller metrics.
469    #[schema(value_type = GlobalControllerMetrics)]
470    pub global_metrics: ExternalGlobalControllerMetrics,
471    /// Reason why the pipeline cannot be suspended or checkpointed (if any).
472    pub suspend_error: Option<SuspendError>,
473    /// Current checkpoint activity (idle, delayed, or in-progress).
474    /// `None` when the pipeline binary predates checkpoint activity tracking.
475    pub checkpoint_activity: Option<CheckpointActivity>,
476    /// If the pipeline fundamentally cannot checkpoint (e.g. storage is not
477    /// configured, or an input endpoint does not support suspend), the reasons
478    /// are listed here.  Unlike a checkpoint failure, this means *no*
479    /// checkpoint can succeed until the pipeline configuration changes.
480    pub permanent_checkpoint_errors: Option<Vec<PermanentSuspendError>>,
481    /// Input endpoint configs and metrics.
482    #[schema(value_type = Vec<InputEndpointStatus>)]
483    pub inputs: Vec<ExternalInputEndpointStatus>,
484    /// Output endpoint configs and metrics.
485    #[schema(value_type = Vec<OutputEndpointStatus>)]
486    pub outputs: Vec<ExternalOutputEndpointStatus>,
487}
488
489fn serialize_timestamp_micros<S>(
490    timestamp: &DateTime<Utc>,
491    serializer: S,
492) -> Result<S::Ok, S::Error>
493where
494    S: serde::Serializer,
495{
496    serializer.serialize_str(&timestamp.to_rfc3339_opts(SecondsFormat::Micros, true))
497}
498
499#[cfg(test)]
500mod tests {
501    use super::ConnectorError;
502    use chrono::{DateTime, Utc};
503
504    #[test]
505    fn connector_error_timestamp_serializes_with_microsecond_precision() {
506        let error = ConnectorError {
507            timestamp: DateTime::parse_from_rfc3339("2026-03-08T05:26:42.442438448Z")
508                .unwrap()
509                .with_timezone(&Utc),
510            index: 1,
511            tag: None,
512            message: "boom".to_string(),
513        };
514
515        let json = serde_json::to_string(&error).unwrap();
516        assert!(json.contains(r#""timestamp":"2026-03-08T05:26:42.442438Z""#));
517    }
518
519    #[test]
520    fn output_metrics_batch_records_written_serializes() {
521        use super::ExternalOutputEndpointMetrics;
522
523        let metrics = ExternalOutputEndpointMetrics {
524            batch_records_written: Some(42),
525            ..Default::default()
526        };
527        let json = serde_json::to_string(&metrics).unwrap();
528        assert!(json.contains(r#""batch_records_written":42"#));
529
530        let deserialized: ExternalOutputEndpointMetrics = serde_json::from_str(&json).unwrap();
531        assert_eq!(deserialized.batch_records_written, Some(42));
532    }
533
534    #[test]
535    fn output_metrics_batch_records_written_defaults_to_none() {
536        use super::ExternalOutputEndpointMetrics;
537
538        // Old JSON without batch_records_written defaults to None.
539        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}"#;
540        let metrics: ExternalOutputEndpointMetrics = serde_json::from_str(json).unwrap();
541        assert_eq!(metrics.batch_records_written, None);
542    }
543}
feldera_types/adapter_stats.rs

feldera_types/
adapter_stats.rs
