Skip to content

pipeline-manager: use runtime status details for connector stats #6564

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snkas merged 1 commit into from
Jul 15, 2026
Merged

pipeline-manager: use runtime status details for connector stats #6564

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 55 additions & 17 deletions crates/adapters/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use crate::server::metrics::{
use crate::static_compile::catalog::OUTPUT_MAPPING;
use crate::transport::http::HttpOutputFormat;
use crate::util::{LongOperationWarning, RateLimitCheckResult, TokenBucketRateLimiter};
use crate::{Catalog, dyn_event};
use crate::{Catalog, ControllerStatus, dyn_event};
use crate::{
CircuitCatalog, Controller, ControllerError, FormatConfig, InputEndpointConfig, OutputEndpoint,
OutputEndpointConfig, PipelineConfig, TransportInputEndpoint,
Expand Down Expand Up @@ -74,8 +74,9 @@ use feldera_types::query_params::{
SamplyProfileParams,
};
use feldera_types::runtime_status::{
BootstrapConfig, BootstrapPolicy, ExtendedRuntimeStatus, ExtendedRuntimeStatusError,
RuntimeDesiredStatus, RuntimeStatus, StorageStatusDetails,
BootstrapConfig, BootstrapPolicy, ConnectorStats, ExtendedRuntimeStatus,
ExtendedRuntimeStatusError, RuntimeDesiredStatus, RuntimeStatus, RuntimeStatusDetails,
StorageStatusDetails,
};
use feldera_types::suspend::{SuspendError, SuspendableResponse};
use feldera_types::time_series::TimeSeries;
Expand Down Expand Up @@ -1469,7 +1470,7 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
let storage_status_details = match &state.storage {
Some(backend) => match Checkpointer::read_checkpoints(&**backend) {
Ok(list_checkpoints) => Some(StorageStatusDetails {
checkpoints: list_checkpoints,
checkpoints: list_checkpoints.into(),
}),
Err(e) => {
error!(
Expand All @@ -1490,6 +1491,23 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
default_status: RuntimeStatus,
storage_status_details: Option<StorageStatusDetails>,
) -> ExtendedRuntimeStatus {
// Error statistics across connectors
let (inputs, outputs) = retrieve_error_stats(controller.status());
let mut total_errors = 0u64;
for endpoint in inputs {
total_errors = total_errors
.saturating_add(endpoint.metrics.num_transport_errors)
.saturating_add(endpoint.metrics.num_parse_errors);
}
for endpoint in outputs {
total_errors = total_errors
.saturating_add(endpoint.metrics.num_encode_errors)
.saturating_add(endpoint.metrics.num_transport_errors);
}
let connector_stats = ConnectorStats {
num_errors: total_errors,
};

ExtendedRuntimeStatus {
runtime_status: if controller.status().bootstrap_in_progress() {
RuntimeStatus::Bootstrapping
Expand All @@ -1498,7 +1516,11 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
} else {
default_status
},
runtime_status_details: json!(""),
runtime_status_details: RuntimeStatusDetails {
connector_stats: Some(connector_stats),
..Default::default()
}
.serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}
Expand Down Expand Up @@ -1532,25 +1554,32 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
PipelinePhase::Initializing(inner) => match inner {
InitializationState::Starting => Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::Initializing,
runtime_status_details: json!(""),
runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}),
InitializationState::DownloadingCheckpoint => Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::Initializing,
runtime_status_details: json!("downloading checkpoint from object storage"),
runtime_status_details: RuntimeStatusDetails::new_only_reason(
"downloading checkpoint from object storage",
)
.serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}),
InitializationState::Standby => Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::Standby,
runtime_status_details: json!(""),
runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}),
InitializationState::AwaitingApproval(diff) => Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::AwaitingApproval,
runtime_status_details: serde_json::to_value(&diff).unwrap_or_default(),
runtime_status_details: RuntimeStatusDetails {
approval_diff: Some(serde_json::to_value(&diff).unwrap_or_default()),
..Default::default()
}
.serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}),
Expand Down Expand Up @@ -1582,7 +1611,7 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
if matches!(*e, ControllerError::RestoreInProgress) {
Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::Replaying,
runtime_status_details: json!(""),
runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
})
Expand All @@ -1609,7 +1638,7 @@ fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRunt
}
PipelinePhase::Suspended => Ok(ExtendedRuntimeStatus {
runtime_status: RuntimeStatus::Suspended,
runtime_status_details: json!(""),
runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(),
runtime_desired_status,
storage_status_details,
}),
Expand Down Expand Up @@ -1678,12 +1707,13 @@ async fn stats(
Ok(HttpResponse::Ok().json(state.controller()?.api_status(include_connector_errors)))
}

/// This endpoint returns a subset of stats that don't need updating and so is more performant than /stats
#[get("/stats/errors")]
async fn error_stats(state: WebData<ServerState>) -> Result<HttpResponse, PipelineError> {
let controller = state.controller()?;
let controller_status = controller.status();

/// Retrieves the error statistics across all endpoints.
fn retrieve_error_stats(
controller_status: &ControllerStatus,
) -> (
Vec<EndpointErrorStats<InputEndpointErrorMetrics>>,
Vec<EndpointErrorStats<OutputEndpointErrorMetrics>>,
) {
let inputs: Vec<EndpointErrorStats<InputEndpointErrorMetrics>> = controller_status
.input_status()
.values()
Expand All @@ -1707,7 +1737,15 @@ async fn error_stats(state: WebData<ServerState>) -> Result<HttpResponse, Pipeli
},
})
.collect();
(inputs, outputs)
}

