diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index bdd35c2cb..c41165726 100644 --- a/crates/obs/src/lib.rs +++ b/crates/obs/src/lib.rs @@ -72,7 +72,12 @@ pub use error::*; pub use global::*; pub use logging::*; pub use metrics::schema::*; -pub use metrics::{init_metrics_collectors, init_metrics_runtime}; +pub use metrics::{ + MetricsRuntimeCancellationSource, MetricsRuntimeController, MetricsRuntimeControllerSnapshot, MetricsRuntimeDesiredSnapshot, + MetricsRuntimeDesiredState, MetricsRuntimeIntervalsSnapshot, MetricsRuntimeReconcilePlan, MetricsRuntimeServiceState, + MetricsRuntimeShutdownHandle, MetricsRuntimeStatusSnapshot, MetricsRuntimeWorkerMutation, init_metrics_collectors, + init_metrics_runtime, metrics_runtime_controller_snapshot, metrics_runtime_status_snapshot, +}; pub use telemetry::{OtelGuard, Recorder}; // Dial9 Tokio runtime telemetry diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index 4b1a725d1..592c1a42d 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -22,4 +22,9 @@ pub mod stats_collector; pub use collectors::*; pub use config::*; pub use report::{PrometheusMetric, report_metrics}; -pub use scheduler::{init_metrics_collectors, init_metrics_runtime}; +pub use scheduler::{ + MetricsRuntimeCancellationSource, MetricsRuntimeController, MetricsRuntimeControllerSnapshot, MetricsRuntimeDesiredSnapshot, + MetricsRuntimeDesiredState, MetricsRuntimeIntervalsSnapshot, MetricsRuntimeReconcilePlan, MetricsRuntimeServiceState, + MetricsRuntimeShutdownHandle, MetricsRuntimeStatusSnapshot, MetricsRuntimeWorkerMutation, init_metrics_collectors, + init_metrics_runtime, metrics_runtime_controller_snapshot, metrics_runtime_status_snapshot, +}; diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index e02ab8fda..507c43485 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -85,6 +85,7 @@ use rustfs_audit::audit_target_metrics; use rustfs_ecstore::global::get_global_bucket_monitor; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; +use serde::Serialize; use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::time::Duration; @@ -99,14 +100,286 @@ const DEFAULT_SYSTEM_METRICS_INTERVAL: Duration = Duration::from_secs(15); const ENV_SYSTEM_METRICS_INTERVAL: &str = "RUSTFS_METRICS_SYSTEM_INTERVAL_SEC"; /// Legacy environment variable for system monitoring interval const LEGACY_SYSTEM_METRICS_INTERVAL: &str = "RUSTFS_OBS_METRICS_SYSTEM_INTERVAL_MS"; +const LEGACY_CLUSTER_INTERVAL: &str = "RUSTFS_METRICS_CLUSTER_INTERVAL"; +const LEGACY_BUCKET_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_INTERVAL"; +const LEGACY_NODE_INTERVAL: &str = "RUSTFS_METRICS_NODE_INTERVAL"; +const LEGACY_REPLICATION_BANDWIDTH_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL"; +const LEGACY_RESOURCE_INTERVAL: &str = "RUSTFS_METRICS_RESOURCE_INTERVAL"; +const LEGACY_AUDIT_INTERVAL: &str = "RUSTFS_METRICS_AUDIT_INTERVAL"; +const LEGACY_NOTIFICATION_INTERVAL: &str = "RUSTFS_METRICS_NOTIFICATION_INTERVAL"; +const LEGACY_DEFAULT_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL"; /// Default cycles to emit zero for removed replication bandwidth series before letting them expire. const DEFAULT_REPL_BW_ZERO_TOMBSTONE_CYCLES: u8 = 3; /// Env var that overrides the zero-emission tombstone cycles for removed replication bandwidth series. const ENV_REPL_BW_ZERO_TOMBSTONE_CYCLES: &str = "RUSTFS_METRICS_REPL_BW_ZERO_TOMBSTONE_CYCLES"; +const METRICS_RUNTIME_SERVICE_NAME: &str = "metrics_runtime"; +const METRICS_RUNTIME_COLLECTOR_TASKS: u8 = 10; type ReplBwKey = (String, String); // (bucket, target_arn) +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MetricsRuntimeServiceState { + Disabled, + Running, + Stopping, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MetricsRuntimeCancellationSource { + RuntimeToken, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MetricsRuntimeShutdownHandle { + RuntimeTokenOnly, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct MetricsRuntimeIntervalsSnapshot { + pub cluster_interval_secs: u64, + pub bucket_interval_secs: u64, + pub bucket_replication_bandwidth_interval_secs: u64, + pub node_interval_secs: u64, + pub resource_interval_secs: u64, + pub audit_interval_secs: u64, + pub notification_interval_secs: u64, + pub system_interval_secs: u64, + pub process_interval_secs: u64, + pub replication_bandwidth_zero_tombstone_cycles: u8, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct MetricsRuntimeStatusSnapshot { + pub service: &'static str, + pub state: MetricsRuntimeServiceState, + pub metrics_enabled: bool, + pub collector_tasks: u8, + pub intervals: MetricsRuntimeIntervalsSnapshot, + pub cancellation_source: MetricsRuntimeCancellationSource, + pub shutdown_handle: MetricsRuntimeShutdownHandle, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MetricsRuntimeDesiredState { + Disabled, + Enabled, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct MetricsRuntimeDesiredSnapshot { + pub state: MetricsRuntimeDesiredState, + pub collector_tasks: u8, + pub intervals: MetricsRuntimeIntervalsSnapshot, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct MetricsRuntimeControllerSnapshot { + pub desired: MetricsRuntimeDesiredSnapshot, + pub status: MetricsRuntimeStatusSnapshot, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MetricsRuntimeWorkerMutation { + None, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct MetricsRuntimeReconcilePlan { + pub service: &'static str, + pub desired: MetricsRuntimeDesiredSnapshot, + pub current_state: MetricsRuntimeServiceState, + pub worker_mutation: MetricsRuntimeWorkerMutation, +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct MetricsRuntimeController; + +impl MetricsRuntimeController { + pub fn snapshot(&self, token: &CancellationToken) -> MetricsRuntimeControllerSnapshot { + metrics_runtime_controller_snapshot(token) + } + + pub fn reconcile(&self, token: &CancellationToken) -> MetricsRuntimeReconcilePlan { + let snapshot = self.snapshot(token); + self.reconcile_snapshot(snapshot) + } + + pub fn reconcile_snapshot(&self, snapshot: MetricsRuntimeControllerSnapshot) -> MetricsRuntimeReconcilePlan { + MetricsRuntimeReconcilePlan { + service: METRICS_RUNTIME_SERVICE_NAME, + desired: snapshot.desired, + current_state: snapshot.status.state, + worker_mutation: MetricsRuntimeWorkerMutation::None, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct MetricsRuntimeConfig { + cluster_interval: Duration, + bucket_interval: Duration, + bucket_replication_bandwidth_interval: Duration, + node_interval: Duration, + resource_interval: Duration, + audit_interval: Duration, + notification_interval: Duration, + system_interval: Duration, + process_interval: Duration, + replication_bandwidth_zero_tombstone_cycles: u8, +} + +fn parse_repl_bw_zero_tombstone_cycles() -> u8 { + get_env_opt_u64(ENV_REPL_BW_ZERO_TOMBSTONE_CYCLES) + .filter(|&v| v > 0) + .map(|v| u8::try_from(v).unwrap_or(u8::MAX)) + .unwrap_or(DEFAULT_REPL_BW_ZERO_TOMBSTONE_CYCLES) +} + +/// Parse metrics interval from environment variables with fallback to default. +/// +/// Priority: primary_env > legacy_env > default_env > legacy_default > default_value +fn parse_metrics_interval(primary_env: &str, legacy_env: &str, default_interval: Duration) -> Duration { + get_env_opt_u64(primary_env) + .or_else(|| get_env_opt_u64(legacy_env)) + .or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL)) + .or_else(|| get_env_opt_u64(LEGACY_DEFAULT_INTERVAL)) + .filter(|&v| v > 0) + .map(Duration::from_secs) + .unwrap_or(default_interval) +} + +fn parse_system_metrics_interval() -> Duration { + get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL) + .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| ms / 1000)) + .or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL)) + .filter(|&v| v > 0) + .map(Duration::from_secs) + .unwrap_or(DEFAULT_SYSTEM_METRICS_INTERVAL) +} + +fn configured_metrics_runtime_config() -> MetricsRuntimeConfig { + let cluster_interval = + parse_metrics_interval(ENV_CLUSTER_METRICS_INTERVAL, LEGACY_CLUSTER_INTERVAL, DEFAULT_CLUSTER_METRICS_INTERVAL); + let bucket_interval = + parse_metrics_interval(ENV_BUCKET_METRICS_INTERVAL, LEGACY_BUCKET_INTERVAL, DEFAULT_BUCKET_METRICS_INTERVAL); + let bucket_replication_bandwidth_interval = parse_metrics_interval( + ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, + LEGACY_REPLICATION_BANDWIDTH_INTERVAL, + DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, + ); + let node_interval = parse_metrics_interval(ENV_NODE_METRICS_INTERVAL, LEGACY_NODE_INTERVAL, DEFAULT_NODE_METRICS_INTERVAL); + let resource_interval = + parse_metrics_interval(ENV_RESOURCE_METRICS_INTERVAL, LEGACY_RESOURCE_INTERVAL, DEFAULT_RESOURCE_METRICS_INTERVAL); + let audit_interval = + parse_metrics_interval(ENV_AUDIT_METRICS_INTERVAL, LEGACY_AUDIT_INTERVAL, DEFAULT_AUDIT_METRICS_INTERVAL); + let notification_interval = parse_metrics_interval( + ENV_NOTIFICATION_METRICS_INTERVAL, + LEGACY_NOTIFICATION_INTERVAL, + DEFAULT_NOTIFICATION_METRICS_INTERVAL, + ); + let system_interval = parse_system_metrics_interval(); + let process_interval = resource_interval.min(system_interval); + + MetricsRuntimeConfig { + cluster_interval, + bucket_interval, + bucket_replication_bandwidth_interval, + node_interval, + resource_interval, + audit_interval, + notification_interval, + system_interval, + process_interval, + replication_bandwidth_zero_tombstone_cycles: parse_repl_bw_zero_tombstone_cycles(), + } +} + +fn metrics_runtime_intervals_snapshot(config: MetricsRuntimeConfig) -> MetricsRuntimeIntervalsSnapshot { + MetricsRuntimeIntervalsSnapshot { + cluster_interval_secs: config.cluster_interval.as_secs(), + bucket_interval_secs: config.bucket_interval.as_secs(), + bucket_replication_bandwidth_interval_secs: config.bucket_replication_bandwidth_interval.as_secs(), + node_interval_secs: config.node_interval.as_secs(), + resource_interval_secs: config.resource_interval.as_secs(), + audit_interval_secs: config.audit_interval.as_secs(), + notification_interval_secs: config.notification_interval.as_secs(), + system_interval_secs: config.system_interval.as_secs(), + process_interval_secs: config.process_interval.as_secs(), + replication_bandwidth_zero_tombstone_cycles: config.replication_bandwidth_zero_tombstone_cycles, + } +} + +fn build_metrics_runtime_status_snapshot( + metrics_enabled: bool, + cancellation_requested: bool, + config: MetricsRuntimeConfig, +) -> MetricsRuntimeStatusSnapshot { + let state = if !metrics_enabled { + MetricsRuntimeServiceState::Disabled + } else if cancellation_requested { + MetricsRuntimeServiceState::Stopping + } else { + MetricsRuntimeServiceState::Running + }; + + MetricsRuntimeStatusSnapshot { + service: METRICS_RUNTIME_SERVICE_NAME, + state, + metrics_enabled, + collector_tasks: METRICS_RUNTIME_COLLECTOR_TASKS, + intervals: metrics_runtime_intervals_snapshot(config), + cancellation_source: MetricsRuntimeCancellationSource::RuntimeToken, + shutdown_handle: MetricsRuntimeShutdownHandle::RuntimeTokenOnly, + } +} + +fn build_metrics_runtime_desired_snapshot(metrics_enabled: bool, config: MetricsRuntimeConfig) -> MetricsRuntimeDesiredSnapshot { + let state = if metrics_enabled { + MetricsRuntimeDesiredState::Enabled + } else { + MetricsRuntimeDesiredState::Disabled + }; + + MetricsRuntimeDesiredSnapshot { + state, + collector_tasks: METRICS_RUNTIME_COLLECTOR_TASKS, + intervals: metrics_runtime_intervals_snapshot(config), + } +} + +fn build_metrics_runtime_controller_snapshot( + metrics_enabled: bool, + cancellation_requested: bool, + config: MetricsRuntimeConfig, +) -> MetricsRuntimeControllerSnapshot { + MetricsRuntimeControllerSnapshot { + desired: build_metrics_runtime_desired_snapshot(metrics_enabled, config), + status: build_metrics_runtime_status_snapshot(metrics_enabled, cancellation_requested, config), + } +} + +pub fn metrics_runtime_status_snapshot(token: &CancellationToken) -> MetricsRuntimeStatusSnapshot { + build_metrics_runtime_status_snapshot( + crate::observability_metric_enabled(), + token.is_cancelled(), + configured_metrics_runtime_config(), + ) +} + +pub fn metrics_runtime_controller_snapshot(token: &CancellationToken) -> MetricsRuntimeControllerSnapshot { + build_metrics_runtime_controller_snapshot( + crate::observability_metric_enabled(), + token.is_cancelled(), + configured_metrics_runtime_config(), + ) +} + fn repl_bw_live_keys(stats: &[BucketReplicationBandwidthStats]) -> HashSet { stats.iter().map(|s| (s.bucket.clone(), s.target_arn.clone())).collect() } @@ -202,59 +475,14 @@ fn expire_repl_bw_zero_tombstones(monitor_available: bool, zero_tombstones: &mut /// - `RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL` /// - `RUSTFS_METRICS_RESOURCE_INTERVAL` pub fn init_metrics_runtime(token: CancellationToken) { - const LEGACY_CLUSTER_INTERVAL: &str = "RUSTFS_METRICS_CLUSTER_INTERVAL"; - const LEGACY_BUCKET_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_INTERVAL"; - const LEGACY_NODE_INTERVAL: &str = "RUSTFS_METRICS_NODE_INTERVAL"; - const LEGACY_REPLICATION_BANDWIDTH_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL"; - const LEGACY_RESOURCE_INTERVAL: &str = "RUSTFS_METRICS_RESOURCE_INTERVAL"; - const LEGACY_AUDIT_INTERVAL: &str = "RUSTFS_METRICS_AUDIT_INTERVAL"; - const LEGACY_NOTIFICATION_INTERVAL: &str = "RUSTFS_METRICS_NOTIFICATION_INTERVAL"; - const LEGACY_DEFAULT_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL"; - - fn parse_repl_bw_zero_tombstone_cycles() -> u8 { - get_env_opt_u64(ENV_REPL_BW_ZERO_TOMBSTONE_CYCLES) - .filter(|&v| v > 0) - .map(|v| v.min(u8::MAX as u64) as u8) - .unwrap_or(DEFAULT_REPL_BW_ZERO_TOMBSTONE_CYCLES) - } - - /// Parse metrics interval from environment variables with fallback to default. - /// - /// Priority: primary_env > legacy_env > default_env > legacy_default > default_value - fn parse_metrics_interval(primary_env: &str, legacy_env: &str, default_interval: Duration) -> Duration { - get_env_opt_u64(primary_env) - .or_else(|| get_env_opt_u64(legacy_env)) - .or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL)) - .or_else(|| get_env_opt_u64(LEGACY_DEFAULT_INTERVAL)) - .filter(|&v| v > 0) - .map(Duration::from_secs) - .unwrap_or(default_interval) - } - - // Read intervals from environment or use defaults - let cluster_interval = - parse_metrics_interval(ENV_CLUSTER_METRICS_INTERVAL, LEGACY_CLUSTER_INTERVAL, DEFAULT_CLUSTER_METRICS_INTERVAL); - - let bucket_interval = - parse_metrics_interval(ENV_BUCKET_METRICS_INTERVAL, LEGACY_BUCKET_INTERVAL, DEFAULT_BUCKET_METRICS_INTERVAL); - - let bucket_replication_bandwidth_interval = parse_metrics_interval( - ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, - LEGACY_REPLICATION_BANDWIDTH_INTERVAL, - DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, - ); - - let node_interval = parse_metrics_interval(ENV_NODE_METRICS_INTERVAL, LEGACY_NODE_INTERVAL, DEFAULT_NODE_METRICS_INTERVAL); - - let resource_interval = - parse_metrics_interval(ENV_RESOURCE_METRICS_INTERVAL, LEGACY_RESOURCE_INTERVAL, DEFAULT_RESOURCE_METRICS_INTERVAL); - let audit_interval = - parse_metrics_interval(ENV_AUDIT_METRICS_INTERVAL, LEGACY_AUDIT_INTERVAL, DEFAULT_AUDIT_METRICS_INTERVAL); - let notification_interval = parse_metrics_interval( - ENV_NOTIFICATION_METRICS_INTERVAL, - LEGACY_NOTIFICATION_INTERVAL, - DEFAULT_NOTIFICATION_METRICS_INTERVAL, - ); + let config = configured_metrics_runtime_config(); + let cluster_interval = config.cluster_interval; + let bucket_interval = config.bucket_interval; + let bucket_replication_bandwidth_interval = config.bucket_replication_bandwidth_interval; + let node_interval = config.node_interval; + let resource_interval = config.resource_interval; + let audit_interval = config.audit_interval; + let notification_interval = config.notification_interval; // Spawn task for cluster metrics let token_clone = token.clone(); @@ -360,7 +588,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { let token_clone = token.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(bucket_replication_bandwidth_interval); - let repl_bw_zero_tombstone_cycles = parse_repl_bw_zero_tombstone_cycles(); + let repl_bw_zero_tombstone_cycles = config.replication_bandwidth_zero_tombstone_cycles; let mut prev_live_keys: HashSet = HashSet::new(); let mut zero_tombstones: HashMap = HashMap::new(); let mut has_seen_valid_snapshot = false; @@ -498,18 +726,13 @@ pub fn init_metrics_runtime(token: CancellationToken) { }); // Spawn task for system monitoring metrics (migrated from rustfs-obs::system) - let system_interval = get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL) - .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| ms / 1000)) // Convert ms to seconds - .or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL)) - .filter(|&v| v > 0) - .map(Duration::from_secs) - .unwrap_or(DEFAULT_SYSTEM_METRICS_INTERVAL); + let system_interval = config.system_interval; let token_clone = token.clone(); tokio::spawn(async move { let labels = current_process_metric_labels(); let mut host_system = System::new_all(); - let process_interval = resource_interval.min(system_interval); + let process_interval = config.process_interval; let mut interval = tokio::time::interval(process_interval); let now = Instant::now(); let mut next_resource_run = now; @@ -687,6 +910,21 @@ mod tests { use std::time::Duration; use tokio::time::Instant; + fn fixed_metrics_runtime_config() -> MetricsRuntimeConfig { + MetricsRuntimeConfig { + cluster_interval: Duration::from_secs(60), + bucket_interval: Duration::from_secs(300), + bucket_replication_bandwidth_interval: Duration::from_secs(30), + node_interval: Duration::from_secs(60), + resource_interval: Duration::from_secs(15), + audit_interval: Duration::from_secs(45), + notification_interval: Duration::from_secs(20), + system_interval: Duration::from_secs(10), + process_interval: Duration::from_secs(10), + replication_bandwidth_zero_tombstone_cycles: 3, + } + } + fn repl_bw_key(bucket: &str, target_arn: &str) -> ReplBwKey { (bucket.to_string(), target_arn.to_string()) } @@ -697,6 +935,60 @@ mod tests { .collect() } + #[test] + fn metrics_runtime_status_reports_disabled_state() { + let snapshot = build_metrics_runtime_status_snapshot(false, false, fixed_metrics_runtime_config()); + + assert_eq!(snapshot.service, METRICS_RUNTIME_SERVICE_NAME); + assert_eq!(snapshot.state, MetricsRuntimeServiceState::Disabled); + assert!(!snapshot.metrics_enabled); + assert_eq!(snapshot.collector_tasks, METRICS_RUNTIME_COLLECTOR_TASKS); + assert_eq!(snapshot.intervals.cluster_interval_secs, 60); + assert_eq!(snapshot.intervals.bucket_interval_secs, 300); + assert_eq!(snapshot.intervals.process_interval_secs, 10); + assert_eq!(snapshot.intervals.replication_bandwidth_zero_tombstone_cycles, 3); + assert_eq!(snapshot.cancellation_source, MetricsRuntimeCancellationSource::RuntimeToken); + assert_eq!(snapshot.shutdown_handle, MetricsRuntimeShutdownHandle::RuntimeTokenOnly); + } + + #[test] + fn metrics_runtime_status_reports_running_and_stopping_states() { + let running = build_metrics_runtime_status_snapshot(true, false, fixed_metrics_runtime_config()); + let stopping = build_metrics_runtime_status_snapshot(true, true, fixed_metrics_runtime_config()); + + assert_eq!(running.state, MetricsRuntimeServiceState::Running); + assert_eq!(stopping.state, MetricsRuntimeServiceState::Stopping); + assert!(running.metrics_enabled); + assert!(stopping.metrics_enabled); + } + + #[test] + fn metrics_runtime_controller_reconcile_is_idempotent() { + let controller = MetricsRuntimeController; + let snapshot = build_metrics_runtime_controller_snapshot(true, false, fixed_metrics_runtime_config()); + + let first = controller.reconcile_snapshot(snapshot); + let second = controller.reconcile_snapshot(snapshot); + + assert_eq!(first, second); + assert_eq!(first.service, METRICS_RUNTIME_SERVICE_NAME); + assert_eq!(first.desired.state, MetricsRuntimeDesiredState::Enabled); + assert_eq!(first.current_state, MetricsRuntimeServiceState::Running); + assert_eq!(first.worker_mutation, MetricsRuntimeWorkerMutation::None); + } + + #[test] + fn metrics_runtime_controller_reports_disabled_without_worker_mutation() { + let controller = MetricsRuntimeController; + let snapshot = build_metrics_runtime_controller_snapshot(false, false, fixed_metrics_runtime_config()); + let plan = controller.reconcile_snapshot(snapshot); + + assert_eq!(snapshot.desired.state, MetricsRuntimeDesiredState::Disabled); + assert_eq!(snapshot.status.state, MetricsRuntimeServiceState::Disabled); + assert_eq!(plan.current_state, MetricsRuntimeServiceState::Disabled); + assert_eq!(plan.worker_mutation, MetricsRuntimeWorkerMutation::None); + } + #[test] fn advance_deadline_keeps_future_deadline_unchanged() { let base = Instant::now(); diff --git a/docs/architecture/background-controller-contract.md b/docs/architecture/background-controller-contract.md index 05a441fbc..8557e570f 100644 --- a/docs/architecture/background-controller-contract.md +++ b/docs/architecture/background-controller-contract.md @@ -138,6 +138,20 @@ allocator reclaim loop. Existing backend-specific force handling, idle-streak logic, metrics emission, and runtime-token shutdown behavior remain owned by the current loop. +## BGC-006 Metrics Runtime Status And Controller Surface + +The third low-risk controller/status surface is metrics runtime. It reports the +service name, observability metrics enablement, collector task count, configured +collector intervals, replication bandwidth zero-tombstone cycle count, +runtime-token cancellation state, and the absence of a dedicated shutdown +handle. + +The only allowed worker mutation for this surface is `none`. Reconcile output is +read-only and must not start, stop, resize, wake, or otherwise drive metrics +collector tasks. Existing collector grouping, interval parsing, metrics +emission, replication bandwidth tombstone handling, and runtime-token shutdown +behavior remain owned by the current loops. + ## Future Reconcile Rules Future reconcile work is allowed only after a read-only status snapshot exists. diff --git a/docs/architecture/background-services-inventory.md b/docs/architecture/background-services-inventory.md index e643b59b8..f8606da66 100644 --- a/docs/architecture/background-services-inventory.md +++ b/docs/architecture/background-services-inventory.md @@ -47,7 +47,7 @@ not define a new scheduler, controller framework, or shutdown contract. | Notification runtime | `rustfs/src/server/event.rs::init_event_notifier` initializes live event stream support even when notification targets are disabled; when enabled, it loads server config and activates targets. Config reload uses `NotificationConfigManager::reload_config`. | Installs ECStore event dispatch hook, activates notification targets, manages replay/runtime target state, and supports live event streams. | Notification module state is refreshed from persisted module switches; target health is available through runtime target status. | Keep live event stream support separate from target delivery enablement. Reload must remain admin-triggered and peer-signaled. | | Audit runtime | `rustfs/src/server/audit.rs::start_audit_system` starts audit only when module switches and configured targets allow it. `AuditSystem::reload_config` replaces runtime targets. | Dispatches audit events to configured targets and manages replay workers. | Audit observability records config reloads and target delivery metrics. | Do not couple audit lifecycle to notification lifecycle even though the runtime patterns are similar. | | Dynamic config reload | Admin config handlers call `apply_dynamic_config_for_subsystem`, then `signal_dynamic_config_reload` or `signal_config_snapshot_reload` through the global notification system. | Applies scanner/heal runtime config, audit reloads, notification reloads, and peer reload signals. | Logs local and peer reload failures. Audit reload increments audit config reload metrics. | This is admin-triggered fanout, not a background scheduler. Controller work should preserve per-subsystem validation and error boundaries. | -| Metrics runtime | `crates/obs/src/metrics/scheduler.rs::init_metrics_runtime` spawns multiple interval loops for cluster, bucket, node, resource, audit, notification, and replication bandwidth metrics. | Periodically collects and reports metrics. | Reports through the metrics runtime and logs cancellation warnings. | Keep intervals and collector grouping stable while adding status snapshots. | +| Metrics runtime | `crates/obs/src/metrics/scheduler.rs::init_metrics_runtime` spawns multiple interval loops for cluster, bucket, node, resource, audit, notification, and replication bandwidth metrics. | Periodically collects and reports metrics. | Reports through the metrics runtime, logs cancellation warnings, and exposes a typed read-only status snapshot plus a no-op reconcile plan for enablement, collector task count, intervals, replication bandwidth tombstone cycles, cancellation source, and shutdown handle shape. | Keep intervals and collector grouping stable. The current controller surface does not mutate workers. | | Memory observability | `rustfs/src/memory_observability.rs::init_memory_observability` spawns a token-cancelled sampler. | Periodically records memory snapshots. | Emits memory observability metrics and exposes a read-only status snapshot plus a no-op reconcile plan for metrics enablement, interval, cancellation source, and shutdown handle shape. | This is the first low-risk pilot for controller status because it already has a simple token loop and the pilot does not mutate workers. | | Allocator reclaim | `rustfs/src/allocator_reclaim.rs::init_allocator_reclaim` spawns a token-cancelled reclaim loop when enabled. | Observes reclaimable work and may run allocator reclaim after idle intervals. | Emits reclaim enabled/backend counters, active-request gauges, scanner/heal activity gauges, and reclaim result counters. Exposes a typed read-only status snapshot plus a no-op reconcile plan for enablement, backend, effective force, intervals, cancellation source, and shutdown handle shape. | A controller must preserve idle-streak logic and backend-specific force behavior. The current controller surface does not mutate workers. | | Auto-tuner | `rustfs/src/init.rs::init_auto_tuner` optionally spawns a 60-second loop when `RUSTFS_AUTOTUNER_ENABLED` is true. | Tunes concurrency manager settings from performance metrics. | Logs iteration success/failure. | Treat as behavior-sensitive; a future controller needs explicit rollback because it can change runtime concurrency. | diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 4fe478b89..dd84c6a4e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +5,16 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-background-controller-harness-and-status` -- Baseline: `origin/main` at `624de3114398338c80b87c9b452fc70b1d76fa8a` +- Branch: `overtrue/arch-metrics-runtime-controller-status` +- Baseline: `origin/main` at `76b375c478a8c2f5c79e62e848156a55915ed769` - PR type for this branch: `behavior-change` -- Runtime behavior changes: none; the allocator reclaim controller/status - surface only returns read-only snapshot/reconcile data and never starts, - stops, resizes, or wakes workers. -- Rust code changes: add allocator reclaim status/controller snapshots, - reconcile plans with explicit no-op worker mutation, and controller harness - tests for memory observability plus allocator reclaim. +- Runtime behavior changes: none; the metrics runtime controller/status surface + only returns read-only snapshot/reconcile data and never starts, stops, + resizes, or wakes collectors. +- Rust code changes: add metrics runtime status/controller snapshots and + reconcile plans with explicit no-op worker mutation. - CI/script changes: none. -- Docs changes: record `TEST-BGC-001` harness coverage and the allocator reclaim - controller/status slice. +- Docs changes: record `BGC-006` metrics runtime controller/status scope. ## Phase 0 Tasks @@ -417,27 +415,37 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block and startup call shape. - Verification: focused allocator reclaim tests, compile checks, formatting, migration guards, Rust risk scan, and pre-commit quality gate. +- [x] `BGC-006` Add metrics runtime controller/status surface. + - Acceptance: metrics runtime exposes typed desired/status/controller + snapshots and a typed reconcile plan that reports observability enablement, + collector task count, configured intervals, runtime cancellation, shutdown + handle shape, and no-op worker mutation. + - Must preserve: existing metrics collector grouping, interval parsing, + replication bandwidth tombstone cycles, metrics emission, runtime-token + cancellation, and startup call shape. + - Verification: focused metrics runtime tests, compile checks, formatting, + migration guards, Rust risk scan, and pre-commit quality gate. ## Next PRs -1. `behavior-change`: add the next low-risk background status surface before - broader reconcile work. -2. `test-only`: add config-reload preservation coverage for scanner/heal/ +1. `test-only`: add config-reload preservation coverage for scanner/heal/ lifecycle/replication in `TEST-BGC-002`. +2. `behavior-change`: add another low-risk read-only status surface only after + preserving config-reload and shutdown assumptions. ## Pre-Push Review Log | Expert | Status | Notes | |---|---|---| -| Quality/architecture | pass | Confirmed allocator reclaim follows the established typed desired/status/reconcile shape without adding a generic scheduler, registry, or admin route. | -| Migration preservation | pass | Confirmed reconcile reports `worker_mutation: none`, preserving enablement, force handling, idle-streak logic, cancellation, and metrics emission behavior. | +| Quality/architecture | pass | Confirmed metrics runtime follows the established typed desired/status/reconcile shape without adding a generic scheduler, registry, or admin route. | +| Migration preservation | pass | Confirmed reconcile reports `worker_mutation: none`, preserving collector grouping, interval parsing, tombstone cycles, cancellation, and metrics emission behavior. | | Testing/verification | pass | Focused tests, compile checks, formatting, diff hygiene, migration guards, Rust risk scan, and `make pre-commit` passed. | ## Verification Notes -Passed on `624de3114398338c80b87c9b452fc70b1d76fa8a`: -- `cargo test -p rustfs allocator_reclaim --lib`; 9 passed. -- `cargo test -p rustfs memory_observability --lib`; 9 passed. +Passed on `76b375c478a8c2f5c79e62e848156a55915ed769`: +- `cargo test -p rustfs-obs metrics_runtime --lib`; 4 passed. +- `cargo check -p rustfs-obs`. - `cargo check -p rustfs --lib`. - `cargo fmt --all`. - `cargo fmt --all --check`. @@ -445,20 +453,20 @@ Passed on `624de3114398338c80b87c9b452fc70b1d76fa8a`: - `./scripts/check_architecture_migration_rules.sh`. - `./scripts/check_layer_dependencies.sh`. - `./scripts/check_metrics_migration_refs.sh`. -- Rust risk scan for changed Rust files; only test `expect` calls and existing - internal allocator reclaim `Result<(), String>` helpers matched. -- `make pre-commit`; all checks passed, including nextest 5874 passed and 111 +- Rust risk scan for changed Rust files; only an existing doc-comment + `println!` example and an existing bounded scheduler numeric cast matched. +- `make pre-commit`; all checks passed, including nextest 5880 passed and 111 skipped. Notes: - This slice adds reconcile/status data only. It does not apply reconcile output - to the running allocator reclaim loop or memory observability sampler. + to the running metrics collectors. - There is no admin/API exposure in this PR; future controller harness work should stay isolated from scanner, heal, lifecycle, and replication. ## Handoff Notes -- Next migration slice can start another read-only low-risk status surface or - `TEST-BGC-002` config-reload preservation coverage. +- Next migration slice should start `TEST-BGC-002` config-reload preservation + coverage before broader controller surfaces. - Keep scanner, heal, lifecycle, replication, disk health, and config reload out of broad controller movement until dedicated preservation tests exist.