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

feldera_types/
adapter_stats.rs