/// This endpoint returns a subset of stats that don't need updating and so is more performant than /stats
#[get("/stats/errors")]
async fn error_stats(state: WebData<ServerState>) -> Result<HttpResponse, PipelineError> {
let controller = state.controller()?;
let controller_status = controller.status();
let (inputs, outputs) = retrieve_error_stats(controller_status);
Ok(HttpResponse::Ok().json(PipelineStatsErrorsResponse { inputs, outputs }))
}

Expand Down
9 changes: 5 additions & 4 deletions crates/fda/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1076,7 +1076,7 @@ async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client)
)
.await;
if status == CombinedStatus::AwaitingApproval {
let diff = client
let runtime_status_details = client
.get_pipeline()
.pipeline_name(name.clone())
.selector(PipelineFieldSelector::Status)
Expand All @@ -1095,10 +1095,11 @@ async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client)
"Pipeline definition has changed since the last checkpoint. The pipeline is awaiting approval to proceed with bootstrapping the modified components. Run 'fda approve' to approve the changes or 'fda stop' to terminate the pipeline. Summary of changes:"
);

let diff_str = match diff {
let diff_str = match runtime_status_details.map(|details| details.approval_diff)
{
// Normally shouldn't happen.
None => "<not available>".to_string(),
Some(diff) => match format {
None | Some(None) => "<not available>".to_string(),
Some(Some(diff)) => match format {
OutputFormat::Text => json_to_table(&diff)
.collapse()
.into_pool_table()
Expand Down
67 changes: 64 additions & 3 deletions crates/feldera-types/src/runtime_status.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use actix_web::{HttpRequest, HttpResponse, HttpResponseBuilder, Responder, Respo
use bytemuck::NoUninit;
use clap::ValueEnum;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use serde_json::json;
use std::fmt;
use std::fmt::Display;
use utoipa::ToSchema;
Expand Down Expand Up @@ -260,10 +260,10 @@ impl BootstrapConfig {

/// Details about pipeline storage, which are returned as part of the regular runtime status polling
/// by the runner.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, ToSchema)]
pub struct StorageStatusDetails {
/// Present checkpoints.
pub checkpoints: VecDeque<CheckpointMetadata>,
pub checkpoints: Vec<CheckpointMetadata>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
Expand Down Expand Up @@ -348,3 +348,64 @@ impl ResponseError for ExtendedRuntimeStatusError {
HttpResponseBuilder::new(self.status_code()).json(self.error.clone())
}
}

/// Details about the current runtime status. The fields in this struct should all be **optional**
/// and set only by a runtime status when they are known. Otherwise, they can just be set `None`.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default, Eq, ToSchema)]
pub struct RuntimeStatusDetails {
/// Free form text giving an explanation why it is currently in this runtime status.
///
/// Specifically useful for: `Unavailable`, `Initializing`.
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,

/// Statistics across all connectors.
///
/// Specifically useful for: `Paused`, `Running`.

@mythical-fred mythical-fred Jun 30, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this field is part of the OpenAPI spec (via ToSchema), the rustdoc should mention the staleness window. Something like: "Updated periodically by the runner's status polling (approximately every 1–15 seconds). Values may not reflect the most recent state."

#[serde(skip_serializing_if = "Option::is_none")]
pub connector_stats: Option<ConnectorStats>,

/// The diff which is awaiting approval.
///
/// Specifically useful for: `AwaitingApproval`.
#[serde(skip_serializing_if = "Option::is_none")]
pub approval_diff: Option<serde_json::Value>,
// Backward compatibility: in older versions, the approval diff was the runtime status details
// value itself. To distinguish between the old and new version, clients can check if the
// `program_diff` field (one of the fields within the `approval_diff`) is present if they
// expect there to be a diff (i.e., when the runtime status is `AwaitingApproval`). As such,
// `program_diff` is a reserved field name that cannot be added here in the future.
}

impl RuntimeStatusDetails {
pub fn new_only_reason(reason: &str) -> Self {
Self {
reason: Some(reason.to_string()),
..Self::default()
}
}

/// Serializes the runtime status details to JSON. If the serialization errors, an error JSON
/// is returned instead. This makes sure this method does not panic unexpectedly and that the
/// error bubbles up. The details are only supplementary information, and as such are not
/// critical to operation.
pub fn serialize_guaranteed(self) -> serde_json::Value {
serde_json::to_value(self).unwrap_or_else(|e| {
json!({
"reason": format!("unable to serialize runtime status details due to: {e}")
})
})
}
}

/// Statistics across all connectors.
#[derive(Serialize, Deserialize, ToSchema, Eq, PartialEq, Debug, Clone)]
pub struct ConnectorStats {
/// Total number of errors across all connectors.
///
/// - `num_transport_errors` from all input connectors
/// - `num_parse_errors` from all input connectors
/// - `num_encode_errors` from all output connectors
/// - `num_transport_errors` from all output connectors
pub num_errors: u64,
}
Loading
Loading