From c56c5c16662ea030f3db5ba951698e083dbd9b60 Mon Sep 17 00:00:00 2001 From: Simon Kassing Date: Mon, 13 Jul 2026 11:25:15 +0200 Subject: [PATCH] pipeline-manager: use runtime status details for connector stats Instead of the manager polling all pipelines every GET request for its list when using the `status_with_connectors` selector, instead incorporate the connector stats into the runtime status details. The details are regularly updated by the runner as it does its `/status` checks. As such, it will take some time for the latest connector stats to show up after this change (approximately between 1-15 seconds). At the cost of this slight staleness, the latency of the GET request for the list of pipelines with that selector will be significantly reduced (up to 10x faster is observed). `StorageStatusDetails` is now the type shown in the API for the `storage_status_details`, instead of the generic JSON value type. The same for `RuntimeStatusDetails`. Backward incompatibility notes: - Pipelines of older versions will continue to have their connector stats shown as before. After the user base has migrated past this change, the GET request fetching fallback can be removed. - Runtime status details were directly used for the approval diff. Now, if this is present (during the `AwaitingApproval` runtime status), it is located at: `deployment_runtime_status_details.approval_diff`. The API converts automatically the old JSON values into the JSON matching the strongly typed `RuntimeStatusDetails`. The Web Console and `fda` clients are adjusted to read the approval diff from the new nested location. Co-authored-by: Karakatiza666 Signed-off-by: Karakatiza666 Signed-off-by: Simon Kassing --- crates/adapters/src/server.rs | 72 +++-- crates/fda/src/main.rs | 9 +- crates/feldera-types/src/runtime_status.rs | 67 ++++- .../src/api/endpoints/pipeline_management.rs | 174 ++++++++--- crates/pipeline-manager/src/api/examples.rs | 21 +- crates/pipeline-manager/src/api/main.rs | 4 +- crates/pipeline-manager/src/db/test.rs | 22 +- .../src/runner/pipeline_automata.rs | 39 ++- docs.feldera.com/docs/changelog.md | 13 +- .../components/pipelines/Table.svelte.spec.ts | 1 + .../src/lib/compositions/duplicatePipeline.ts | 1 + .../functions/pipelines/pipelineDiff.spec.ts | 81 ++++++ .../lib/functions/pipelines/pipelineDiff.ts | 8 +- .../src/lib/services/manager/index.ts | 13 + .../src/lib/services/manager/sdk.gen.ts | 29 ++ .../src/lib/services/manager/types.gen.ts | 275 ++++++++++++++++-- openapi.json | 63 +++- python/tests/platform/test_bootstrapping.py | 14 +- .../tests/platform/test_pipeline_lifecycle.py | 34 +++ 19 files changed, 797 insertions(+), 143 deletions(-) create mode 100644 js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.spec.ts diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index e6f821fcfa5..4d119019216 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -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, @@ -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; @@ -1469,7 +1470,7 @@ fn get_status(state: &ServerState) -> Result match Checkpointer::read_checkpoints(&**backend) { Ok(list_checkpoints) => Some(StorageStatusDetails { - checkpoints: list_checkpoints, + checkpoints: list_checkpoints.into(), }), Err(e) => { error!( @@ -1490,6 +1491,23 @@ fn get_status(state: &ServerState) -> Result, ) -> 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 @@ -1498,7 +1516,11 @@ fn get_status(state: &ServerState) -> Result Result 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, }), @@ -1582,7 +1611,7 @@ fn get_status(state: &ServerState) -> Result Result Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Suspended, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status, storage_status_details, }), @@ -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) -> Result { - let controller = state.controller()?; - let controller_status = controller.status(); - +/// Retrieves the error statistics across all endpoints. +fn retrieve_error_stats( + controller_status: &ControllerStatus, +) -> ( + Vec>, + Vec>, +) { let inputs: Vec> = controller_status .input_status() .values() @@ -1707,7 +1737,15 @@ async fn error_stats(state: WebData) -> Result) -> Result { + let controller = state.controller()?; + let controller_status = controller.status(); + let (inputs, outputs) = retrieve_error_stats(controller_status); Ok(HttpResponse::Ok().json(PipelineStatsErrorsResponse { inputs, outputs })) } diff --git a/crates/fda/src/main.rs b/crates/fda/src/main.rs index ad62757f00c..0fc4f4048db 100644 --- a/crates/fda/src/main.rs +++ b/crates/fda/src/main.rs @@ -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) @@ -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 => "".to_string(), - Some(diff) => match format { + None | Some(None) => "".to_string(), + Some(Some(diff)) => match format { OutputFormat::Text => json_to_table(&diff) .collapse() .into_pool_table() diff --git a/crates/feldera-types/src/runtime_status.rs b/crates/feldera-types/src/runtime_status.rs index d83ad2734b6..24b3c0d9880 100644 --- a/crates/feldera-types/src/runtime_status.rs +++ b/crates/feldera-types/src/runtime_status.rs @@ -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; @@ -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, + pub checkpoints: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -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, + + /// Statistics across all connectors. + /// + /// Specifically useful for: `Paused`, `Running`. + #[serde(skip_serializing_if = "Option::is_none")] + pub connector_stats: Option, + + /// The diff which is awaiting approval. + /// + /// Specifically useful for: `AwaitingApproval`. + #[serde(skip_serializing_if = "Option::is_none")] + pub approval_diff: Option, + // 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, +} diff --git a/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs b/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs index ee66d6bf2a9..f2ae451e57b 100644 --- a/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs +++ b/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs @@ -32,7 +32,8 @@ use feldera_types::config::{InputEndpointConfig, OutputEndpointConfig, RuntimeCo use feldera_types::error::ErrorResponse; use feldera_types::program_schema::ProgramSchema; use feldera_types::runtime_status::{ - BootstrapConfig, BootstrapPolicy, RuntimeDesiredStatus, RuntimeStatus, + BootstrapConfig, BootstrapPolicy, ConnectorStats, RuntimeDesiredStatus, RuntimeStatus, + RuntimeStatusDetails, StorageStatusDetails, }; use futures_util::future::join_all; use serde::{Deserialize, Serialize}; @@ -74,22 +75,6 @@ fn remove_large_fields_from_program_info( program_info } -/// Aggregated connector error statistics. -/// -/// This structure contains the sum of all error counts across all input and output connectors -/// for a pipeline. -#[derive(Serialize, Deserialize, ToSchema, Eq, PartialEq, Debug, Clone)] -pub struct ConnectorStats { - /// Total number of errors across all connectors. - /// - /// This is the sum of: - /// - `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, -} - /// Pipeline information. /// It both includes fields which are user-provided and system-generated. #[derive(Serialize, ToSchema, PartialEq, Debug, Clone)] @@ -116,7 +101,7 @@ pub struct PipelineInfo { pub deployment_error: Option, pub refresh_version: Version, pub storage_status: StorageStatus, - pub storage_status_details: Option, + pub storage_status_details: Option, pub deployment_id: Option, pub deployment_initial: Option, pub deployment_status: CombinedStatus, @@ -129,7 +114,7 @@ pub struct PipelineInfo { pub deployment_resources_desired_status: ResourcesDesiredStatus, pub deployment_resources_desired_status_since: DateTime, pub deployment_runtime_status: Option, - pub deployment_runtime_status_details: Option, + pub deployment_runtime_status_details: Option, pub deployment_runtime_status_since: Option>, pub deployment_runtime_desired_status: Option, pub deployment_runtime_desired_status_since: Option>, @@ -139,8 +124,9 @@ pub struct PipelineInfo { /// /// This is the struct that is actually serialized when a response body type /// is [`PipelineInfo`] according to the OpenAPI specification. -/// The difference are the types of `runtime_config`, `program_config` and -/// `program_info` fields, which are JSON values rather than their actual ones. +/// The difference are the types of `runtime_config`, `program_config`, +/// `program_info`, `storage_status_details` and `deployment_runtime_status_details` fields, +/// which are JSON values rather than their actual ones. /// This ensures that even when a backward incompatible change occurred for /// any of these fields, the API still works (i.e., able to serialize them in /// order to return pipeline(s)). @@ -237,7 +223,9 @@ impl PipelineInfoInternal { deployment_resources_desired_status_since: extended_pipeline .deployment_resources_desired_status_since, deployment_runtime_status: extended_pipeline.deployment_runtime_status, - deployment_runtime_status_details: extended_pipeline.deployment_runtime_status_details, + deployment_runtime_status_details: backward_compatible_runtime_status_details( + extended_pipeline.deployment_runtime_status_details, + ), deployment_runtime_status_since: extended_pipeline.deployment_runtime_status_since, deployment_runtime_desired_status: extended_pipeline.deployment_runtime_desired_status, deployment_runtime_desired_status_since: extended_pipeline @@ -280,7 +268,7 @@ pub struct PipelineSelectedInfo { pub deployment_error: Option, pub refresh_version: Version, pub storage_status: StorageStatus, - pub storage_status_details: Option, + pub storage_status_details: Option, pub deployment_id: Option, pub deployment_initial: Option, pub deployment_status: CombinedStatus, @@ -294,7 +282,7 @@ pub struct PipelineSelectedInfo { pub deployment_resources_desired_status: ResourcesDesiredStatus, pub deployment_resources_desired_status_since: DateTime, pub deployment_runtime_status: Option, - pub deployment_runtime_status_details: Option, + pub deployment_runtime_status_details: Option, pub deployment_runtime_status_since: Option>, pub deployment_runtime_desired_status: Option, pub deployment_runtime_desired_status_since: Option>, @@ -417,7 +405,9 @@ impl PipelineSelectedInfoInternal { deployment_resources_desired_status_since: extended_pipeline .deployment_resources_desired_status_since, deployment_runtime_status: extended_pipeline.deployment_runtime_status, - deployment_runtime_status_details: extended_pipeline.deployment_runtime_status_details, + deployment_runtime_status_details: backward_compatible_runtime_status_details( + extended_pipeline.deployment_runtime_status_details, + ), deployment_runtime_status_since: extended_pipeline.deployment_runtime_status_since, deployment_runtime_desired_status: extended_pipeline.deployment_runtime_desired_status, deployment_runtime_desired_status_since: extended_pipeline @@ -477,7 +467,9 @@ impl PipelineSelectedInfoInternal { deployment_resources_desired_status_since: extended_pipeline .deployment_resources_desired_status_since, deployment_runtime_status: extended_pipeline.deployment_runtime_status, - deployment_runtime_status_details: extended_pipeline.deployment_runtime_status_details, + deployment_runtime_status_details: backward_compatible_runtime_status_details( + extended_pipeline.deployment_runtime_status_details, + ), deployment_runtime_status_since: extended_pipeline.deployment_runtime_status_since, deployment_runtime_desired_status: extended_pipeline.deployment_runtime_desired_status, deployment_runtime_desired_status_since: extended_pipeline @@ -739,6 +731,40 @@ pub struct PostStopPipelineParameters { force: bool, } +/// Converts old runtime status details to the new strongly typed one. +fn backward_compatible_runtime_status_details( + details: Option, +) -> Option { + details.map(|details| match details { + serde_json::Value::Null => RuntimeStatusDetails::default().serialize_guaranteed(), + serde_json::Value::Bool(b) => { + RuntimeStatusDetails::new_only_reason(&format!("Boolean: {b}")).serialize_guaranteed() + } + serde_json::Value::Number(n) => { + RuntimeStatusDetails::new_only_reason(&format!("Number: {n}")).serialize_guaranteed() + } + serde_json::Value::String(s) => { + RuntimeStatusDetails::new_only_reason(&s).serialize_guaranteed() + } + serde_json::Value::Array(a) => { + RuntimeStatusDetails::new_only_reason(&format!("Array: {a:?}")).serialize_guaranteed() + } + serde_json::Value::Object(obj) => { + if obj.get("program_diff").is_some() { + // Backward compatibility: the entire runtime status details is actually the approval + // diff, as such it must be restructured. + RuntimeStatusDetails { + approval_diff: Some(serde_json::Value::Object(obj)), + ..Default::default() + } + .serialize_guaranteed() + } else { + serde_json::Value::Object(obj) + } + } + }) +} + /// List Pipelines /// /// Retrieve the list of pipelines. @@ -798,13 +824,8 @@ pub(crate) async fn list_pipelines( let tenant_id = *tenant_id; let pipeline_name = pipeline.name.clone(); async move { - fetch_connector_error_stats( - &state, - tenant_id, - &pipeline_name, - pipeline.deployment_runtime_status, - ) - .await + fetch_connector_error_stats(&state, tenant_id, &pipeline_name, pipeline) + .await } }) .collect(); @@ -838,10 +859,25 @@ async fn fetch_connector_error_stats( state: &WebData, tenant_id: TenantId, pipeline_name: &str, - deployment_runtime_status: Option, + pipeline: &ExtendedPipelineDescrMonitoring, ) -> Option { + // First attempt to retrieve the connector statistics from the runtime status details if they + // are available there. The fetching afterward is added for backward compatibility. Once the + // runtime status details changes are sufficiently long present and the user base has migrated + // past it, then the HTTP fetching can be removed. + let details = backward_compatible_runtime_status_details( + pipeline.deployment_runtime_status_details.clone(), + ); + if let Some(value) = details { + if let Ok(details) = serde_json::from_value::(value) { + if let Some(connector_stats) = details.connector_stats { + return Some(connector_stats); + } + } + }; + // Only forward the request if the pipeline is in a valid runtime status - match deployment_runtime_status { + match pipeline.deployment_runtime_status { Some(RuntimeStatus::Bootstrapping) | Some(RuntimeStatus::Replaying) | Some(RuntimeStatus::Running) @@ -976,13 +1012,8 @@ pub(crate) async fn get_pipeline( .await .get_pipeline_for_monitoring(*tenant_id, &pipeline_name) .await?; - let connector_stats = fetch_connector_error_stats( - &state, - *tenant_id, - &pipeline_name, - pipeline.deployment_runtime_status, - ) - .await; + let connector_stats = + fetch_connector_error_stats(&state, *tenant_id, &pipeline_name, &pipeline).await; PipelineSelectedInfoInternal::new_status_with_connectors(pipeline, connector_stats) } }; @@ -1805,3 +1836,62 @@ pub(crate) async fn post_pipeline_testing( Ok(HttpResponse::Ok().finish()) } + +#[cfg(test)] +mod tests { + use crate::api::endpoints::pipeline_management::backward_compatible_runtime_status_details; + use feldera_types::runtime_status::{ConnectorStats, RuntimeStatusDetails}; + use serde_json::json; + + #[test] + fn test_backward_compatible_runtime_status_details() { + for (input, expected) in [ + (None, None), + (Some(json!(null)), Some(json!({}))), + ( + Some(json!(false)), + Some(json!({"reason": "Boolean: false"})), + ), + (Some(json!(123)), Some(json!({"reason": "Number: 123"}))), + (Some(json!("abc")), Some(json!({"reason": "abc"}))), + ( + Some(json!([1, 2, 3])), + Some(json!({"reason": "Array: [Number(1), Number(2), Number(3)]"})), + ), + (Some(json!({"abc": "def"})), Some(json!({"abc": "def"}))), + ( + Some(json!({"program_diff": "xyz"})), + Some(json!({"approval_diff": { "program_diff": "xyz" }})), + ), + ( + Some(RuntimeStatusDetails::default().serialize_guaranteed()), + Some(json!({})), + ), + ( + Some(RuntimeStatusDetails::new_only_reason("abc").serialize_guaranteed()), + Some(json!({"reason": "abc"})), + ), + ( + Some( + RuntimeStatusDetails { + reason: Some("abc".to_string()), + connector_stats: Some(ConnectorStats { num_errors: 123 }), + approval_diff: Some(json!({"a": "b"})), + } + .serialize_guaranteed(), + ), + Some(json!({ + "reason": "abc", + "connector_stats": { + "num_errors": 123 + }, + "approval_diff": { + "a": "b" + } + })), + ), + ] { + assert_eq!(backward_compatible_runtime_status_details(input), expected); + } + } +} diff --git a/crates/pipeline-manager/src/api/examples.rs b/crates/pipeline-manager/src/api/examples.rs index c965e7e44a6..7463b9e257d 100644 --- a/crates/pipeline-manager/src/api/examples.rs +++ b/crates/pipeline-manager/src/api/examples.rs @@ -23,6 +23,7 @@ use crate::runner::interaction::{ format_disconnected_error_message, format_timeout_error_message, RunnerInteraction, }; use feldera_types::config::{DevTweaks, FtConfig, ResourceConfig, StorageOptions}; +use feldera_types::runtime_status::{RuntimeStatusDetails, StorageStatusDetails}; use feldera_types::{config::RuntimeConfig, error::ErrorResponse}; use uuid::uuid; @@ -215,7 +216,10 @@ fn pipeline_info_internal_to_external(pipeline: PipelineInfoInternal) -> Pipelin deployment_error: pipeline.deployment_error, refresh_version: pipeline.refresh_version, storage_status: pipeline.storage_status, - storage_status_details: pipeline.storage_status_details, + storage_status_details: pipeline.storage_status_details.map(|v| { + serde_json::from_value::(v) + .expect("example should be deserializable") + }), deployment_id: pipeline.deployment_id, deployment_initial: pipeline.deployment_initial, deployment_status: pipeline.deployment_status, @@ -229,7 +233,10 @@ fn pipeline_info_internal_to_external(pipeline: PipelineInfoInternal) -> Pipelin deployment_resources_desired_status_since: pipeline .deployment_resources_desired_status_since, deployment_runtime_status: pipeline.deployment_runtime_status, - deployment_runtime_status_details: pipeline.deployment_runtime_status_details, + deployment_runtime_status_details: pipeline.deployment_runtime_status_details.map(|v| { + serde_json::from_value::(v) + .expect("example should be deserializable") + }), deployment_runtime_status_since: pipeline.deployment_runtime_status_since, deployment_runtime_desired_status: pipeline.deployment_runtime_desired_status, deployment_runtime_desired_status_since: pipeline.deployment_runtime_desired_status_since, @@ -278,7 +285,10 @@ fn pipeline_selected_info_internal_to_external( deployment_error: pipeline.deployment_error, refresh_version: pipeline.refresh_version, storage_status: pipeline.storage_status, - storage_status_details: pipeline.storage_status_details, + storage_status_details: pipeline.storage_status_details.map(|v| { + serde_json::from_value::(v) + .expect("example should be deserializable") + }), deployment_id: pipeline.deployment_id, deployment_initial: pipeline.deployment_initial, deployment_status: pipeline.deployment_status, @@ -292,7 +302,10 @@ fn pipeline_selected_info_internal_to_external( deployment_resources_desired_status_since: pipeline .deployment_resources_desired_status_since, deployment_runtime_status: pipeline.deployment_runtime_status, - deployment_runtime_status_details: pipeline.deployment_runtime_status_details, + deployment_runtime_status_details: pipeline.deployment_runtime_status_details.map(|v| { + serde_json::from_value::(v) + .expect("example should be deserializable") + }), deployment_runtime_status_since: pipeline.deployment_runtime_status_since, deployment_runtime_desired_status: pipeline.deployment_runtime_desired_status, deployment_runtime_desired_status_since: pipeline.deployment_runtime_desired_status_since, diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs index 4121329c5c4..42e7f13c711 100644 --- a/crates/pipeline-manager/src/api/main.rs +++ b/crates/pipeline-manager/src/api/main.rs @@ -283,7 +283,9 @@ It contains the following fields: feldera_types::runtime_status::RuntimeStatus, feldera_types::runtime_status::RuntimeDesiredStatus, feldera_types::runtime_status::BootstrapPolicy, - crate::api::endpoints::pipeline_management::ConnectorStats, + feldera_types::runtime_status::RuntimeStatusDetails, + feldera_types::runtime_status::StorageStatusDetails, + feldera_types::runtime_status::ConnectorStats, crate::api::endpoints::pipeline_management::PipelineInfo, crate::api::endpoints::pipeline_management::PipelineSelectedInfo, crate::api::endpoints::pipeline_management::PipelineFieldSelector, diff --git a/crates/pipeline-manager/src/db/test.rs b/crates/pipeline-manager/src/db/test.rs index a094f2c74c2..fbbb8d59368 100644 --- a/crates/pipeline-manager/src/db/test.rs +++ b/crates/pipeline-manager/src/db/test.rs @@ -50,7 +50,7 @@ use proptest_derive::Arbitrary; use serde_json::json; use std::borrow::BorrowMut; use std::borrow::Cow; -use std::collections::{BTreeMap, VecDeque}; +use std::collections::BTreeMap; use std::fmt::Debug; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -617,8 +617,8 @@ fn limited_optional_storage_status_details() -> impl Strategy PipelineAutomaton { deployment_resources_status_details: details, extended_runtime_status: ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Initializing, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::new_only_reason( + "initializing set by the runner", + ) + .serialize_guaranteed(), runtime_desired_status: deployment_initial, storage_status_details: None, }, @@ -1506,19 +1510,19 @@ impl PipelineAutomaton { match response_str { "Paused" => Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Paused, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Paused, storage_status_details: None, }), "Running" => Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Running, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Running, storage_status_details: None, }), "Initializing" => Ok(ExtendedRuntimeStatus { // Backward compatibility: in anticipation of recent change of 503 to 200 runtime_status: RuntimeStatus::Initializing, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Paused, storage_status_details: None, }), @@ -1528,7 +1532,7 @@ impl PipelineAutomaton { })), _ => Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!("Pipeline status response (200 OK) is an unexpected JSON string: '{response_str}'")), + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!("Pipeline status response (200 OK) is an unexpected JSON string: '{response_str}'")).serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, storage_status_details: None, }), @@ -1539,7 +1543,7 @@ impl PipelineAutomaton { Ok(response) => Ok(response), Err(e) => Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!("Pipeline status response (200 OK) cannot be deserialized due to: {e}")), + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!("Pipeline status response (200 OK) cannot be deserialized due to: {e}")).serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, storage_status_details: None, }), @@ -1548,7 +1552,7 @@ impl PipelineAutomaton { // JSON response must be either a string or an object. Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!("Pipeline status response (200 OK) is not a string or an object:\n{body:#}")), + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!("Pipeline status response (200 OK) is not a string or an object:\n{body:#}")).serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, storage_status_details: None, }) @@ -1559,13 +1563,13 @@ impl PipelineAutomaton { match error_response.error_code.as_ref() { "Initializing" => Ok(ExtendedRuntimeStatus { // For backward compatibility runtime_status: RuntimeStatus::Initializing, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Paused, storage_status_details: None, }), "Suspended" => Ok(ExtendedRuntimeStatus { // For backward compatibility runtime_status: RuntimeStatus::Suspended, - runtime_status_details: json!(""), + runtime_status_details: RuntimeStatusDetails::default().serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Suspended, storage_status_details: None, }), @@ -1577,16 +1581,16 @@ impl PipelineAutomaton { ); Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!("Pipeline status response (503 Service Unavailable) is an error:\n{error_response:?}")), + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!("Pipeline status response (503 Service Unavailable) is an error:\n{error_response:?}")).serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, - storage_status_details: None, + storage_status_details: None, }) }, } } Err(e) => Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!("Pipeline status response (503 Service Unavailable) cannot be deserialized due to: {e}. Response was:\n{body:#}")), + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!("Pipeline status response (503 Service Unavailable) cannot be deserialized due to: {e}. Response was:\n{body:#}")).serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, storage_status_details: None, }) @@ -1615,7 +1619,9 @@ impl PipelineAutomaton { { Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Initializing, - runtime_status_details: json!(format!("Still in the grace period for initializing. Pipeline status endpoint cannot yet be reached due to: {e}")), + runtime_status_details: RuntimeStatusDetails::new_only_reason( + &format!("Still in the grace period for initializing. Pipeline status endpoint cannot yet be reached due to: {e}") + ).serialize_guaranteed(), runtime_desired_status: deployment_initial, storage_status_details: None, }) @@ -1627,9 +1633,10 @@ impl PipelineAutomaton { ); Ok(ExtendedRuntimeStatus { runtime_status: RuntimeStatus::Unavailable, - runtime_status_details: json!(format!( + runtime_status_details: RuntimeStatusDetails::new_only_reason(&format!( "Pipeline status endpoint could not be reached: {e}" - )), + )) + .serialize_guaranteed(), runtime_desired_status: RuntimeDesiredStatus::Unavailable, storage_status_details: None, }) diff --git a/docs.feldera.com/docs/changelog.md b/docs.feldera.com/docs/changelog.md index 34f641ac25b..880d5c23063 100644 --- a/docs.feldera.com/docs/changelog.md +++ b/docs.feldera.com/docs/changelog.md @@ -12,7 +12,16 @@ import TabItem from '@theme/TabItem'; - ## Unreleased + ## Unreleased + + - Pipeline API field `deployment_runtime_status_details` is now strongly typed, + whereas before it was just a generic JSON value type. While `AwaitingApproval`, + the diff is now located at `deployment_runtime_status_details.approval_diff` + instead of being the whole details itself. + + - Pipelines from this latest version onward will have their GET selector + `status_with_connectors` connector stats cached, which are now updated along + with the runtime status details within roughly 1-15s. - Cluster monitor events with information on the backing (Kubernetes) resources is no longer gated behind unstable feature `cluster_monitor_resources` (deprecated). @@ -22,7 +31,7 @@ import TabItem from '@theme/TabItem'; The cluster monitoring of resources can still be disabled by setting in the Helm chart `disableClusterMonitorResources` to `true`. - - A bug fix introduced a backward incompatible change to the replay journal format. + - A bug fix introduced a backward incompatible change to the replay journal format. This only affects pipelines configured with exactly-once fault tolerance. Such pipelines should not be upgraded to the new Feldera runtime if they are in a failed state with non-empty replay journal. Upgrade is possible once the pipeline has been diff --git a/js-packages/web-console/src/lib/components/pipelines/Table.svelte.spec.ts b/js-packages/web-console/src/lib/components/pipelines/Table.svelte.spec.ts index 9193a3b05ed..ff72e078672 100644 --- a/js-packages/web-console/src/lib/components/pipelines/Table.svelte.spec.ts +++ b/js-packages/web-console/src/lib/components/pipelines/Table.svelte.spec.ts @@ -58,6 +58,7 @@ const thumb = (name: string): PipelineThumb => programConfig: { runtime_version: null }, deploymentResourcesStatus: 'Stopped', deploymentResourcesStatusSince: new Date(lastChange[name]), + deploymentRuntimeStatusDetails: { connector_stats: { num_errors: 0 } }, connectors: { numErrors: 0 } }) as unknown as PipelineThumb diff --git a/js-packages/web-console/src/lib/compositions/duplicatePipeline.ts b/js-packages/web-console/src/lib/compositions/duplicatePipeline.ts index 68d9f0f63a5..4aacb1c598b 100644 --- a/js-packages/web-console/src/lib/compositions/duplicatePipeline.ts +++ b/js-packages/web-console/src/lib/compositions/duplicatePipeline.ts @@ -40,6 +40,7 @@ export const optimisticDuplicatePipelineThumb = ( programStatusSince: now.toISOString(), deploymentResourcesStatus: 'Stopped', deploymentResourcesStatusSince: now, + deploymentRuntimeStatusDetails: undefined, connectors: undefined } } diff --git a/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.spec.ts b/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.spec.ts new file mode 100644 index 00000000000..a5a7a3654a8 --- /dev/null +++ b/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.spec.ts @@ -0,0 +1,81 @@ +/** + * Unit tests for `parsePipelineDiff`, which reads the approval diff from + * `deployment_runtime_status_details.approval_diff`. Older pipelines stored the + * diff as the whole runtime status details value; the manager rewrites those + * into the `approval_diff` shape, so the parser only handles the new location. + */ +import { describe, expect, it } from 'vitest' +import type { ExtendedPipeline } from '$lib/services/pipelineManager' +import { parsePipelineDiff } from './pipelineDiff' + +const approvalDiff = { + program_diff: { + added_tables: ['t_added'], + removed_tables: [], + modified_tables: ['t_modified'], + added_views: [], + removed_views: ['v_removed'], + modified_views: [] + }, + program_diff_error: null, + added_input_connectors: ['in_added'], + modified_input_connectors: [], + removed_input_connectors: [], + added_output_connectors: [], + modified_output_connectors: ['out_modified'], + removed_output_connectors: [] +} + +const pipeline = ( + details: unknown +): Pick => + ({ + status: 'AwaitingApproval', + deploymentRuntimeStatusDetails: details + }) as Pick + +describe('parsePipelineDiff', () => { + it('parses the diff nested under approval_diff', () => { + const diff = parsePipelineDiff(pipeline({ approval_diff: approvalDiff })) + expect(diff.tables).toEqual({ added: ['t_added'], removed: [], modified: ['t_modified'] }) + expect(diff.views).toEqual({ added: [], removed: ['v_removed'], modified: [] }) + expect(diff.inputConnectors.added).toEqual(['in_added']) + expect(diff.outputConnectors.modified).toEqual(['out_modified']) + expect(diff.error).toBeUndefined() + }) + + it('defaults program table/view diffs when program_diff is null', () => { + const diff = parsePipelineDiff( + pipeline({ approval_diff: { ...approvalDiff, program_diff: null } }) + ) + expect(diff.tables).toEqual({ added: [], removed: [], modified: [] }) + expect(diff.views).toEqual({ added: [], removed: [], modified: [] }) + }) + + it('surfaces program_diff_error', () => { + const diff = parsePipelineDiff( + pipeline({ approval_diff: { ...approvalDiff, program_diff_error: 'boom' } }) + ) + expect(diff.error).toBe('boom') + }) + + it('throws when the expected approval info is not there', () => { + expect(() => + parsePipelineDiff({ + status: 'AwaitingApproval', + deploymentRuntimeStatusDetails: {} + } as Pick) + ).toThrow('data is not available') + }) + + it('throws when approval_diff is absent', () => { + // A running-connector runtime status may set only `connector_stats`. + expect(() => parsePipelineDiff(pipeline({ connector_stats: { num_errors: 0 } }))).toThrow( + 'not available' + ) + }) + + it('throws when there are no runtime status details at all', () => { + expect(() => parsePipelineDiff(pipeline(undefined))).toThrow('not available') + }) +}) diff --git a/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.ts b/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.ts index 7ebc91234bb..ac711c6c4c0 100644 --- a/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.ts +++ b/js-packages/web-console/src/lib/functions/pipelines/pipelineDiff.ts @@ -32,12 +32,16 @@ export const parsePipelineDiff = ( throw new Error('Pipeline is not awaiting approval') } - if (!pipeline.deploymentRuntimeStatusDetails) { + // The approval diff lives at `deployment_runtime_status_details.approval_diff`. + // The manager rewrites older pipelines, whose entire runtime status details + // were the diff, into this shape, so a single path covers all versions. + const approvalDiff = pipeline.deploymentRuntimeStatusDetails?.approval_diff + if (!approvalDiff) { throw new Error('Pipeline diff data is not available') } try { - const rawDiff = va.parse(pipelineDiffSchema, pipeline.deploymentRuntimeStatusDetails) + const rawDiff = va.parse(pipelineDiffSchema, approvalDiff) return { tables: rawDiff.program_diff ? { diff --git a/js-packages/web-console/src/lib/services/manager/index.ts b/js-packages/web-console/src/lib/services/manager/index.ts index 1555ae309bd..0b973bbd576 100644 --- a/js-packages/web-console/src/lib/services/manager/index.ts +++ b/js-packages/web-console/src/lib/services/manager/index.ts @@ -34,6 +34,7 @@ export { getPipelineSupportBundle, getPipelineTimeSeries, getPipelineTimeSeriesStream, + getRemoteCheckpoints, httpInput, httpOutput, listApiKeys, @@ -89,6 +90,7 @@ export type { CheckpointResponse, CheckpointStatus, CheckpointSyncFailure, + CheckpointSyncResponse, CheckpointSyncStatus, Chunk, ClientMetadata, @@ -161,6 +163,8 @@ export type { Demo, DevTweaks, DisplaySchedule, + DynamoDbWriteMode, + DynamoDbWriterConfig, ErrorResponse, ExtendedClusterMonitorEvent, Field, @@ -302,6 +306,11 @@ export type { GetPipelineTimeSeriesStreamErrors, GetPipelineTimeSeriesStreamResponse, GetPipelineTimeSeriesStreamResponses, + GetRemoteCheckpointsData, + GetRemoteCheckpointsError, + GetRemoteCheckpointsErrors, + GetRemoteCheckpointsResponse, + GetRemoteCheckpointsResponses, GlobalControllerMetrics, GlueCatalogConfig, HealthStatus, @@ -467,6 +476,7 @@ export type { PostPipelineTestingErrors, PostPipelineTestingResponses, PostPutPipeline, + PostprocessorConfig, PostStopPipelineParameters, PostUpdateRuntimeData, PostUpdateRuntimeError, @@ -493,6 +503,7 @@ export type { RedisOutputConfig, Rel, Relation, + RemoteCheckpoint, ReplayPolicy, ResourceConfig, ResourcesDesiredStatus, @@ -502,6 +513,7 @@ export type { RuntimeConfig, RuntimeDesiredStatus, RuntimeStatus, + RuntimeStatusDetails, RustCompilationInfo, S3InputConfig, SampleStatistics, @@ -530,6 +542,7 @@ export type { StorageConfig, StorageOptions, StorageStatus, + StorageStatusDetails, SuspendError, SyncCheckpointData, SyncCheckpointError, diff --git a/js-packages/web-console/src/lib/services/manager/sdk.gen.ts b/js-packages/web-console/src/lib/services/manager/sdk.gen.ts index 8f4ecab02cd..fd4a1ec0bf4 100644 --- a/js-packages/web-console/src/lib/services/manager/sdk.gen.ts +++ b/js-packages/web-console/src/lib/services/manager/sdk.gen.ts @@ -101,6 +101,9 @@ import type { GetPipelineTimeSeriesStreamData, GetPipelineTimeSeriesStreamErrors, GetPipelineTimeSeriesStreamResponses, + GetRemoteCheckpointsData, + GetRemoteCheckpointsErrors, + GetRemoteCheckpointsResponses, HttpInputData, HttpInputErrors, HttpInputResponses, @@ -696,6 +699,10 @@ export const getCheckpointStatus = ( * Get the checkpoints for a pipeline * * Retrieve the current checkpoints made by a pipeline. + * + * **Stability note**: for multihost pipelines, this endpoint returns the + * combined checkpoint list from all hosts. The shape of this response may + * change in a future release. */ export const getCheckpoints = ( options: Options @@ -712,6 +719,28 @@ export const getCheckpoints = ( ...options }) +/** + * List checkpoints in remote object storage + * + * Retrieve the list of checkpoints available in the configured remote object + * storage (e.g., S3). Requires the pipeline to be running with a sync + * storage configuration. + */ +export const getRemoteCheckpoints = ( + options: Options +) => + (options.client ?? client).get< + GetRemoteCheckpointsResponses, + GetRemoteCheckpointsErrors, + ThrowOnError, + 'data' + >({ + responseStyle: 'data', + security: [{ scheme: 'bearer', type: 'http' }], + url: '/v0/pipelines/{pipeline_name}/checkpoints/remote', + ...options + }) + /** * Performance Profile JSON * diff --git a/js-packages/web-console/src/lib/services/manager/types.gen.ts b/js-packages/web-console/src/lib/services/manager/types.gen.ts index 14fb6b841be..c3d591b3267 100644 --- a/js-packages/web-console/src/lib/services/manager/types.gen.ts +++ b/js-packages/web-console/src/lib/services/manager/types.gen.ts @@ -266,6 +266,13 @@ export type CheckpointSyncFailure = { uuid: string } +/** + * Response to a sync checkpoint request. + */ +export type CheckpointSyncResponse = { + checkpoint_uuid: string +} + /** * Checkpoint status returned by the `/checkpoint/sync_status` endpoint. */ @@ -780,6 +787,13 @@ export type ConnectorConfig = OutputBufferConfig & { * The default is `false`. */ paused?: boolean + /** + * Optional postprocessor configuration + */ + postprocessor?: Array | null + /** + * Optional preprocessor configuration + */ preprocessor?: Array | null /** * Send a full snapshot of a materialized view when the connector first @@ -840,16 +854,12 @@ export type ConnectorHealth = { export type ConnectorHealthStatus = 'Healthy' | 'Unhealthy' /** - * Aggregated connector error statistics. - * - * This structure contains the sum of all error counts across all input and output connectors - * for a pipeline. + * Statistics across all connectors. */ export type ConnectorStats = { /** * Total number of errors across all connectors. * - * This is the sum of: * - `num_transport_errors` from all input connectors * - `num_parse_errors` from all input connectors * - `num_encode_errors` from all output connectors @@ -1770,6 +1780,93 @@ export type DisplaySchedule = } | 'Always' +/** + * DynamoDB write API used by the output connector. + */ +export type DynamoDbWriteMode = 'batch' | 'transactional' + +/** + * DynamoDB output connector configuration. + */ +export type DynamoDbWriterConfig = { + /** + * AWS access key ID. + * + * If both `aws_access_key_id` and `aws_secret_access_key` are specified, + * the connector uses these static credentials. Otherwise it uses the + * default AWS credential provider chain, including IAM Roles for Service + * Accounts (IRSA) in EKS. + * + * Static credentials are treated as long-lived IAM keys: no session token + * is sent, so STS-issued temporary credentials are not supported via these + * fields. To use temporary credentials, rely on the default provider chain + * instead (leave these unset). + */ + aws_access_key_id?: string | null + /** + * AWS secret access key. + */ + aws_secret_access_key?: string | null + /** + * Maximum number of write requests in one DynamoDB write call. + * + * DynamoDB supports at most 100 for `TransactWriteItems` and at most 25 + * for `BatchWriteItem`. If omitted, the connector uses the maximum for + * the selected `write_mode`. + */ + batch_size?: number | null + /** + * Optional endpoint URL, for example when using a local + * DynamoDB-compatible service. + */ + endpoint_url?: string | null + /** + * Maximum number of bytes buffered by each worker before flushing writes. + * + * This is an approximate size based on encoded DynamoDB attributes. + */ + max_buffer_size_bytes?: number + /** + * Maximum number of DynamoDB write requests in flight per worker thread. + * + * The total in-flight request count across the connector is + * `threads × max_concurrent_requests`. Size this accordingly when + * tuning against a provisioned-throughput table to avoid excessive + * throttling. + */ + max_concurrent_requests?: number + /** + * Maximum number of retries for a failed or partially-applied DynamoDB write chunk. + * + * For `batch` writes, `BatchWriteItem` may return some items as "unprocessed" in a + * successful 200 response; those are re-submitted and counted as retries. For + * `transactional` writes, a failed `TransactWriteItems` call is retried in full. + * + * Transient errors (throttling, network failures) are first handled transparently by the + * AWS SDK. Only attempts that reach this connector's retry loop count against this limit. + * Each retry waits longer than the previous one (exponential backoff), up to a ceiling + * of roughly 13 seconds. + * + * Set to `null` to retry indefinitely. After the backoff ceiling is reached the connector + * keeps retrying at that interval, providing backpressure until DynamoDB recovers. + * Defaults to `10`. + */ + max_retries?: number | null + /** + * AWS region. + */ + region: string + /** + * Name of the DynamoDB table to write to. + */ + table: string + /** + * Number of worker threads used to encode and write disjoint key ranges. + */ + threads?: number + write_mode?: DynamoDbWriteMode +} + /** * Information returned by REST API endpoints on error. */ @@ -3149,8 +3246,7 @@ export type OutputBufferConfig = { * total number of updates output by the pipeline. Updates to the * same record can overwrite or cancel previous updates. * - * By default, the buffer can grow indefinitely until one of - * the other output conditions is satisfied. + * The default is 10,000,000. * * NOTE: this configuration option requires the `enable_output_buffer` flag * to be set. @@ -3634,7 +3730,7 @@ export type PipelineInfo = ClientMetadata & { deployment_runtime_desired_status?: RuntimeDesiredStatus | null deployment_runtime_desired_status_since?: string | null deployment_runtime_status?: RuntimeStatus | null - deployment_runtime_status_details?: unknown + deployment_runtime_status_details?: RuntimeStatusDetails | null deployment_runtime_status_since?: string | null deployment_status: CombinedStatus deployment_status_since: string @@ -3651,7 +3747,7 @@ export type PipelineInfo = ClientMetadata & { refresh_version: Version runtime_config: RuntimeConfig storage_status: StorageStatus - storage_status_details?: unknown + storage_status_details?: StorageStatusDetails | null udf_rust: string udf_toml: string version: Version @@ -3705,7 +3801,7 @@ export type PipelineSelectedInfo = ClientMetadata & { deployment_runtime_desired_status?: RuntimeDesiredStatus | null deployment_runtime_desired_status_since?: string | null deployment_runtime_status?: RuntimeStatus | null - deployment_runtime_status_details?: unknown + deployment_runtime_status_details?: RuntimeStatusDetails | null deployment_runtime_status_since?: string | null deployment_status: CombinedStatus deployment_status_since: string @@ -3722,7 +3818,7 @@ export type PipelineSelectedInfo = ClientMetadata & { refresh_version: Version runtime_config?: RuntimeConfig | null storage_status: StorageStatus - storage_status_details?: unknown + storage_status_details?: StorageStatusDetails | null udf_rust?: string | null udf_toml?: string | null version: Version @@ -4073,6 +4169,22 @@ export type PostgresWriterConfig = { uri: string } +/** + * Configuration for describing a postprocessor + */ +export type PostprocessorConfig = { + /** + * Arbitrary additional configuration expected by the postprocessor + * encoded as a JSON Value. + */ + config: unknown + /** + * Name of the postprocessor. + * All postprocessors with the same name will perform the same task. + */ + name: string +} + /** * Configuration for describing a preprocessor */ @@ -4130,6 +4242,24 @@ export type ProgramConfig = { * If not set (null), the runtime version will be the same as the platform version. */ runtime_version?: string | null + /** + * Use the platform SQL compiler when a non-platform `runtime_version` is specified. + * + * Warning: This setting is experimental and may change in the future. + * Requires the platform to run with the unstable feature `runtime_version` enabled. + * + * When `false` (default), the SQL compiler matching the `runtime_version` is + * downloaded and used. When `true`, the platform's SQL compiler is used instead. + * + * Setting this to `true` avoids downloading the runtime-version-specific SQL + * compiler JAR (e.g., when network access is unavailable or slow), at the cost + * of potentially using a mismatched SQL compiler. The Rust runtime sources are + * still checked out and compiled from the requested `runtime_version`. + * + * Has no effect when `runtime_version` is not set or the platform does not have + * the unstable feature `runtime_version` enabled. + */ + use_platform_compiler?: boolean } /** @@ -4391,6 +4521,16 @@ export type Relation = SqlIdentifier & { } } +/** + * A checkpoint that exists in remote object storage. + */ +export type RemoteCheckpoint = { + /** + * UUID of the checkpoint. + */ + uuid: string +} + export type ReplayPolicy = 'Instant' | 'Original' export type ResourceConfig = { @@ -4843,6 +4983,26 @@ export type RuntimeStatus = | 'Running' | 'Suspended' +/** + * 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`. + */ +export type RuntimeStatusDetails = { + /** + * The diff which is awaiting approval. + * + * Specifically useful for: `AwaitingApproval`. + */ + approval_diff?: unknown + connector_stats?: ConnectorStats | null + /** + * Free form text giving an explanation why it is currently in this runtime status. + * + * Specifically useful for: `Unavailable`, `Initializing`. + */ + reason?: string | null +} + /** * Rust compilation information. */ @@ -5025,7 +5185,7 @@ export type SqlIdentifier = { } /** - * The available SQL types as specified in `CREATE` statements. + * The available SQL column type names. Each value is the platform's wire encoding of the type (e.g. `BIGINT`, `INTEGER`, `INTERVAL_DAY`), not valid SQL type syntax. */ export type SqlType = | 'BOOLEAN' @@ -5180,6 +5340,17 @@ export type StorageOptions = { */ export type StorageStatus = 'Cleared' | 'InUse' | 'Clearing' +/** + * Details about pipeline storage, which are returned as part of the regular runtime status polling + * by the runner. + */ +export type StorageStatusDetails = { + /** + * Present checkpoints. + */ + checkpoints: Array +} + /** * Whether a pipeline supports checkpointing and suspend-and-resume. */ @@ -5268,6 +5439,19 @@ export type SyncConfig = { * Default: 10 */ multi_thread_streams?: number | null + /** + * When true, checkpoint downloads use the maximum resources available on + * the host: `transfers` and `checkers` are scaled to the number of CPUs, + * and the download buffer is allowed to grow up to most of the available + * memory. This maximizes download throughput at the cost of higher CPU and + * memory usage during a pull. + * + * When false, downloads use the values configured via `transfers`, + * `checkers`, and the rclone defaults instead. + * + * Default: true + */ + optimize_download_resources?: boolean /** * The name of the cloud storage provider (e.g., `"AWS"`, `"Minio"`). * @@ -5334,19 +5518,9 @@ export type SyncConfig = { */ secret_key?: string | null /** - * When `true`, the pipeline starts in **standby** mode; processing doesn't - * start until activation (`POST /activate`). - * If this pipeline was previously activated and the storage has not been - * cleared, the pipeline will auto activate, no newer checkpoints will be - * fetched. - * - * Standby behavior depends on `start_from_checkpoint`: - * - If `latest`, pipeline continuously fetches the latest available - * checkpoint until activated. - * - If checkpoint UUID, pipeline fetches this checkpoint once and waits - * in standby until activated. + * **Deprecated.** Use `initial=standby` when starting the pipeline instead. * - * Default: `false` + * @deprecated */ standby?: boolean start_from_checkpoint?: StartFromCheckpoint | null @@ -5465,6 +5639,10 @@ export type TransportConfig = config: DeltaTableWriterConfig name: 'delta_table_output' } + | { + config: DynamoDbWriterConfig + name: 'dynamodb_output' + } | { config: RedisOutputConfig name: 'redis_output' @@ -5690,7 +5868,7 @@ export type GetApiKeyResponses = { /** * API key retrieved successfully */ - 200: ApiKeyDescr + 200: Array } export type GetApiKeyResponse = GetApiKeyResponses[keyof GetApiKeyResponses] @@ -6095,7 +6273,7 @@ export type PostPipelineActivateResponses = { /** * Pipeline activation initiated */ - 202: CheckpointResponse + 202: string } export type PostPipelineActivateResponse = @@ -6131,9 +6309,9 @@ export type PostPipelineApproveError = PostPipelineApproveErrors[keyof PostPipel export type PostPipelineApproveResponses = { /** - * Pipeline activation initiated + * Bootstrap approved */ - 202: CheckpointResponse + 200: string } export type PostPipelineApproveResponse = @@ -6197,9 +6375,9 @@ export type SyncCheckpointError = SyncCheckpointErrors[keyof SyncCheckpointError export type SyncCheckpointResponses = { /** - * Checkpoint synced to object store + * Checkpoint sync to object store initiated */ - 200: CheckpointResponse + 202: CheckpointSyncResponse } export type SyncCheckpointResponse = SyncCheckpointResponses[keyof SyncCheckpointResponses] @@ -6296,13 +6474,46 @@ export type GetCheckpointsError = GetCheckpointsErrors[keyof GetCheckpointsError export type GetCheckpointsResponses = { /** - * Checkpoints retrieved successfully + * Checkpoints retrieved successfully. For multihost pipelines the list contains entries from all hosts; the shape of this response may change in a future release. */ - 200: CheckpointMetadata + 200: Array } export type GetCheckpointsResponse = GetCheckpointsResponses[keyof GetCheckpointsResponses] +export type GetRemoteCheckpointsData = { + body?: never + path: { + /** + * Unique pipeline name + */ + pipeline_name: string + } + query?: never + url: '/v0/pipelines/{pipeline_name}/checkpoints/remote' +} + +export type GetRemoteCheckpointsErrors = { + /** + * Pipeline with that name does not exist + */ + 404: ErrorResponse + 500: ErrorResponse + 503: ErrorResponse +} + +export type GetRemoteCheckpointsError = GetRemoteCheckpointsErrors[keyof GetRemoteCheckpointsErrors] + +export type GetRemoteCheckpointsResponses = { + /** + * Remote checkpoints retrieved successfully. + */ + 200: Array +} + +export type GetRemoteCheckpointsResponse = + GetRemoteCheckpointsResponses[keyof GetRemoteCheckpointsResponses] + export type GetPipelineCircuitJsonProfileData = { body?: never path: { diff --git a/openapi.json b/openapi.json index 15e8043bcee..ae463787026 100644 --- a/openapi.json +++ b/openapi.json @@ -8189,7 +8189,7 @@ }, "ConnectorStats": { "type": "object", - "description": "Aggregated connector error statistics.\n\nThis structure contains the sum of all error counts across all input and output connectors\nfor a pipeline.", + "description": "Statistics across all connectors.", "required": [ "num_errors" ], @@ -8197,7 +8197,7 @@ "num_errors": { "type": "integer", "format": "int64", - "description": "Total number of errors across all connectors.\n\nThis is the sum of:\n- `num_transport_errors` from all input connectors\n- `num_parse_errors` from all input connectors\n- `num_encode_errors` from all output connectors\n- `num_transport_errors` from all output connectors", + "description": "Total number of errors across all connectors.\n\n- `num_transport_errors` from all input connectors\n- `num_parse_errors` from all input connectors\n- `num_encode_errors` from all output connectors\n- `num_transport_errors` from all output connectors", "minimum": 0 } } @@ -11425,6 +11425,11 @@ "nullable": true }, "deployment_runtime_status_details": { + "allOf": [ + { + "$ref": "#/components/schemas/RuntimeStatusDetails" + } + ], "nullable": true }, "deployment_runtime_status_since": { @@ -11485,6 +11490,11 @@ "$ref": "#/components/schemas/StorageStatus" }, "storage_status_details": { + "allOf": [ + { + "$ref": "#/components/schemas/StorageStatusDetails" + } + ], "nullable": true }, "udf_rust": { @@ -11690,6 +11700,11 @@ "nullable": true }, "deployment_runtime_status_details": { + "allOf": [ + { + "$ref": "#/components/schemas/RuntimeStatusDetails" + } + ], "nullable": true }, "deployment_runtime_status_since": { @@ -11766,6 +11781,11 @@ "$ref": "#/components/schemas/StorageStatus" }, "storage_status_details": { + "allOf": [ + { + "$ref": "#/components/schemas/StorageStatusDetails" + } + ], "nullable": true }, "udf_rust": { @@ -13145,6 +13165,29 @@ "Suspended" ] }, + "RuntimeStatusDetails": { + "type": "object", + "description": "Details about the current runtime status. The fields in this struct should all be **optional**\nand set only by a runtime status when they are known. Otherwise, they can just be set `None`.", + "properties": { + "approval_diff": { + "description": "The diff which is awaiting approval.\n\nSpecifically useful for: `AwaitingApproval`.", + "nullable": true + }, + "connector_stats": { + "allOf": [ + { + "$ref": "#/components/schemas/ConnectorStats" + } + ], + "nullable": true + }, + "reason": { + "type": "string", + "description": "Free form text giving an explanation why it is currently in this runtime status.\n\nSpecifically useful for: `Unavailable`, `Initializing`.", + "nullable": true + } + } + }, "RustCompilationInfo": { "type": "object", "description": "Rust compilation information.", @@ -13685,6 +13728,22 @@ "Clearing" ] }, + "StorageStatusDetails": { + "type": "object", + "description": "Details about pipeline storage, which are returned as part of the regular runtime status polling\nby the runner.", + "required": [ + "checkpoints" + ], + "properties": { + "checkpoints": { + "type": "array", + "items": { + "$ref": "#/components/schemas/CheckpointMetadata" + }, + "description": "Present checkpoints." + } + } + }, "SuspendError": { "oneOf": [ { diff --git a/python/tests/platform/test_bootstrapping.py b/python/tests/platform/test_bootstrapping.py index 325e7dad9e9..3c232405a1a 100644 --- a/python/tests/platform/test_bootstrapping.py +++ b/python/tests/platform/test_bootstrapping.py @@ -91,7 +91,7 @@ def test_bootstrap_enterprise(pipeline_name): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff["program_diff"] == { "added_tables": [], @@ -132,7 +132,7 @@ def test_bootstrap_enterprise(pipeline_name): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff["program_diff"] == { "added_tables": ["t2"], @@ -173,7 +173,7 @@ def test_bootstrap_enterprise(pipeline_name): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff["program_diff"] == { "added_tables": [], @@ -219,7 +219,7 @@ def test_bootstrap_enterprise(pipeline_name): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff["program_diff"] == { "added_tables": [], @@ -253,7 +253,7 @@ def test_bootstrap_enterprise(pipeline_name): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff["program_diff"] == { "added_tables": [], @@ -624,7 +624,7 @@ def gen_sql(connectors): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff == { "added_input_connectors": ["t1.unnamed-0", "t1.unnamed-1"], @@ -673,7 +673,7 @@ def gen_sql(connectors): pipeline.start(bootstrap_policy=BootstrapPolicy.AWAIT_APPROVAL) assert pipeline.status() == PipelineStatus.AWAITINGAPPROVAL - diff = pipeline.deployment_runtime_status_details() + diff = pipeline.deployment_runtime_status_details()["approval_diff"] print(f"Pipeline diff: {diff}") assert diff == { "added_input_connectors": [], diff --git a/python/tests/platform/test_pipeline_lifecycle.py b/python/tests/platform/test_pipeline_lifecycle.py index 2fb70487eb1..a57bd22457d 100644 --- a/python/tests/platform/test_pipeline_lifecycle.py +++ b/python/tests/platform/test_pipeline_lifecycle.py @@ -850,3 +850,37 @@ def test_start_failed_compilation(pipeline_name): except FelderaAPIError as e: error = e assert error is not None and error.error_code == "CannotStartWithCompilationError" + + +@gen_pipeline_name +def test_connector_stats_errors_count(pipeline_name): + """ + Tests that the connector statistics number of errors is set. + """ + pipeline = PipelineBuilder( + TEST_CLIENT, pipeline_name, "CREATE TABLE t1 (v1 INT);" + ).create_or_replace() + pipeline.start() + assert _ingress(pipeline_name, "t1", "1\n2\n3\n").status_code == HTTPStatus.OK + assert ( + _ingress(pipeline_name, "t1", "a\nb\nc\nd\n").status_code + == HTTPStatus.BAD_REQUEST + ) + assert _ingress(pipeline_name, "t1", "4\n5\n6\n").status_code == HTTPStatus.OK + start_s = time.monotonic() + while True: + num_errors = pipeline.deployment_runtime_status_details()["connector_stats"][ + "num_errors" + ] + if num_errors == 4: + break + elif time.monotonic() - start_s >= 30.0 or num_errors > 4: + raise ValueError(f"Number of errors is not 4 but {num_errors}") + time.sleep(0.5) + assert pipeline.stats().global_metrics.total_completed_records == 6 + assert ( + TEST_CLIENT.http.get( + f"/pipelines/{pipeline_name}?selector=status_with_connectors" + )["connectors"]["num_errors"] + == 4 + )