From 77a4fc5f32e0bc1c8997dc508440bd59cdcf2c4d Mon Sep 17 00:00:00 2001 From: Subhra264 Date: Mon, 13 Jul 2026 17:53:21 +0530 Subject: [PATCH] fix: backfill should work if only remote destination is present (#13152) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Adds backfill support for pipelines whose only destination is a **remote destination**. ## Why Backfilling a remote-only pipeline failed at runtime: `get_destination_stream_from_pipeline` only recognized local `Stream` nodes, so a pipeline with only a `RemoteStream` destination returned "No destination streams found". That error path **deleted the scheduler trigger** while keeping the `backfill_jobs` config row, which then surfaced as: - `GET .../backfill/{job_id}` returning **404** (`DbError# key /backfill/ does not exist`) - the backfill row **disappearing from the UI** on the next refresh Remote destinations were never actually unsupported by execution — `ExecutablePipeline::process_batch` (already called by the backfill handler) routes remote records to the remote WAL. The pipeline was just being rejected by an over-strict pre-execution gate before it could run. ## Changes - `get_destination_stream_from_pipeline` now returns an empty list (not an error) for a remote-only pipeline, and only errors when there is no destination leaf of either kind. Remote records continue to be handled inside `process_batch`. - Added a defensive guard so `delete_before_backfill` is skipped (with a clear log) when a pipeline has no local destination streams — the edit-after-create case. Creation already rejects `delete_before_backfill` for remote pipelines with a user-facing message. - Added unit tests for remote-only (→ `Ok(empty)`) and mixed local+remote (→ locals only) pipelines. ## Testing - `cargo test -p openobserve get_destination_stream` → 8 passed. - ⚠️ Enterprise build not compiled locally (private `o2_enterprise` crate unavailable). The remote-WAL write path should be smoke-tested on an enterprise build with a real remote-only pipeline backfill before merge. ## Out of scope The pre-existing trigger-orphaning inconsistency (get 404s while list silently skips when a trigger is missing) is not addressed here. --- src/service/alerts/scheduler/handlers.rs | 146 +++++++++++++++++++++-- 1 file changed, 137 insertions(+), 9 deletions(-) diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 17e814955d..092ed34a2b 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -2809,8 +2809,21 @@ async fn handle_backfill_triggers( } }; - // 5. Handle deletion phase if required - if config.delete_before_backfill { + // 5. Handle deletion phase if required. + // + // Pre-deletion only applies to local destination streams; there is nothing to delete in a + // remote destination. Creation-time validation already rejects delete_before_backfill for + // remote pipelines, so an empty local-stream list here means the pipeline was edited to a + // remote-only destination after this backfill job was created. Skip deletion in that case and + // proceed straight to backfilling the remote destination. + let deletion_requested = config.delete_before_backfill && !destination_streams.is_empty(); + if config.delete_before_backfill && destination_streams.is_empty() { + log::warn!( + "[BACKFILL trace_id {trace_id}] [job_id: {}] delete_before_backfill is enabled but the pipeline has no local destination streams (remote-only); skipping deletion — data in remote destinations cannot be pre-deleted.", + job_id + ); + } + if deletion_requested { match &backfill_job.deletion_status { DeletionStatus::NotRequired => { // Not required, proceed to backfill @@ -3625,23 +3638,38 @@ async fn handle_backfill_triggers( Ok(()) } -/// Helper function to get destination streams from pipeline +/// Helper function to get the local destination streams from a pipeline. +/// +/// Returns only local `Stream` destinations. Remote (`RemoteStream`) destinations are routed to +/// the remote WAL inside `ExecutablePipeline::process_batch` and never need to be ingested here, +/// so they are intentionally excluded from the returned list. A remote-only pipeline therefore +/// returns an empty vec (not an error) — it is still a valid backfill target. An error is only +/// returned when the pipeline has no destination leaf of either kind. fn get_destination_stream_from_pipeline( pipeline: &config::meta::pipeline::Pipeline, ) -> Result, anyhow::Error> { let mut destination_streams = Vec::new(); + let mut has_remote_destination = false; for node in &pipeline.nodes { - if let NodeData::Stream(stream_params) = &node.data { - // Destination stream node (not the query source node) - let node_id = node.get_node_id(); - if !matches!(node_id.as_str(), "source" | "query") { - destination_streams.push(stream_params.clone()); + match &node.data { + NodeData::Stream(stream_params) => { + // Destination stream node (not the query source node) + let node_id = node.get_node_id(); + if !matches!(node_id.as_str(), "source" | "query") { + destination_streams.push(stream_params.clone()); + } } + NodeData::RemoteStream(_) => { + // Remote destinations are handled inside process_batch; just note their presence + // so a remote-only pipeline is not rejected as having "no destinations". + has_remote_destination = true; + } + _ => {} } } - if destination_streams.is_empty() { + if destination_streams.is_empty() && !has_remote_destination { Err(anyhow::anyhow!("No destination streams found in pipeline")) } else { Ok(destination_streams) @@ -4324,4 +4352,104 @@ mod tests { let result = get_destination_stream_from_pipeline(&pipeline); assert!(result.is_err()); } + + #[test] + fn test_get_destination_stream_remote_only_returns_empty_ok() { + use config::meta::{ + pipeline::{ + Pipeline, + components::{Node, PipelineSource}, + }, + stream::RemoteStreamParams, + }; + + let source_node = Node::new( + "source".to_string(), + NodeData::Stream(StreamParams::new("org", "input-stream", StreamType::Logs)), + 0.0, + 0.0, + "input".to_string(), + ); + let remote_node = Node::new( + "remote-1".to_string(), + NodeData::RemoteStream(RemoteStreamParams { + org_id: "org".into(), + destination_name: "my-remote".into(), + }), + 100.0, + 0.0, + "output".to_string(), + ); + let pipeline = Pipeline { + id: "p4".to_string(), + version: 0, + enabled: true, + org: "org".to_string(), + name: "test".to_string(), + description: String::new(), + source: PipelineSource::default(), + nodes: vec![source_node, remote_node], + edges: vec![], + kind: config::meta::pipeline::PipelineKind::User, + }; + + // Remote-only pipeline: no local streams to ingest, but it is a valid backfill target, + // so the function returns an empty list rather than an error. + let result = get_destination_stream_from_pipeline(&pipeline).unwrap(); + assert!(result.is_empty()); + } + + #[test] + fn test_get_destination_stream_mixed_returns_only_local() { + use config::meta::{ + pipeline::{ + Pipeline, + components::{Node, PipelineSource}, + }, + stream::RemoteStreamParams, + }; + + let source_node = Node::new( + "source".to_string(), + NodeData::Stream(StreamParams::new("org", "input-stream", StreamType::Logs)), + 0.0, + 0.0, + "input".to_string(), + ); + let local_node = Node::new( + "output-1".to_string(), + NodeData::Stream(StreamParams::new("org", "output-stream", StreamType::Logs)), + 100.0, + 0.0, + "output".to_string(), + ); + let remote_node = Node::new( + "remote-1".to_string(), + NodeData::RemoteStream(RemoteStreamParams { + org_id: "org".into(), + destination_name: "my-remote".into(), + }), + 200.0, + 0.0, + "output".to_string(), + ); + let pipeline = Pipeline { + id: "p5".to_string(), + version: 0, + enabled: true, + org: "org".to_string(), + name: "test".to_string(), + description: String::new(), + source: PipelineSource::default(), + nodes: vec![source_node, local_node, remote_node], + edges: vec![], + kind: config::meta::pipeline::PipelineKind::User, + }; + + // Mixed pipeline: only the local destination is returned; the remote destination is routed + // inside process_batch and excluded here. + let result = get_destination_stream_from_pipeline(&pipeline).unwrap(); + assert_eq!(result.len(), 1); + assert_eq!(result[0].stream_name.as_str(), "output-stream"); + } }