From e0bac669418d957f375f5b2c3ab4adb305c7659e Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 22 Jul 2026 14:50:36 +0800 Subject: [PATCH] fix(targets): unify runtime health snapshots (#5110) * fix(targets): unify runtime health snapshots * fix(targets): stabilize health snapshot merge * fix(admin): import runtime health test type --- crates/audit/src/pipeline.rs | 35 +- crates/notify/src/runtime_view.rs | 44 ++- crates/targets/src/lib.rs | 7 +- crates/targets/src/runtime/mod.rs | 225 +++++++++++-- crates/targets/src/target/mod.rs | 132 ++++++++ crates/targets/src/target/webhook.rs | 308 ++++++++++++++---- rustfs/src/admin/handlers/audit.rs | 35 +- rustfs/src/admin/handlers/event.rs | 56 +++- rustfs/src/admin/handlers/extensions.rs | 2 + .../src/admin/handlers/plugins_instances.rs | 80 ++++- .../src/admin/handlers/target_descriptor.rs | 122 ++++--- rustfs/src/admin/plugin_contract.rs | 10 + ...ract__tests__plugin_instance_contract.snap | 2 + ...ests__plugin_instance_detail_contract.snap | 2 + 14 files changed, 884 insertions(+), 176 deletions(-) diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index 6a83317c8..316dea93c 100644 --- a/crates/audit/src/pipeline.rs +++ b/crates/audit/src/pipeline.rs @@ -292,8 +292,8 @@ impl AuditPipeline { } pub async fn snapshot_target_health(&self) -> Vec { - let registry = self.registry.lock().await; - registry.runtime_manager().health_snapshots().await + let targets = self.registry.lock().await.list_target_values(); + rustfs_targets::health_snapshots_for_targets(targets).await } } @@ -570,7 +570,7 @@ mod tests { use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; use rustfs_targets::{StoreError, Target, TargetError}; use std::sync::Arc; - use tokio::sync::Mutex; + use tokio::sync::{Mutex, Notify}; /// Mock target whose `save()` outcome is fixed at construction so tests can /// force full-success / full-failure / partial-failure fan-outs. @@ -578,6 +578,7 @@ mod tests { struct MockTarget { id: TargetID, fail: bool, + health_gate: Option<(Arc, Arc)>, } impl MockTarget { @@ -585,8 +586,14 @@ mod tests { Self { id: TargetID::new(id.to_string(), "webhook".to_string()), fail, + health_gate: None, } } + + fn with_health_gate(mut self, started: Arc, release: Arc) -> Self { + self.health_gate = Some((started, release)); + self + } } #[async_trait] @@ -599,6 +606,10 @@ mod tests { } async fn is_active(&self) -> Result { + if let Some((started, release)) = &self.health_gate { + started.notify_one(); + release.notified().await; + } Ok(true) } @@ -673,6 +684,24 @@ mod tests { pipeline.dispatch(entry()).await.expect("no targets should return Ok"); } + #[tokio::test] + async fn health_probe_does_not_hold_the_registry_lock() { + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let pipeline = pipeline_with(vec![MockTarget::new("blocked", false).with_health_gate(started.clone(), release.clone())]); + let registry = Arc::clone(&pipeline.registry); + let snapshot_task = tokio::spawn(async move { pipeline.snapshot_target_health().await }); + started.notified().await; + + let guard = tokio::time::timeout(std::time::Duration::from_secs(1), registry.lock()) + .await + .expect("network health probe must not retain the audit registry lock"); + drop(guard); + release.notify_one(); + + assert_eq!(snapshot_task.await.expect("snapshot task should finish").len(), 1); + } + // backlog#962: dispatch_batch must mirror dispatch and propagate a // whole-batch loss instead of returning Ok. #[tokio::test] diff --git a/crates/notify/src/runtime_view.rs b/crates/notify/src/runtime_view.rs index e8530b1b3..b17df852e 100644 --- a/crates/notify/src/runtime_view.rs +++ b/crates/notify/src/runtime_view.rs @@ -61,7 +61,8 @@ impl NotifyRuntimeView { } pub async fn snapshot_target_health(&self) -> Vec { - self.target_list.read().await.runtime_health_snapshots().await + let targets = self.target_list.read().await.values(); + rustfs_targets::health_snapshots_for_targets(targets).await } pub async fn runtime_status_snapshot(&self) -> rustfs_targets::RuntimeStatusSnapshot { @@ -84,7 +85,7 @@ mod tests { Arc, atomic::{AtomicU64, Ordering}, }; - use tokio::sync::RwLock; + use tokio::sync::{Notify, RwLock}; #[derive(Clone)] struct TestTarget { @@ -93,6 +94,8 @@ mod tests { failed_messages: Arc, failed_store_length: u64, id: TargetID, + health_started: Option>, + health_release: Option>, total_messages: Arc, } @@ -104,6 +107,8 @@ mod tests { failed_messages: Arc::new(AtomicU64::new(0)), failed_store_length: 0, id: TargetID::new(id.to_string(), name.to_string()), + health_started: None, + health_release: None, total_messages: Arc::new(AtomicU64::new(0)), } } @@ -123,6 +128,12 @@ mod tests { self } + fn with_health_gate(mut self, started: Arc, release: Arc) -> Self { + self.health_started = Some(started); + self.health_release = Some(release); + self + } + fn record_successes(&self, count: u64) { self.total_messages.store(count, Ordering::Relaxed); } @@ -142,6 +153,10 @@ mod tests { } async fn is_active(&self) -> Result { + if let (Some(started), Some(release)) = (&self.health_started, &self.health_release) { + started.notify_one(); + release.notified().await; + } Ok(self.active) } @@ -268,4 +283,29 @@ mod tests { assert_eq!(status.target_count, 2); assert_eq!(status.replay_worker_count, 0); } + + #[tokio::test] + async fn health_probe_does_not_hold_the_target_list_read_lock() { + let target_list = Arc::new(RwLock::new(TargetList::new())); + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let target = Arc::new(TestTarget::new("blocked", "webhook").with_health_gate(started.clone(), release.clone())); + target_list + .write() + .await + .add(target as Arc + Send + Sync>) + .expect("test target should be added"); + let runtime_view = NotifyRuntimeView::new(target_list.clone(), Arc::new(RwLock::new(ReplayWorkerManager::new()))); + + let snapshot_task = tokio::spawn(async move { runtime_view.snapshot_target_health().await }); + started.notified().await; + + let write_guard = tokio::time::timeout(std::time::Duration::from_secs(1), target_list.write()) + .await + .expect("network health probe must not retain the target-list read lock"); + drop(write_guard); + release.notify_one(); + + assert_eq!(snapshot_task.await.expect("snapshot task should finish").len(), 1); + } } diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index e3a410b36..7b9c77ce7 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -67,10 +67,10 @@ pub use plugin::{ }; pub use runtime::{ OpenedActivation, PreparedActivation, ReplayEvent, ReplayWorkerManager, RuntimeActivation, RuntimeStatusSnapshot, - RuntimeTargetHealthSnapshot, RuntimeTargetHealthState, RuntimeTargetSnapshot, SharedTarget, TargetRuntimeManager, - activate_targets_with_replay, + RuntimeTargetHealthReason, RuntimeTargetHealthSnapshot, RuntimeTargetHealthState, RuntimeTargetSnapshot, SharedTarget, + TargetRuntimeManager, activate_targets_with_replay, adapter::{BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter}, - init_target_and_optionally_start_replay, + health_snapshots_for_targets, init_target_and_optionally_start_replay, ops_diagnostics::{ OpsDiagnosticsAccessDecision, OpsDiagnosticsReadRequest, OpsDiagnosticsRegistration, OpsDiagnosticsRegistry, OpsDiagnosticsRegistryError, @@ -87,6 +87,7 @@ pub use rustfs_s3_types::EventName; use serde::{Deserialize, Serialize}; pub use sys::user_agent::*; pub use target::{Target, TargetDeliverySnapshot}; +pub use target::{TargetHealth, TargetHealthReason, TargetHealthState}; /// Represents a log of events for sending to targets #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/targets/src/runtime/mod.rs b/crates/targets/src/runtime/mod.rs index bfad78dd1..3a235e069 100644 --- a/crates/targets/src/runtime/mod.rs +++ b/crates/targets/src/runtime/mod.rs @@ -26,6 +26,7 @@ use crate::plugin::PluginEvent; use crate::store::{Key, Store, ensure_store_entry_raw_readable}; use crate::target::QueuedPayload; use crate::target::TargetDeliverySnapshot; +use crate::target::{TargetHealth, TargetHealthReason, TargetHealthState}; use crate::{StoreError, TargetError}; use futures_util::stream::{FuturesUnordered, StreamExt}; use std::sync::Arc; @@ -78,6 +79,9 @@ pub(crate) fn inter_attempt_backoff_sum(attempts: usize) -> Duration { pub type SharedTarget = Arc + Send + Sync>; type ReplayHook = Arc) -> Pin + Send>> + Send + Sync>; +const HEALTH_COLLECTION_TIMEOUT: Duration = Duration::from_secs(5); +const HEALTH_PROBE_CONCURRENCY: usize = 10; + pub(crate) enum PrepareTargetResult where E: PluginEvent, @@ -269,19 +273,16 @@ pub struct RuntimeTargetSnapshot { pub total_messages: u64, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RuntimeTargetHealthState { - Disabled, - Error, - Offline, - Online, -} +pub type RuntimeTargetHealthState = TargetHealthState; +pub type RuntimeTargetHealthReason = TargetHealthReason; #[derive(Debug, Clone, PartialEq, Eq)] pub struct RuntimeTargetHealthSnapshot { + pub account_id: String, pub enabled: bool, pub error_message: Option, pub state: RuntimeTargetHealthState, + pub reason: RuntimeTargetHealthReason, pub target_id: String, pub target_type: String, } @@ -459,33 +460,57 @@ where } pub async fn health_snapshots(&self) -> Vec { - let mut snapshots = Vec::with_capacity(self.targets.len()); - for target in self.targets.values() { - let enabled = target.is_enabled(); - let target_id = target.id(); - let (state, error_message) = if !enabled { - (RuntimeTargetHealthState::Disabled, None) - } else { - match target.is_active().await { - Ok(true) => (RuntimeTargetHealthState::Online, None), - Ok(false) => (RuntimeTargetHealthState::Offline, None), - Err(err) => (RuntimeTargetHealthState::Error, Some(err.to_string())), - } - }; - - snapshots.push(RuntimeTargetHealthSnapshot { - enabled, - error_message, - state, - target_id: target_id.to_string(), - target_type: target_id.name, - }); - } - snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id)); - snapshots + health_snapshots_for_targets(self.values()).await } } +pub async fn health_snapshots_for_targets(targets: Vec>) -> Vec +where + E: PluginEvent, +{ + let deadline = tokio::time::Instant::now() + HEALTH_COLLECTION_TIMEOUT; + let permits = Arc::new(Semaphore::new(HEALTH_PROBE_CONCURRENCY)); + let mut probes = FuturesUnordered::new(); + for target in targets { + let permits = Arc::clone(&permits); + let enabled = target.is_enabled(); + let target_id = target.id(); + probes.push(async move { + let health = if !enabled { + TargetHealth::disabled() + } else if let Ok(Ok(_permit)) = tokio::time::timeout_at(deadline, permits.acquire_owned()).await { + if tokio::time::Instant::now() >= deadline { + TargetHealth::error(TargetHealthReason::HealthCheckFailed) + } else { + match tokio::time::timeout_at(deadline, target.health()).await { + Ok(health) => health, + Err(_) => TargetHealth::error(TargetHealthReason::TimedOut), + } + } + } else { + TargetHealth::error(TargetHealthReason::HealthCheckFailed) + }; + (enabled, target_id, health) + }); + } + + let mut snapshots = Vec::with_capacity(probes.len()); + while let Some((enabled, target_id, health)) = probes.next().await { + snapshots.push(RuntimeTargetHealthSnapshot { + account_id: target_id.id.clone(), + enabled, + error_message: (health.state == TargetHealthState::Error).then(|| health.reason.as_str().to_string()), + state: health.state, + reason: health.reason, + target_id: target_id.to_string(), + target_type: target_id.name, + }); + } + + snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id)); + snapshots +} + fn snapshot_from_delivery(target_id: TargetID, delivery: TargetDeliverySnapshot) -> RuntimeTargetSnapshot { RuntimeTargetSnapshot { failed_messages: delivery.failed_messages, @@ -934,16 +959,17 @@ where #[cfg(test)] mod tests { - use super::TargetRuntimeManager; + use super::{HEALTH_PROBE_CONCURRENCY, TargetRuntimeManager, health_snapshots_for_targets}; use crate::PluginEvent; use crate::StoreError; use crate::arn::TargetID; use crate::store::{Key, Store}; use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; - use crate::{Target, TargetError}; + use crate::{SharedTarget, Target, TargetError}; use async_trait::async_trait; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::time::Duration; use tokio::sync::{Notify, Semaphore}; #[tokio::test(start_paused = true)] @@ -1012,9 +1038,21 @@ mod tests { block_on_close: Arc, close_gate: Arc, close_calls: Arc, + enabled: bool, + health_delay: Duration, + health_drops: Arc, + health_started: Arc, close_started: Arc, } + struct HealthDropGuard(Arc); + + impl Drop for HealthDropGuard { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + impl TestTarget { fn new(id: &str, name: &str) -> Self { Self { @@ -1022,9 +1060,27 @@ mod tests { block_on_close: Arc::new(AtomicBool::new(false)), close_gate: Arc::new(Semaphore::new(0)), close_calls: Arc::new(AtomicUsize::new(0)), + enabled: true, + health_delay: Duration::ZERO, + health_drops: Arc::new(AtomicUsize::new(0)), + health_started: Arc::new(Notify::new()), close_started: Arc::new(Notify::new()), } } + + fn with_health_delay(id: &str, delay: Duration) -> Self { + Self { + health_delay: delay, + ..Self::new(id, "webhook") + } + } + + fn disabled(id: &str) -> Self { + Self { + enabled: false, + ..Self::new(id, "webhook") + } + } } #[async_trait] @@ -1037,6 +1093,9 @@ mod tests { } async fn is_active(&self) -> Result { + self.health_started.notify_one(); + let _drop_guard = HealthDropGuard(Arc::clone(&self.health_drops)); + tokio::time::sleep(self.health_delay).await; Ok(true) } @@ -1066,7 +1125,7 @@ mod tests { } fn is_enabled(&self) -> bool { - true + self.enabled } } @@ -1135,6 +1194,104 @@ mod tests { assert_eq!(snapshots[0].target_type, "webhook"); } + #[tokio::test(start_paused = true)] + async fn health_snapshot_allows_a_four_second_probe() { + let mut manager = TargetRuntimeManager::::new(); + manager.add_boxed(Box::new(TestTarget::with_health_delay("slow", Duration::from_secs(4)))); + + let snapshots = manager.health_snapshots().await; + + assert_eq!(snapshots.len(), 1); + assert_eq!(snapshots[0].state, crate::TargetHealthState::Online); + assert_eq!(snapshots[0].reason, crate::TargetHealthReason::Reachable); + } + + #[tokio::test(start_paused = true)] + async fn health_snapshot_times_out_after_five_seconds() { + let mut manager = TargetRuntimeManager::::new(); + manager.add_boxed(Box::new(TestTarget::with_health_delay("stalled", Duration::from_secs(6)))); + + let snapshots = manager.health_snapshots().await; + + assert_eq!(snapshots.len(), 1); + assert_eq!(snapshots[0].state, crate::TargetHealthState::Error); + assert_eq!(snapshots[0].reason, crate::TargetHealthReason::TimedOut); + } + + #[tokio::test(start_paused = true)] + async fn health_collection_deadline_does_not_scale_with_target_count() { + let mut manager = TargetRuntimeManager::::new(); + for index in 0..24 { + manager.add_boxed(Box::new(TestTarget::with_health_delay( + &format!("stalled-{index}"), + Duration::from_secs(30), + ))); + } + let started = tokio::time::Instant::now(); + + let snapshots = manager.health_snapshots().await; + + assert_eq!(started.elapsed(), Duration::from_secs(5)); + assert_eq!(snapshots.len(), 24); + assert!( + snapshots + .iter() + .all(|snapshot| snapshot.state == crate::TargetHealthState::Error) + ); + assert_eq!( + snapshots + .iter() + .filter(|snapshot| snapshot.reason == crate::TargetHealthReason::TimedOut) + .count(), + HEALTH_PROBE_CONCURRENCY + ); + assert_eq!( + snapshots + .iter() + .filter(|snapshot| snapshot.reason == crate::TargetHealthReason::HealthCheckFailed) + .count(), + 24 - HEALTH_PROBE_CONCURRENCY + ); + } + + #[tokio::test(start_paused = true)] + async fn disabled_target_does_not_wait_for_probe_capacity() { + let mut targets: Vec> = (0..HEALTH_PROBE_CONCURRENCY) + .map(|index| { + Arc::new(TestTarget::with_health_delay(&format!("stalled-{index}"), Duration::from_secs(30))) + as SharedTarget + }) + .collect(); + targets.push(Arc::new(TestTarget::disabled("disabled"))); + + let snapshots = health_snapshots_for_targets(targets).await; + let disabled = snapshots + .iter() + .find(|snapshot| snapshot.target_id == "disabled:webhook") + .expect("disabled target snapshot"); + + assert_eq!(disabled.state, crate::TargetHealthState::Disabled); + assert_eq!(disabled.reason, crate::TargetHealthReason::Disabled); + } + + #[tokio::test] + async fn cancelling_health_collection_drops_in_flight_probe() { + let target = TestTarget::with_health_delay("slow", Duration::from_secs(30)); + let health_drops = Arc::clone(&target.health_drops); + let health_started = Arc::clone(&target.health_started); + let targets: Vec> = vec![Arc::new(target)]; + let collector = tokio::spawn(health_snapshots_for_targets(targets)); + + tokio::time::timeout(Duration::from_secs(1), health_started.notified()) + .await + .expect("health probe should start"); + collector.abort(); + let join_error = collector.await.expect_err("health collector should be cancelled"); + + assert!(join_error.is_cancelled()); + assert_eq!(health_drops.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn sleep_or_cancelled_returns_immediately_on_cancel() { let (cancel_tx, mut cancel_rx) = tokio::sync::mpsc::channel::<()>(1); diff --git a/crates/targets/src/target/mod.rs b/crates/targets/src/target/mod.rs index 9e3e2abe1..def914ce6 100644 --- a/crates/targets/src/target/mod.rs +++ b/crates/targets/src/target/mod.rs @@ -72,6 +72,125 @@ pub struct TargetDeliveryCounters { total_messages: AtomicU64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TargetHealthState { + Disabled, + Error, + Offline, + Online, +} + +impl TargetHealthState { + pub const fn as_str(self) -> &'static str { + match self { + Self::Disabled => "disabled", + Self::Error => "error", + Self::Offline => "offline", + Self::Online => "online", + } + } + + pub const fn status(self) -> &'static str { + match self { + Self::Online => "online", + Self::Disabled | Self::Error | Self::Offline => "offline", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TargetHealthReason { + AuthenticationFailed, + ConfigurationInvalid, + ConnectionRefused, + Disabled, + DnsFailure, + HealthCheckFailed, + InitializationFailed, + NotLoadedInRuntime, + Reachable, + RequestFailed, + TimedOut, + TlsFailure, + Unreachable, +} + +impl TargetHealthReason { + pub const fn as_str(self) -> &'static str { + match self { + Self::AuthenticationFailed => "authentication_failed", + Self::ConfigurationInvalid => "configuration_invalid", + Self::ConnectionRefused => "connection_refused", + Self::Disabled => "disabled", + Self::DnsFailure => "dns_failure", + Self::HealthCheckFailed => "health_check_failed", + Self::InitializationFailed => "initialization_failed", + Self::NotLoadedInRuntime => "not_loaded_in_runtime", + Self::Reachable => "reachable", + Self::RequestFailed => "request_failed", + Self::TimedOut => "timed_out", + Self::TlsFailure => "tls_failure", + Self::Unreachable => "unreachable", + } + } + + fn from_target_error(err: &TargetError) -> Self { + match err { + TargetError::Authentication(_) => Self::AuthenticationFailed, + TargetError::Configuration(_) | TargetError::ParseError(_) => Self::ConfigurationInvalid, + TargetError::Initialization(_) | TargetError::ServerNotInitialized(_) => Self::InitializationFailed, + TargetError::Network(_) | TargetError::NotConnected => Self::Unreachable, + TargetError::Request(_) => Self::RequestFailed, + TargetError::Timeout(_) => Self::TimedOut, + TargetError::Storage(_) + | TargetError::JetStreamPublish { .. } + | TargetError::Encoding(_) + | TargetError::Serialization(_) + | TargetError::InvalidARN(_) + | TargetError::Unknown(_) + | TargetError::Disabled + | TargetError::Dropped(_) + | TargetError::SaveConfig(_) => Self::HealthCheckFailed, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TargetHealth { + pub state: TargetHealthState, + pub reason: TargetHealthReason, +} + +impl TargetHealth { + pub const fn disabled() -> Self { + Self { + state: TargetHealthState::Disabled, + reason: TargetHealthReason::Disabled, + } + } + + pub const fn error(reason: TargetHealthReason) -> Self { + Self { + state: TargetHealthState::Error, + reason, + } + } + + pub const fn offline(reason: TargetHealthReason) -> Self { + Self { + state: TargetHealthState::Offline, + reason, + } + } + + pub const fn online(reason: TargetHealthReason) -> Self { + Self { + state: TargetHealthState::Online, + reason, + } + } +} + pub(crate) type BoxedQueuedStore = Box + Send + Sync>; impl TargetDeliveryCounters { @@ -113,6 +232,19 @@ where /// Checks if the target is active and reachable async fn is_active(&self) -> Result; + /// Returns a credential-free, machine-readable health result. + async fn health(&self) -> TargetHealth { + if !self.is_enabled() { + return TargetHealth::disabled(); + } + + match self.is_active().await { + Ok(true) => TargetHealth::online(TargetHealthReason::Reachable), + Ok(false) => TargetHealth::offline(TargetHealthReason::Unreachable), + Err(err) => TargetHealth::error(TargetHealthReason::from_target_error(&err)), + } + } + /// Saves an event (either sends it immediately or stores it for later). /// /// A target whose [`Self::store`] returns `Some` must only persist the event diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index cfa64ba08..101860424 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -24,8 +24,8 @@ use crate::{ store::{Key, Store}, target::{ ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot, - TargetTlsState, TargetType, build_queued_payload, build_target_tls_fingerprint, open_target_queue_store, - persist_queued_payload_to_store, redacted_secret, + TargetHealth, TargetHealthReason, TargetHealthState, TargetTlsState, TargetType, build_queued_payload, + build_target_tls_fingerprint, open_target_queue_store, persist_queued_payload_to_store, redacted_secret, }, }; use async_trait::async_trait; @@ -34,6 +34,7 @@ use reqwest::{Client, StatusCode, Url}; use rustfs_tls_runtime::load_cert_bundle_der_bytes; use rustfs_utils::egress::validate_outbound_url; use std::{ + error::Error as StdError, fmt, marker::PhantomData, sync::{ @@ -49,6 +50,69 @@ const LOG_COMPONENT_TARGETS: &str = "targets"; const LOG_SUBSYSTEM_WEBHOOK: &str = "webhook"; const EVENT_WEBHOOK_TARGET_STATE: &str = "webhook_target_state"; const EVENT_WEBHOOK_DELIVERY_STATE: &str = "webhook_delivery_state"; +const WEBHOOK_HEALTH_TIMEOUT: Duration = Duration::from_secs(5); + +fn classify_probe_error(err: &reqwest::Error) -> TargetHealthReason { + if err.is_timeout() { + return TargetHealthReason::TimedOut; + } + + let mut source = err.source(); + while let Some(cause) = source { + if cause.downcast_ref::().is_some() { + return TargetHealthReason::TlsFailure; + } + if let Some(io_error) = cause.downcast_ref::() { + match io_error.kind() { + std::io::ErrorKind::ConnectionRefused => return TargetHealthReason::ConnectionRefused, + std::io::ErrorKind::NotFound | std::io::ErrorKind::AddrNotAvailable => { + return TargetHealthReason::DnsFailure; + } + _ => {} + } + } + + let label = cause.to_string().to_ascii_lowercase(); + if label.contains("dns error") || label.contains("failed to lookup") || label.contains("name or service not known") { + return TargetHealthReason::DnsFailure; + } + if label.contains("certificate") || label.contains("tls") { + return TargetHealthReason::TlsFailure; + } + source = cause.source(); + } + + TargetHealthReason::Unreachable +} + +async fn probe_health_url(client: &Client, health_check_url: &Url) -> TargetHealth { + match tokio::time::timeout(WEBHOOK_HEALTH_TIMEOUT, client.head(health_check_url.as_str()).send()).await { + Ok(Ok(_)) => TargetHealth::online(TargetHealthReason::Reachable), + Ok(Err(err)) => TargetHealth::error(classify_probe_error(&err)), + Err(_) => TargetHealth::error(TargetHealthReason::TimedOut), + } +} + +fn classify_delivery_status(status: StatusCode) -> Result<(), TargetError> { + if status.is_success() { + Ok(()) + } else if status.is_redirection() { + Err(TargetError::Request(format!( + "Webhook endpoint returned redirect '{}'; redirects are not followed for webhook delivery", + status + ))) + } else if status == StatusCode::FORBIDDEN || status == StatusCode::UNAUTHORIZED { + Err(TargetError::Authentication(format!( + "Webhook endpoint returned '{}', please check if your auth token is correctly set", + status + ))) + } else { + Err(TargetError::Request(format!( + "Webhook endpoint returned '{}', please check your endpoint configuration", + status + ))) + } +} /// Arguments for configuring a Webhook target #[derive(Clone)] @@ -79,7 +143,7 @@ impl fmt::Debug for WebhookArgs { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("WebhookArgs") .field("enable", &self.enable) - .field("endpoint", &self.endpoint) + .field("endpoint_origin", &self.endpoint.origin().ascii_serialization()) .field("auth_token", &redacted_secret(&self.auth_token)) .field("queue_dir", &self.queue_dir) .field("queue_limit", &self.queue_limit) @@ -243,7 +307,6 @@ where event = EVENT_WEBHOOK_TARGET_STATE, component = LOG_COMPONENT_TARGETS, subsystem = LOG_SUBSYSTEM_WEBHOOK, - endpoint = %args.endpoint, state = "tls_verification_skipped", fallback = "danger_accept_invalid_certs", "webhook target state" @@ -308,8 +371,14 @@ where fn health_check_url(endpoint: &Url) -> Result { endpoint .host() - .ok_or_else(|| TargetError::Configuration(format!("Webhook endpoint '{}' is missing a host", endpoint)))?; + .ok_or_else(|| TargetError::Configuration("Webhook endpoint is missing a host".to_string()))?; let mut health_check_url = endpoint.clone(); + health_check_url + .set_username("") + .map_err(|_| TargetError::Configuration("Webhook endpoint contains invalid user information".to_string()))?; + health_check_url + .set_password(None) + .map_err(|_| TargetError::Configuration("Webhook endpoint contains invalid user information".to_string()))?; health_check_url.set_path("/"); health_check_url.set_query(None); health_check_url.set_fragment(None); @@ -317,39 +386,34 @@ where Ok(health_check_url) } - async fn probe_reachability(&self) -> Result { + async fn probe_health(&self) -> TargetHealth { let Some(health_check_url) = self.health_check_url.as_ref() else { - return Ok(false); + return TargetHealth::offline(TargetHealthReason::Unreachable); }; - let client = self.http_client.lock().clone(); - match tokio::time::timeout(Duration::from_secs(5), client.head(health_check_url.as_str()).send()).await { - Ok(Ok(resp)) => { - debug!( - event = EVENT_WEBHOOK_TARGET_STATE, - component = LOG_COMPONENT_TARGETS, - subsystem = LOG_SUBSYSTEM_WEBHOOK, - target_id = %self.id, - status = %resp.status(), - health_check_url = %health_check_url, - state = "reachability_probe_succeeded", - "webhook target state" - ); - Ok(true) - } - Ok(Err(err)) if err.is_timeout() => Err(TargetError::Timeout(format!( - "Webhook health check request to {} timed out", - health_check_url - ))), - Ok(Err(err)) if err.is_connect() => Ok(false), - Ok(Err(err)) => Err(TargetError::Network(format!( - "Webhook health check request to {} failed: {}", - health_check_url, err - ))), - Err(_) => Err(TargetError::Timeout(format!( - "Webhook health check request to {} timed out", - health_check_url - ))), + let health = probe_health_url(&client, health_check_url).await; + if health.state == TargetHealthState::Online { + debug!( + event = EVENT_WEBHOOK_TARGET_STATE, + component = LOG_COMPONENT_TARGETS, + subsystem = LOG_SUBSYSTEM_WEBHOOK, + target_id = %self.id, + state = "reachability_probe_succeeded", + "webhook target state" + ); + } + health + } + + async fn probe_reachability(&self) -> Result { + let health = self.probe_health().await; + match health.state { + TargetHealthState::Online => Ok(true), + TargetHealthState::Offline | TargetHealthState::Disabled => Ok(false), + TargetHealthState::Error => match health.reason { + TargetHealthReason::TimedOut => Err(TargetError::Timeout("Webhook health check timed out".to_string())), + _ => Err(TargetError::Network(format!("Webhook health check failed: {}", health.reason.as_str()))), + }, } } @@ -371,7 +435,6 @@ where component = LOG_COMPONENT_TARGETS, subsystem = LOG_SUBSYSTEM_WEBHOOK, target_id = %self.id, - health_check_url = ?self.health_check_url, state = "reachable", "webhook target state" ); @@ -447,7 +510,7 @@ where if e.is_timeout() || e.is_connect() { TargetError::NotConnected } else { - TargetError::Request(format!("Failed to send request: {e}")) + TargetError::Request("Webhook delivery request failed".to_string()) } })?; @@ -456,7 +519,8 @@ where // pool and can be reused (keep-alive) instead of being closed mid-stream // (backlog#983). The body content is not needed for delivery accounting. let _ = resp.bytes().await; - if status.is_success() { + let result = classify_delivery_status(status); + if result.is_ok() { debug!( event = EVENT_WEBHOOK_DELIVERY_STATE, component = LOG_COMPONENT_TARGETS, @@ -467,26 +531,8 @@ where "webhook delivery state" ); self.delivery_counters.record_success(); - Ok(()) - } else if status.is_redirection() { - // SSRF hardening (backlog#974): redirects are intentionally not followed - // (see build_http_client). Treat a 3xx response as a delivery failure rather - // than silently chasing the redirect to a potentially internal target. - Err(TargetError::Request(format!( - "{} returned redirect '{}'; redirects are not followed for webhook delivery", - self.args.endpoint, status - ))) - } else if status == StatusCode::FORBIDDEN { - Err(TargetError::Authentication(format!( - "{} returned '{}', please check if your auth token is correctly set", - self.args.endpoint, status - ))) - } else { - Err(TargetError::Request(format!( - "{} returned '{}', please check your endpoint configuration", - self.args.endpoint, status - ))) } + result } } @@ -507,6 +553,13 @@ where self.probe_reachability().await } + async fn health(&self) -> TargetHealth { + if !self.args.enable { + return TargetHealth::disabled(); + } + self.probe_health().await + } + async fn save(&self, event: Arc>) -> Result<(), TargetError> { let queued = match self.build_queued_payload(&event) { Ok(queued) => queued, @@ -700,8 +753,9 @@ where #[cfg(test)] mod tests { - use super::{WebhookArgs, WebhookTarget}; - use crate::target::{REDACTED_SECRET, Target, TargetType, decode_object_name}; + use super::{WebhookArgs, WebhookTarget, classify_delivery_status, probe_health_url}; + use crate::target::{REDACTED_SECRET, Target, TargetHealthReason, TargetHealthState, TargetType, decode_object_name}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; use url::Url; use url::form_urlencoded; @@ -720,9 +774,25 @@ mod tests { } } + async fn http_status_url(status: u16) -> Url { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind test server"); + let address = listener.local_addr().expect("test server address"); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept health probe"); + let mut request = [0u8; 1024]; + let _ = stream.read(&mut request).await; + stream + .write_all(format!("HTTP/1.1 {status} Test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()) + .await + .expect("write health response"); + }); + Url::parse(&format!("http://{address}/")).expect("health probe URL") + } + #[test] fn debug_redacts_webhook_secret_fields() { let args = WebhookArgs { + endpoint: Url::parse("https://user:password@example.com/private?token=query-secret").expect("debug URL"), auth_token: "webhook-token".to_string(), client_key: "/etc/rustfs/webhook.key".to_string(), ..base_args() @@ -732,7 +802,11 @@ mod tests { assert!(!rendered.contains("webhook-token")); assert!(!rendered.contains("/etc/rustfs/webhook.key")); + assert!(!rendered.contains("password")); + assert!(!rendered.contains("/private")); + assert!(!rendered.contains("query-secret")); assert!(rendered.contains(REDACTED_SECRET)); + assert!(rendered.contains("https://example.com")); assert!(rendered.contains("WebhookArgs")); } @@ -824,12 +898,124 @@ mod tests { #[test] fn test_health_check_url_ignores_endpoint_path() { - let endpoint = Url::parse("https://example.com:9443/hook/path").unwrap(); - let health_check_url = WebhookTarget::::health_check_url(&endpoint).unwrap(); + let endpoint = Url::parse("https://user:password@example.com:9443/hook/path?token=secret").expect("webhook endpoint URL"); + let health_check_url = WebhookTarget::::health_check_url(&endpoint).expect("webhook health-check URL"); assert_eq!(health_check_url.as_str(), "https://example.com:9443/"); } + #[tokio::test] + async fn head_http_responses_only_measure_reachability() { + let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + for status in [401, 404, 500] { + let health = probe_health_url(&client, &http_status_url(status).await).await; + assert_eq!(health.state, TargetHealthState::Online, "HEAD {status} is reachable"); + assert_eq!(health.reason, TargetHealthReason::Reachable); + } + } + + #[tokio::test] + async fn refused_connection_has_stable_health_reason() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve refused port"); + let address = listener.local_addr().expect("refused address"); + drop(listener); + let url = Url::parse(&format!("http://{address}/")).expect("refused URL"); + let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + + let health = probe_health_url(&client, &url).await; + + assert_eq!(health.state, TargetHealthState::Error); + assert_eq!(health.reason, TargetHealthReason::ConnectionRefused); + } + + #[tokio::test] + async fn dns_failure_has_stable_health_reason() { + let url = Url::parse("http://rustfs-health-check.invalid/").expect("invalid test domain URL"); + let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + + let health = probe_health_url(&client, &url).await; + + assert_eq!(health.state, TargetHealthState::Error); + assert_eq!(health.reason, TargetHealthReason::DnsFailure); + } + + #[tokio::test(start_paused = true)] + async fn webhook_health_probe_has_a_five_second_total_budget() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind stalled server"); + let address = listener.local_addr().expect("stalled server address"); + let server = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.expect("accept stalled probe"); + std::future::pending::<()>().await; + }); + let url = Url::parse(&format!("http://{address}/")).expect("stalled URL"); + let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + let started = tokio::time::Instant::now(); + + let health = probe_health_url(&client, &url).await; + + assert_eq!(started.elapsed(), super::WEBHOOK_HEALTH_TIMEOUT); + assert_eq!(health.state, TargetHealthState::Error); + assert_eq!(health.reason, TargetHealthReason::TimedOut); + server.abort(); + } + + #[tokio::test] + async fn tls_failure_has_stable_health_reason() { + use rustls::{ + ServerConfig, ServerConnection, StreamOwned, + pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer}, + }; + use std::io::Read; + use std::sync::{Arc, Once}; + + static INSTALL_CRYPTO_PROVIDER: Once = Once::new(); + INSTALL_CRYPTO_PROVIDER.call_once(|| { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + }); + let rcgen::CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).expect("cert should generate"); + let server_config = Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der())), + ) + .expect("server cert should be valid"), + ); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind TLS server"); + let address = listener.local_addr().expect("TLS server address"); + let server = std::thread::spawn(move || { + let (stream, _) = listener.accept().expect("accept TLS client"); + let connection = ServerConnection::new(server_config).expect("server connection"); + let mut tls_stream = StreamOwned::new(connection, stream); + let mut request = [0u8; 1024]; + let _ = tls_stream.read(&mut request); + }); + let url = Url::parse(&format!("https://localhost:{}/", address.port())).expect("TLS health URL"); + let client = WebhookTarget::::build_http_client(&base_args()).expect("build client"); + + let health = probe_health_url(&client, &url).await; + + assert_eq!(health.state, TargetHealthState::Error); + assert_eq!(health.reason, TargetHealthReason::TlsFailure); + server.join().expect("TLS server thread"); + } + + #[test] + fn post_status_classification_requires_success() { + assert!(classify_delivery_status(reqwest::StatusCode::NO_CONTENT).is_ok()); + for status in [ + reqwest::StatusCode::MOVED_PERMANENTLY, + reqwest::StatusCode::UNAUTHORIZED, + reqwest::StatusCode::INTERNAL_SERVER_ERROR, + ] { + assert!(classify_delivery_status(status).is_err(), "POST {status} must fail"); + } + } + #[tokio::test] async fn test_disabled_target_can_be_constructed_without_origin_probe() { let args = WebhookArgs { diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index d6515a709..dd8bfa17f 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -16,8 +16,8 @@ use crate::admin::{ auth::validate_admin_request, handlers::audit_runtime_config::{load_server_config_from_store, update_audit_config_and_reload}, handlers::target_descriptor::{ - AdminTargetSpec, EndpointKey, TargetEndpointSource, admin_target_spec_from_builtin, build_enabled_target_kvs, - build_json_response, collect_runtime_statuses, extract_supported_target_params, + AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin, + build_enabled_target_kvs, build_json_response, collect_runtime_statuses, extract_supported_target_params, merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, @@ -192,6 +192,8 @@ struct AuditEndpoint { account_id: String, service: String, status: String, + health_state: String, + health_reason: String, source: TargetEndpointSource, } @@ -248,7 +250,10 @@ async fn audit_target_operation_block_reason(action: &str) -> Option { target_module_disabled_reason("audit", rustfs_config::ENV_AUDIT_ENABLE, is_audit_module_enabled(), action) } -fn merge_audit_endpoints(config: &Config, runtime_statuses: HashMap) -> S3Result> { +fn merge_audit_endpoints( + config: &Config, + runtime_statuses: HashMap, +) -> S3Result> { Ok( shared_merge_target_endpoints(audit_target_specs(), AUDIT_ROUTE_PREFIX, config, runtime_statuses)? .into_iter() @@ -256,6 +261,8 @@ fn merge_audit_endpoints(config: &Config, runtime_statuses: HashMap RuntimeHealthStatus { + RuntimeHealthStatus { + status: "online".to_string(), + state: "online".to_string(), + reason: "reachable".to_string(), + } + } + fn with_audit_webhook_target_env_cleared(target_name: &str, f: F) where F: FnOnce(), @@ -472,8 +487,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-target".to_string(), "webhook".to_string()), "online".to_string()), - (("env-only".to_string(), "webhook".to_string()), "online".to_string()), + (("mixed-target".to_string(), "webhook".to_string()), online_health()), + (("env-only".to_string(), "webhook".to_string()), online_health()), ]); let merged = merge_audit_endpoints(&config, runtime).expect("merge audit endpoints"); @@ -515,8 +530,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-kafka".to_string(), "kafka".to_string()), "online".to_string()), - (("env-kafka".to_string(), "kafka".to_string()), "online".to_string()), + (("mixed-kafka".to_string(), "kafka".to_string()), online_health()), + (("env-kafka".to_string(), "kafka".to_string()), online_health()), ]); let merged = merge_audit_endpoints(&config, runtime).expect("merge audit endpoints"); @@ -552,8 +567,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-amqp".to_string(), "amqp".to_string()), "online".to_string()), - (("env-amqp".to_string(), "amqp".to_string()), "online".to_string()), + (("mixed-amqp".to_string(), "amqp".to_string()), online_health()), + (("env-amqp".to_string(), "amqp".to_string()), online_health()), ]); let merged = merge_audit_endpoints(&config, runtime).expect("merge audit endpoints"); @@ -775,7 +790,7 @@ mod tests { ("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_PRIMARYCASE", Some("https://example.com/hook")), ], || { - let runtime = HashMap::from([(("PrimaryCase".to_string(), "webhook".to_string()), "online".to_string())]); + let runtime = HashMap::from([(("PrimaryCase".to_string(), "webhook".to_string()), online_health())]); let merged = merge_audit_endpoints(&config, runtime).expect("merge audit endpoints"); let mixed = merged .iter() diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 56ae22d8a..f4a1ba2d1 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -17,8 +17,8 @@ use crate::admin::{ handlers::notify_runtime_access::{get_notification_system, load_notification_config_snapshot}, handlers::supervise_admin_mutation, handlers::target_descriptor::{ - AdminTargetSpec, EndpointKey, TargetEndpointSource, admin_target_spec_from_builtin, build_enabled_target_kvs, - build_json_response, collect_runtime_statuses, extract_supported_target_params, + AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin, + build_enabled_target_kvs, build_json_response, collect_runtime_statuses, extract_supported_target_params, merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, @@ -240,6 +240,8 @@ struct NotificationEndpoint { account_id: String, service: String, status: String, + health_state: String, + health_reason: String, source: TargetEndpointSource, } @@ -301,7 +303,7 @@ async fn notification_target_operation_block_reason(action: &str) -> Option, + runtime_statuses: HashMap, ) -> S3Result> { Ok( shared_merge_target_endpoints(notification_target_specs(), NOTIFY_ROUTE_PREFIX, config, runtime_statuses)? @@ -310,6 +312,8 @@ fn merge_notification_endpoints( account_id: endpoint.account_id, service: endpoint.service, status: endpoint.status, + health_state: endpoint.health_state, + health_reason: endpoint.health_reason, source: endpoint.source, }) .collect(), @@ -463,7 +467,7 @@ impl Operation for ListTargetsArns { let target_statuses = collect_runtime_statuses(ns.get_target_values().await) .await .into_iter() - .map(|((account_id, service), status)| (rustfs_targets::arn::TargetID::new(account_id, service), status)) + .map(|((account_id, service), health)| (rustfs_targets::arn::TargetID::new(account_id, service), health.status)) .collect(); let data_target_arn_list = collect_online_target_arns(region.as_str(), target_statuses); @@ -548,6 +552,15 @@ mod tests { }]) } + fn runtime_health(status: &str) -> RuntimeHealthStatus { + let online = status == "online"; + RuntimeHealthStatus { + status: status.to_string(), + state: if online { "online" } else { "offline" }.to_string(), + reason: if online { "reachable" } else { "unreachable" }.to_string(), + } + } + #[test] fn notification_target_subsystem_resolves_admin_route_type() { assert_eq!( @@ -565,6 +578,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "online".to_string(), + health_state: "online".to_string(), + health_reason: "reachable".to_string(), source: TargetEndpointSource::Config, }], }; @@ -577,6 +592,8 @@ mod tests { "account_id": "primary", "service": "webhook", "status": "online", + "health_state": "online", + "health_reason": "reachable", "source": "config" }] }) @@ -596,7 +613,14 @@ mod tests { ); let config = Config(cfg_map); - let runtime = HashMap::from([(("webhook-a".to_string(), "webhook".to_string()), "online".to_string())]); + let runtime = HashMap::from([( + ("webhook-a".to_string(), "webhook".to_string()), + RuntimeHealthStatus { + status: "offline".to_string(), + state: "error".to_string(), + reason: "timed_out".to_string(), + }, + )]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); let mqtt = merged @@ -610,7 +634,9 @@ mod tests { .iter() .find(|entry| entry.account_id == "webhook-a" && entry.service == "webhook") .expect("webhook-a should be present"); - assert_eq!(webhook.status, "online"); + assert_eq!(webhook.status, "offline"); + assert_eq!(webhook.health_state, "error"); + assert_eq!(webhook.health_reason, "timed_out"); assert_eq!(webhook.source, TargetEndpointSource::Config); } @@ -623,8 +649,8 @@ mod tests { let config = Config(HashMap::from([(NOTIFY_WEBHOOK_SUB_SYS.to_string(), webhook_targets)])); let runtime = HashMap::from([ - (("webhook-enabled".to_string(), "webhook".to_string()), "online".to_string()), - (("env-only".to_string(), "mqtt".to_string()), "offline".to_string()), + (("webhook-enabled".to_string(), "webhook".to_string()), runtime_health("online")), + (("env-only".to_string(), "mqtt".to_string()), runtime_health("offline")), ]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); @@ -665,8 +691,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-target".to_string(), "webhook".to_string()), "online".to_string()), - (("env-only".to_string(), "webhook".to_string()), "online".to_string()), + (("mixed-target".to_string(), "webhook".to_string()), runtime_health("online")), + (("env-only".to_string(), "webhook".to_string()), runtime_health("online")), ]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); @@ -708,8 +734,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-kafka".to_string(), "kafka".to_string()), "online".to_string()), - (("env-kafka".to_string(), "kafka".to_string()), "online".to_string()), + (("mixed-kafka".to_string(), "kafka".to_string()), runtime_health("online")), + (("env-kafka".to_string(), "kafka".to_string()), runtime_health("online")), ]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); @@ -745,8 +771,8 @@ mod tests { ], || { let runtime = HashMap::from([ - (("mixed-amqp".to_string(), "amqp".to_string()), "online".to_string()), - (("env-amqp".to_string(), "amqp".to_string()), "online".to_string()), + (("mixed-amqp".to_string(), "amqp".to_string()), runtime_health("online")), + (("env-amqp".to_string(), "amqp".to_string()), runtime_health("online")), ]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); @@ -922,7 +948,7 @@ mod tests { ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARYCASE", Some("https://example.com/hook")), ], || { - let runtime = HashMap::from([(("PrimaryCase".to_string(), "webhook".to_string()), "online".to_string())]); + let runtime = HashMap::from([(("PrimaryCase".to_string(), "webhook".to_string()), runtime_health("online"))]); let merged = merge_notification_endpoints(&config, runtime).expect("merge notification endpoints"); let mixed = merged .iter() diff --git a/rustfs/src/admin/handlers/extensions.rs b/rustfs/src/admin/handlers/extensions.rs index 82f66ac34..bc7ec9dc5 100644 --- a/rustfs/src/admin/handlers/extensions.rs +++ b/rustfs/src/admin/handlers/extensions.rs @@ -486,6 +486,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), source: PluginInstanceSource::Config, enabled: true, config: HashMap::from([("endpoint".to_string(), "https://example.test/webhook".to_string())]), diff --git a/rustfs/src/admin/handlers/plugins_instances.rs b/rustfs/src/admin/handlers/plugins_instances.rs index 70e634a27..f66f5b693 100644 --- a/rustfs/src/admin/handlers/plugins_instances.rs +++ b/rustfs/src/admin/handlers/plugins_instances.rs @@ -19,8 +19,8 @@ use crate::admin::{ load_notification_config_snapshot, remove_notification_target_config, set_notification_target_config, }, handlers::target_descriptor::{ - AdminTargetSpec, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin, build_enabled_target_kvs, - build_json_response, collect_runtime_statuses, collect_target_instances, find_target_instance, + AdminTargetSpec, RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin, + build_enabled_target_kvs, build_json_response, collect_runtime_statuses, collect_target_instances, find_target_instance, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, plugin_contract::{ @@ -205,6 +205,8 @@ fn map_instance(instance: TargetInstanceReadModel) -> PluginInstanceEntry { account_id: instance.account_id, service: instance.service, status: instance.status, + health_state: instance.health_state, + health_reason: instance.health_reason, source: map_instance_source(instance.source), enabled: instance.enabled, config, @@ -639,7 +641,9 @@ async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainCon module_disabled_block_reason(context.domain, action) } -async fn plugin_instance_runtime_statuses(context: PluginInstanceDomainContext) -> S3Result> { +async fn plugin_instance_runtime_statuses( + context: PluginInstanceDomainContext, +) -> S3Result> { match context.domain { PluginContractDomain::Notify => { let (ns, _) = load_notification_config_snapshot().await?; @@ -865,7 +869,8 @@ mod tests { parse_plugin_instance_id, parse_plugin_instance_source, resolve_plugin_instance_target, }; use crate::admin::handlers::target_descriptor::{ - TargetEndpointSource, TargetInstanceReadModel, canonical_target_instance_id, collect_target_instances, + RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, canonical_target_instance_id, + collect_target_instances, }; use crate::admin::plugin_contract::{ PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, @@ -967,9 +972,46 @@ mod tests { .expect("configured instance should be present"); assert_eq!(primary.status, "offline"); + assert_eq!(primary.health_state, "offline"); + assert_eq!(primary.health_reason, "not_loaded_in_runtime"); assert_eq!(primary.source, TargetEndpointSource::Config); } + #[test] + fn disabled_instance_without_runtime_reports_disabled_health() { + let config = Config(HashMap::from([( + NOTIFY_WEBHOOK_SUB_SYS.to_string(), + HashMap::from([( + "primary".to_string(), + KVS(vec![ + KV { + key: ENABLE_KEY.to_string(), + value: "off".to_string(), + hidden_if_empty: false, + }, + KV { + key: WEBHOOK_ENDPOINT.to_string(), + value: "https://example.com/webhook".to_string(), + hidden_if_empty: false, + }, + ]), + )]), + )])); + + let instances = + collect_target_instances(super::notification_target_specs(), NOTIFY_ROUTE_PREFIX, &config, HashMap::new()) + .expect("collect target instances"); + let primary = instances + .into_iter() + .find(|instance| instance.account_id == "primary" && instance.service == "webhook") + .expect("disabled instance should be present"); + + assert_eq!(primary.status, "offline"); + assert_eq!(primary.health_state, "disabled"); + assert_eq!(primary.health_reason, "disabled"); + assert!(!primary.runtime_present); + } + #[test] fn env_only_instance_appears_with_env_source() { temp_env::with_vars( @@ -998,7 +1040,14 @@ mod tests { #[test] fn runtime_only_instance_appears_with_runtime_source() { - let runtime_statuses = HashMap::from([(("runtime-only".to_string(), "webhook".to_string()), "online".to_string())]); + let runtime_statuses = HashMap::from([( + ("runtime-only".to_string(), "webhook".to_string()), + RuntimeHealthStatus { + status: "online".to_string(), + state: "online".to_string(), + reason: "reachable".to_string(), + }, + )]); let instances = collect_target_instances( super::notification_target_specs(), NOTIFY_ROUTE_PREFIX, @@ -1014,6 +1063,8 @@ mod tests { assert_eq!(runtime_only.source, TargetEndpointSource::Runtime); assert_eq!(runtime_only.status, "online"); + assert_eq!(runtime_only.health_state, "online"); + assert_eq!(runtime_only.health_reason, "reachable"); assert_eq!(runtime_only.plugin_id, "builtin:webhook"); assert_eq!(runtime_only.subsystem, NOTIFY_WEBHOOK_SUB_SYS); } @@ -1028,6 +1079,8 @@ mod tests { account_id: "Primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), runtime_present: false, source: TargetEndpointSource::Config, enabled: true, @@ -1050,6 +1103,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "online".to_string(), + health_state: "online".to_string(), + health_reason: "reachable".to_string(), runtime_present: true, source: TargetEndpointSource::Config, enabled: true, @@ -1481,6 +1536,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), runtime_present: false, source: TargetEndpointSource::Config, enabled: true, @@ -1505,6 +1562,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "unreachable".to_string(), runtime_present: true, source: TargetEndpointSource::Runtime, enabled: true, @@ -1529,6 +1588,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "disabled".to_string(), + health_reason: "disabled".to_string(), runtime_present: false, source: TargetEndpointSource::Mixed, enabled: false, @@ -1570,6 +1631,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), runtime_present: false, source: TargetEndpointSource::Config, enabled: true, @@ -1683,6 +1746,13 @@ mod tests { account_id: input.account_id.to_string(), service: input.service.to_string(), status: input.status.to_string(), + health_state: input.status.to_string(), + health_reason: if input.status == "online" { + "reachable" + } else { + "unreachable" + } + .to_string(), source: input.source, enabled: input.enabled, config: HashMap::new(), diff --git a/rustfs/src/admin/handlers/target_descriptor.rs b/rustfs/src/admin/handlers/target_descriptor.rs index 54a327edd..cc76fb058 100644 --- a/rustfs/src/admin/handlers/target_descriptor.rs +++ b/rustfs/src/admin/handlers/target_descriptor.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use futures::StreamExt; use futures::future::BoxFuture; use hashbrown::HashSet as HbHashSet; use http::{HeaderMap, HeaderValue, StatusCode}; @@ -22,12 +21,11 @@ use rustfs_config::{ MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, MYSQL_QUEUE_DIR, POSTGRES_QUEUE_DIR, REDIS_QUEUE_DIR, }; -use rustfs_targets::SharedTarget; use rustfs_targets::{ - BuiltinTargetAdminDescriptor, TargetAdminMetadata, TargetDomain, TargetError, TargetRequestValidator, - check_amqp_broker_available, check_kafka_broker_available, check_mqtt_broker_available_with_tls, - check_mysql_server_available, check_nats_server_available, check_postgres_server_available, check_pulsar_broker_available, - check_redis_server_available, + BuiltinTargetAdminDescriptor, SharedTarget, TargetAdminMetadata, TargetDomain, TargetError, TargetHealthReason, + TargetHealthState, TargetRequestValidator, check_amqp_broker_available, check_kafka_broker_available, + check_mqtt_broker_available_with_tls, check_mysql_server_available, check_nats_server_available, + check_postgres_server_available, check_pulsar_broker_available, check_redis_server_available, config::{ TargetPluginInstanceCompatDescriptor, TargetPluginInstanceRecord, build_amqp_args, build_kafka_args, build_mysql_args, build_nats_args, build_postgres_args, build_pulsar_args, build_redis_args, try_normalize_target_plugin_instances, @@ -42,11 +40,36 @@ use std::collections::{HashMap, HashSet}; use std::io::{Error, ErrorKind}; use std::path::Path; use std::sync::Arc; -use tokio::sync::Semaphore; use tokio::time::{Duration, sleep, timeout}; use url::Url; pub(crate) type EndpointKey = (String, String); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct RuntimeHealthStatus { + pub status: String, + pub state: String, + pub reason: String, +} + +impl RuntimeHealthStatus { + fn disabled() -> Self { + Self { + status: TargetHealthState::Disabled.status().to_string(), + state: TargetHealthState::Disabled.as_str().to_string(), + reason: TargetHealthReason::Disabled.as_str().to_string(), + } + } + + fn not_loaded() -> Self { + Self { + status: TargetHealthState::Offline.status().to_string(), + state: TargetHealthState::Offline.as_str().to_string(), + reason: TargetHealthReason::NotLoadedInRuntime.as_str().to_string(), + } + } +} + type AdminRequestValidatorFn = Arc Fn(&'a HashMap, &'a str) -> BoxFuture<'a, S3Result<()>> + Send + Sync>; type DomainScopedValidatorFn = for<'a> fn(&'a HashMap, &'a str, TargetDomain) -> BoxFuture<'a, S3Result<()>>; @@ -64,6 +87,8 @@ pub(crate) struct MergedTargetEndpoint { pub account_id: String, pub service: String, pub status: String, + pub health_state: String, + pub health_reason: String, pub source: TargetEndpointSource, } @@ -76,6 +101,8 @@ pub(crate) struct TargetInstanceReadModel { pub account_id: String, pub service: String, pub status: String, + pub health_state: String, + pub health_reason: String, pub runtime_present: bool, pub source: TargetEndpointSource, pub enabled: bool, @@ -272,43 +299,36 @@ pub(crate) fn build_json_response( S3Response::with_headers((status, body), header) } -pub(crate) async fn collect_runtime_statuses(targets: Vec>) -> HashMap +pub(crate) async fn collect_runtime_statuses(targets: Vec>) -> HashMap where - E: Send + Sync + 'static + Clone + serde::Serialize + serde::de::DeserializeOwned, + E: rustfs_targets::PluginEvent, { - let semaphore = Arc::new(Semaphore::new(10)); - let mut futures = futures::stream::FuturesUnordered::new(); - - for target in targets { - let sem = Arc::clone(&semaphore); - futures.push(async move { - let _permit = sem.acquire().await; - let status = match tokio::time::timeout(Duration::from_secs(3), target.is_active()).await { - Ok(Ok(true)) => "online", - _ => "offline", - }; - ((target.id().id, target.id().name), status.to_string()) - }); - } - - let mut runtime_statuses = HashMap::new(); - while let Some((key, status)) = futures.next().await { - runtime_statuses.insert(key, status); - } - - runtime_statuses + rustfs_targets::health_snapshots_for_targets(targets) + .await + .into_iter() + .map(|snapshot| { + ( + (snapshot.account_id, snapshot.target_type), + RuntimeHealthStatus { + status: snapshot.state.status().to_string(), + state: snapshot.state.as_str().to_string(), + reason: snapshot.reason.as_str().to_string(), + }, + ) + }) + .collect() } pub(crate) fn merge_target_endpoints( specs: &[AdminTargetSpec], route_prefix: &str, config: &Config, - runtime_statuses: HashMap, + runtime_statuses: HashMap, ) -> S3Result> { let mut endpoints = Vec::new(); let mut seen = HashSet::new(); let snapshot = collect_endpoint_snapshot(specs, route_prefix, config)?; - let mut normalized_runtime_statuses: HashMap = HashMap::new(); + let mut normalized_runtime_statuses: HashMap = HashMap::new(); for ((account_id, service), status) in runtime_statuses { let normalized = normalized_endpoint_key(&account_id, &service); @@ -323,15 +343,17 @@ pub(crate) fn merge_target_endpoints( continue; } - let status = normalized_runtime_statuses + let health = normalized_runtime_statuses .remove(&normalized) .map(|(_, _, status)| status) - .unwrap_or_else(|| "offline".to_string()); + .unwrap_or_else(RuntimeHealthStatus::not_loaded); endpoints.push(MergedTargetEndpoint { account_id: key.0, service: key.1, - status, + status: health.status, + health_state: health.state, + health_reason: health.reason, source: classify_endpoint_source(&snapshot.config_targets, &snapshot.env_targets, &normalized), }); } @@ -341,7 +363,9 @@ pub(crate) fn merge_target_endpoints( endpoints.push(MergedTargetEndpoint { account_id, service, - status, + status: status.status, + health_state: status.state, + health_reason: status.reason, source: classify_endpoint_source(&snapshot.config_targets, &snapshot.env_targets, &normalized), }); } @@ -356,6 +380,8 @@ pub(crate) fn merge_target_endpoints( account_id: key.0.clone(), service: key.1.clone(), status: "offline".to_string(), + health_state: TargetHealthState::Offline.as_str().to_string(), + health_reason: TargetHealthReason::NotLoadedInRuntime.as_str().to_string(), source: classify_endpoint_source(&snapshot.config_targets, &snapshot.env_targets, key), }); } @@ -372,11 +398,11 @@ pub(crate) fn collect_target_instances( specs: &[AdminTargetSpec], route_prefix: &str, config: &Config, - runtime_statuses: HashMap, + runtime_statuses: HashMap, ) -> S3Result> { let mut instances = Vec::new(); let mut seen = HashSet::new(); - let mut normalized_runtime_statuses: HashMap = HashMap::new(); + let mut normalized_runtime_statuses: HashMap = HashMap::new(); let domain = inferred_target_domain(route_prefix); let snapshot = collect_endpoint_snapshot(specs, route_prefix, config)?; @@ -394,10 +420,16 @@ pub(crate) fn collect_target_instances( } let runtime_present = normalized_runtime_statuses.contains_key(&key); - let status = normalized_runtime_statuses + let health = normalized_runtime_statuses .remove(&key) .map(|(_, _, status)| status) - .unwrap_or_else(|| "offline".to_string()); + .unwrap_or_else(|| { + if instance.enabled { + RuntimeHealthStatus::not_loaded() + } else { + RuntimeHealthStatus::disabled() + } + }); let source = classify_endpoint_source_flags(instance_has_config_entry(&instance), instance_has_env_entry(&instance)); instances.push(TargetInstanceReadModel { @@ -407,7 +439,9 @@ pub(crate) fn collect_target_instances( subsystem: instance.subsystem, account_id: instance.instance_id, service: instance.target_type, - status, + status: health.status, + health_state: health.state, + health_reason: health.reason, runtime_present, source, enabled: instance.enabled, @@ -430,7 +464,9 @@ pub(crate) fn collect_target_instances( subsystem, account_id, service, - status, + status: status.status, + health_state: status.state, + health_reason: status.reason, runtime_present: true, source: TargetEndpointSource::Runtime, enabled: true, @@ -446,7 +482,7 @@ pub(crate) fn find_target_instance( specs: &[AdminTargetSpec], route_prefix: &str, config: &Config, - runtime_statuses: HashMap, + runtime_statuses: HashMap, canonical_id: &str, ) -> S3Result> { Ok(collect_target_instances(specs, route_prefix, config, runtime_statuses)? diff --git a/rustfs/src/admin/plugin_contract.rs b/rustfs/src/admin/plugin_contract.rs index 880f167fb..43c6d0199 100644 --- a/rustfs/src/admin/plugin_contract.rs +++ b/rustfs/src/admin/plugin_contract.rs @@ -323,6 +323,8 @@ pub(crate) struct PluginInstanceEntry { pub account_id: String, pub service: String, pub status: String, + pub health_state: String, + pub health_reason: String, pub source: PluginInstanceSource, pub enabled: bool, pub config: HashMap, @@ -482,6 +484,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), source: PluginInstanceSource::Config, enabled: true, config: HashMap::from([ @@ -512,6 +516,8 @@ mod tests { "account_id": "primary", "service": "webhook", "status": "offline", + "health_state": "offline", + "health_reason": "not_loaded_in_runtime", "source": "config", "enabled": true, "config": { @@ -541,6 +547,8 @@ mod tests { account_id: "primary".to_string(), service: "webhook".to_string(), status: "offline".to_string(), + health_state: "offline".to_string(), + health_reason: "not_loaded_in_runtime".to_string(), source: PluginInstanceSource::Config, enabled: true, config: HashMap::from([("endpoint".to_string(), "https://example.com/hook".to_string())]), @@ -565,6 +573,8 @@ mod tests { "account_id": "primary", "service": "webhook", "status": "offline", + "health_state": "offline", + "health_reason": "not_loaded_in_runtime", "source": "config", "enabled": true, "config": { diff --git a/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_contract.snap b/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_contract.snap index a7990e03e..60b2f53d8 100644 --- a/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_contract.snap +++ b/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_contract.snap @@ -21,6 +21,8 @@ expression: value ], "domain": "notify", "enabled": true, + "health_reason": "not_loaded_in_runtime", + "health_state": "offline", "id": "builtin:webhook:notify:primary", "plugin_id": "builtin:webhook", "service": "webhook", diff --git a/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_detail_contract.snap b/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_detail_contract.snap index 0b66abf80..9edf511f7 100644 --- a/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_detail_contract.snap +++ b/rustfs/src/admin/snapshots/rustfs__admin__plugin_contract__tests__plugin_instance_detail_contract.snap @@ -18,6 +18,8 @@ expression: value ], "domain": "notify", "enabled": true, + "health_reason": "not_loaded_in_runtime", + "health_state": "offline", "id": "builtin:webhook:notify:primary", "plugin_id": "builtin:webhook", "service": "webhook",