From 1ba1a91e4d7a76ef6a3ffc06f131ffa4b16dec7e Mon Sep 17 00:00:00 2001 From: Loakesh Indiran Date: Tue, 3 Nov 2026 17:55:55 +0530 Subject: [PATCH] fix(synthetics): alert on the merged vote of a confirm run - A run that held a confirm job alerts on its merged class, not on the worst raw job or the last ack's error source. - A confirm run that finalizes in the reaper or dispatcher resolves its alert there and notifies through a registered run notifier. - A confirmed failure below quorum is partial, even with a healthy majority. - The run row stores the merged run result. --- .../management/src/request/synthetics/mod.rs | 43 ++---- src/core/src/synthetics.rs | 59 ++++++++ src/infra/src/table/synthetics_runs.rs | 45 +++++- src/jobs/src/job/mod.rs | 6 + src/synthetics/src/dispatcher.rs | 5 +- src/synthetics/src/job_api.rs | 139 ++++++++++++++++-- src/synthetics/src/reaper/mod.rs | 6 +- 7 files changed, 249 insertions(+), 54 deletions(-) diff --git a/src/api/management/src/request/synthetics/mod.rs b/src/api/management/src/request/synthetics/mod.rs index 562e672678..5dee38ed82 100644 --- a/src/api/management/src/request/synthetics/mod.rs +++ b/src/api/management/src/request/synthetics/mod.rs @@ -1705,46 +1705,23 @@ async fn process_ack( // 3` alerted on the first failure and a 30-minute cooldown sent thirty // notifications. #[cfg(feature = "enterprise")] - use openobserve_synthetics::job_api::AlertDecision; - #[cfg(feature = "enterprise")] - let recovery = matches!(resp.alert, AlertDecision::Recovered); - #[cfg(feature = "enterprise")] - let flaky = matches!(resp.alert, AlertDecision::Flaky); - // A degrading target is `warning` on every run for as long as the condition - // lasts, so this one is throttled by transition upstream, not by cooldown. - #[cfg(feature = "enterprise")] - let degraded = matches!(resp.alert, AlertDecision::Degraded); - #[cfg(feature = "enterprise")] - let should_notify = - !matches!(resp.alert, AlertDecision::Silent) && resp.suppressed_by.is_none(); + let should_notify = !matches!( + resp.alert, + openobserve_synthetics::job_api::AlertDecision::Silent + ) && resp.suppressed_by.is_none(); // Enterprise-gated because alert *destinations* are: the dispatch it ends in // (`alerts::alert::dispatch_notification`) is not built in OSS. An OSS build // runs the check, records the run and serves the result — it just has // nowhere to send a notification, exactly as it has for alerts. #[cfg(feature = "enterprise")] if should_notify && !resp.destinations.is_empty() { - let notification = openobserve_core::synthetics::CheckNotification { - org_id: resp.org_id.clone(), - check_name: resp.synthetics_name.clone(), - check_id: resp.synthetics_id.clone(), - check_type: resp.synthetic_type.clone(), - target: resp.target.clone(), - destinations: resp.destinations.clone(), - run_id: resp.run_id.clone(), - status: resp.run_status.clone().unwrap_or_else(|| status.clone()), - job_count: resp.job_count as i64, - error: error.clone(), + let notification = openobserve_core::synthetics::CheckNotification::from_ack( + &resp, + &status, + error.clone(), + error_source.clone(), checked_at, - recovery, - consecutive_failures: resp.consecutive_failures, - flaky, - degraded, - status_reason: resp.status_reason.clone(), - error_source: error_source.clone(), - failing_locations: resp.failing_locations.clone(), - failing_environments: resp.failing_environments.clone(), - passing_locations: resp.passing_locations.clone(), - }; + ); tokio::spawn(async move { openobserve_core::synthetics::notify_check_result(notification).await; }); diff --git a/src/core/src/synthetics.rs b/src/core/src/synthetics.rs index 6d586bc8c7..b3b11029a3 100644 --- a/src/core/src/synthetics.rs +++ b/src/core/src/synthetics.rs @@ -76,6 +76,65 @@ pub struct CheckNotification { pub failing_environments: Vec, } +#[cfg(feature = "enterprise")] +impl CheckNotification { + /// The notification for a completed run. `status`, `error` and `error_source` describe the + /// ack that completed it, and `status` is the fallback when the run has no aggregate status. + pub fn from_ack( + resp: &openobserve_synthetics::job_api::AckResponse, + status: &str, + error: Option, + error_source: String, + checked_at: i64, + ) -> Self { + use openobserve_synthetics::job_api::AlertDecision; + Self { + org_id: resp.org_id.clone(), + check_name: resp.synthetics_name.clone(), + check_id: resp.synthetics_id.clone(), + check_type: resp.synthetic_type.clone(), + target: resp.target.clone(), + destinations: resp.destinations.clone(), + run_id: resp.run_id.clone(), + status: resp + .run_status + .clone() + .unwrap_or_else(|| status.to_string()), + job_count: resp.job_count as i64, + error, + checked_at, + recovery: matches!(resp.alert, AlertDecision::Recovered), + consecutive_failures: resp.consecutive_failures, + flaky: matches!(resp.alert, AlertDecision::Flaky), + // A degrading target is `warning` on every run for as long as the condition lasts, + // so this one is throttled by transition upstream, not by cooldown. + degraded: matches!(resp.alert, AlertDecision::Degraded), + status_reason: resp.status_reason.clone(), + error_source, + failing_locations: resp.failing_locations.clone(), + failing_environments: resp.failing_environments.clone(), + passing_locations: resp.passing_locations.clone(), + } + } +} + +/// Sends the notification of a run that finalized outside an ack, for +/// `openobserve_synthetics::job_api::register_run_notifier`. +#[cfg(feature = "enterprise")] +pub fn notify_finalized_run( + resp: openobserve_synthetics::job_api::AckResponse, +) -> openobserve_synthetics::alerting::BoxFuture<'static, ()> { + Box::pin(async move { + if resp.destinations.is_empty() { + return; + } + let checked_at = config::utils::time::now_micros(); + let status = resp.run_status.clone().unwrap_or_default(); + let n = CheckNotification::from_ack(&resp, &status, None, String::new(), checked_at); + notify_check_result(n).await; + }) +} + /// Fires once per run (when all jobs have completed) for non-passing runs. /// Passing runs are suppressed — operators want alerts, not confirmations. /// diff --git a/src/infra/src/table/synthetics_runs.rs b/src/infra/src/table/synthetics_runs.rs index b4dbab1bf7..4d10a66845 100644 --- a/src/infra/src/table/synthetics_runs.rs +++ b/src/infra/src/table/synthetics_runs.rs @@ -111,6 +111,8 @@ pub struct RunFinalized { pub run_class: RunClass, /// A confirm job passed, so a failed location counts as up (`flaky`). pub confirmed_flaky: bool, + /// The run held a confirm job, so its answer is the merged vote, not the worst raw result. + pub had_confirm: bool, /// The run earns one extra run: it is a scheduled run that measured nothing because the probe /// could not reach the job API or could not be dispatched. pub gap_retry: bool, @@ -139,6 +141,7 @@ pub struct Tally { /// the run has no confirm job, so the stored `run_result` stands. pub run_result: Option, pub confirmed_flaky: bool, + pub has_confirm: bool, /// Some job measured nothing because of [`GAP_RETRY_SOURCES`]. pub gap_cause: bool, } @@ -220,16 +223,17 @@ pub fn tally(job_count: i32, jobs: &[JobOutcome]) -> Tally { _ if failed + healthy == 0 && config > 0 => RunClass::Misconfigured, _ if failed + healthy == 0 => RunClass::NotMeasured, Verdict::Down => RunClass::Down, - Verdict::Up => RunClass::Up, - Verdict::NoVerdict if failed > 0 => RunClass::Partial, - // Only healthy locations tested the site, too few for a verdict. - Verdict::NoVerdict => RunClass::Up, + // A failure below quorum is partial, even when a healthy majority makes the verdict Up. + _ if failed > 0 => RunClass::Partial, + // Only healthy locations tested the site. + _ => RunClass::Up, }; Tally { verdict, class, run_result: (!confirms.is_empty() && worst > 0).then_some(worst), confirmed_flaky, + has_confirm: !confirms.is_empty(), gap_cause: jobs.iter().any(JobOutcome::is_gap_cause), } } @@ -518,6 +522,17 @@ async fn finalize( .or(run.run_result) .or(job_result) .unwrap_or(STATUS_ERROR); + if t.run_result + .is_some_and(|merged| Some(merged) != run.run_result) + { + // The run history shows the merged vote, as the check row does. + txn.execute(Statement::from_sql_and_values( + txn.get_database_backend(), + "UPDATE synthetics_runs SET run_result = $1 WHERE id = $2", + [Value::from(run_result), Value::from(run_id.to_owned())], + )) + .await?; + } let check_update = write_check_row( &txn, &run.synthetics_id, @@ -537,6 +552,7 @@ async fn finalize( verdict: t.verdict, run_class: t.class, confirmed_flaky: t.confirmed_flaky, + had_confirm: t.has_confirm, gap_retry: t.class == RunClass::NotMeasured && t.gap_cause && run.trigger_type == TRIGGER_SCHEDULE, @@ -1508,6 +1524,21 @@ mod tests { ], ); assert_eq!((two.verdict, two.class), (Verdict::Up, RunClass::Up)); + // A confirmed failure below quorum is partial, even with a healthy majority. + let regional = tally( + 4, + &[ + job("a", JOB_FAILED), + job("b", JOB_PASSED), + job("d", JOB_PASSED), + confirm("c", "a", JOB_FAILED), + ], + ); + assert_eq!( + (regional.verdict, regional.class), + (Verdict::Up, RunClass::Partial) + ); + assert!(regional.has_confirm); // No confirm job: the stored run result stands. assert_eq!(tally(1, &[job("a", JOB_FAILED)]).run_result, None); } @@ -1651,7 +1682,13 @@ mod tests { let done = finalize_run(&db, "r", 1, 300).await.unwrap().unwrap(); assert_eq!(done.run_class, RunClass::Up); assert!(done.confirmed_flaky); + assert!(done.had_confirm); assert_eq!(done.run_result, 2); + assert_eq!( + get_run(&db, "r").await.unwrap().unwrap().run_result, + Some(2), + "the run history shows the merged vote" + ); } #[tokio::test] diff --git a/src/jobs/src/job/mod.rs b/src/jobs/src/job/mod.rs index beeb3557f9..258d4b272d 100644 --- a/src/jobs/src/job/mod.rs +++ b/src/jobs/src/job/mod.rs @@ -942,6 +942,12 @@ pub async fn init() -> Result<(), anyhow::Error> { openobserve_core::synthetics::downtime_may_mute, ); } + // A run a confirm job kept open can finalize in the reaper or the dispatcher, where no ack + // sends its alert. + #[cfg(feature = "enterprise")] + openobserve_synthetics::job_api::register_run_notifier( + openobserve_core::synthetics::notify_finalized_run, + ); // Warm the cache on queriers (UI APIs) and on whichever node role is the configured // processing node so that get_coverage_deficit returns accurate data from startup // rather than always returning (0, 0) until files happen to be processed. diff --git a/src/synthetics/src/dispatcher.rs b/src/synthetics/src/dispatcher.rs index b1627c2692..2a1947b951 100644 --- a/src/synthetics/src/dispatcher.rs +++ b/src/synthetics/src/dispatcher.rs @@ -356,7 +356,10 @@ async fn mark_failure(row: &synthetics_jobs::LeasedRow, reason: Option<&str>) -> return false; } match synthetics_runs::finalize_run(db, &row.run_id, 4, now_us).await { - Ok(Some(done)) => crate::job_api::after_run_finalized(&done).await, + Ok(Some(done)) => { + crate::job_api::after_run_finalized(&done).await; + crate::job_api::alert_for_swept_run(&done).await; + } Ok(None) => {} Err(e) => { tracing::error!(synthetics_id = %row.synthetics_id, run_id = %row.run_id, job_id = %row.id, "[synthetics dispatcher] finalize_run: {e}"); diff --git a/src/synthetics/src/job_api.rs b/src/synthetics/src/job_api.rs index 3ade04caf9..01ad2b48ce 100644 --- a/src/synthetics/src/job_api.rs +++ b/src/synthetics/src/job_api.rs @@ -1680,6 +1680,102 @@ pub(crate) fn wants_confirm_job( && !synthetics_runs::is_not_measured_source(error_source, status_reason) } +/// Sends the notification for a run that finalized outside an ack; the server registers it, as +/// notification delivery lives above this crate. +pub type RunNotifier = fn(AckResponse) -> crate::alerting::BoxFuture<'static, ()>; + +static RUN_NOTIFIER: std::sync::OnceLock = std::sync::OnceLock::new(); + +pub fn register_run_notifier(f: RunNotifier) { + let _ = RUN_NOTIFIER.set(f); +} + +/// The alert inputs of a run that held a confirm job: `(status, status_reason, error_source)` +/// from the merged vote, so neither the worst raw job nor the last ack decides the alert. +pub(crate) fn merged_alert_input( + class: config::meta::synthetics::RunClass, + confirmed_flaky: bool, +) -> (&'static str, Option<&'static str>, &'static str) { + use config::meta::synthetics::RunClass as C; + match class { + C::Up if confirmed_flaky => ("warning", Some(crate::alerting::REASON_FLAKY), ""), + C::Up => ("passed", None, ""), + C::Partial | C::Down => ("failed", None, ""), + C::NotMeasured => ("error", None, crate::alerting::ERROR_SOURCE_CONTROL_PLANE), + C::Misconfigured => ("error", None, crate::alerting::ERROR_SOURCE_CONFIG), + C::Unknown => ("error", None, ""), + } +} + +/// Decides and sends the alert of a run that a confirm job kept open and that finalized outside +/// an ack: a dead letter, a failed dispatch or the stuck-run sweep. No ack runs the alert for it. +pub(crate) async fn alert_for_swept_run(done: &synthetics_runs::RunFinalized) { + if !done.had_confirm { + return; + } + let conn = get_orm_client_rw().await; + let Ok(Some(synthetic)) = + synthetics_checks::get_cached(conn, &done.org_id, &done.synthetics_id).await + else { + return; + }; + let (status, status_reason, error_source) = + merged_alert_input(done.run_class, done.confirmed_flaky); + let now_us = config::utils::time::now_micros(); + let (alert, consecutive_failures, suppressed_by) = resolve_alert( + conn, + &done.synthetics_id, + RunOutcome { + status: Some(status), + attempts: 1, + status_reason, + error_source, + }, + &synthetic, + now_us, + ) + .await; + if matches!(alert, AlertDecision::Silent) || suppressed_by.is_some() { + return; + } + let Some(notify) = RUN_NOTIFIER.get() else { + return; + }; + let outcomes = synthetics_jobs::run_location_outcomes(conn, &done.run_id) + .await + .unwrap_or_default(); + let failing_environments = environment_names(&done.org_id, &done.run_id).await; + let synthetic_type = serde_json::to_value(&synthetic.check_type) + .ok() + .and_then(|v| v.as_str().map(str::to_owned)) + .unwrap_or_default(); + notify(AckResponse { + run_complete: true, + run_status: Some(status.to_string()), + job_count: done.job_count, + org_id: done.org_id.clone(), + job_id: String::new(), + run_id: done.run_id.clone(), + synthetics_id: done.synthetics_id.clone(), + synthetics_name: synthetic.name, + synthetic_type, + target: synthetic.target, + destinations: synthetic.destinations, + location: String::new(), + pool: String::new(), + trigger_type: "scheduled".to_string(), + alert, + status_reason: status_reason.map(str::to_owned), + consecutive_failures, + failing_locations: outcomes.failing, + passing_locations: outcomes.passing, + usage_events: Vec::new(), + failing_environments, + suppressed_by: None, + }) + .await; +} + /// Delay before the one extra run that fills a Not measured gap. const GAP_RETRY_DELAY_US: i64 = 60_000_000; @@ -1870,7 +1966,12 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result (merged.map(|m| m.0.to_string()), done.job_count), Some(done) => { let status = match config::meta::synthetics::SyntheticStatus::from_db(done.run_result) { @@ -1935,26 +2036,16 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result anyhow::Result tracing::error!("[synthetics reaper] finalize_stuck_runs: {e}"), @@ -163,7 +164,10 @@ async fn handle_dead_letter( // Run accounting first: the ingest-token lookup below can return early, and an unfinalized // run never completes. 4 = Error on the SyntheticStatus scale, as the dispatcher reports. match synthetics_runs::finalize_run(db, &row.run_id, 4, now_us).await { - Ok(Some(done)) => crate::job_api::after_run_finalized(&done).await, + Ok(Some(done)) => { + crate::job_api::after_run_finalized(&done).await; + crate::job_api::alert_for_swept_run(&done).await; + } Ok(None) => {} Err(e) => { tracing::error!(