diff --git a/crates/madmin/fixtures/on_demand_migration/status.json b/crates/madmin/fixtures/on_demand_migration/status.json index 848abd355..6806d03b9 100644 --- a/crates/madmin/fixtures/on_demand_migration/status.json +++ b/crates/madmin/fixtures/on_demand_migration/status.json @@ -1 +1 @@ -{"configured":true,"enabled":true,"module_enabled":false} +{"configured":true,"enabled":true,"module_enabled":true,"provider":"minio","endpoint_host":"source.example.com","breaker":{"state":"half_open","opened_at":null},"counters":{"requests_total":{"get":{"breaker_open":0,"filtered":0,"negative_cached":0,"source_error":0,"source_hit":2,"source_miss":0,"unsupported":0},"head":{"breaker_open":0,"filtered":0,"negative_cached":1,"source_error":0,"source_hit":0,"source_miss":0,"unsupported":0}},"pulled_bytes_total":4096,"pulled_objects_total":{"backfill":0,"background":0,"inline":1},"pull_failures_total":{"canceled":0,"etag_mismatch":0,"local_write":0,"queue_full":0,"source_access_denied":0,"source_connect":0,"source_not_found":0,"source_other":0,"source_server_error":0,"source_throttled":0,"source_timeout":1,"source_unsupported":0},"source_latency":{"buckets":[{"le_ms":5,"count":1},{"le_ms":10,"count":1},{"le_ms":20,"count":1},{"le_ms":50,"count":1},{"le_ms":100,"count":1},{"le_ms":200,"count":1},{"le_ms":500,"count":1},{"le_ms":1000,"count":2},{"le_ms":2000,"count":2},{"le_ms":5000,"count":2},{"le_ms":10000,"count":2},{"le_ms":20000,"count":2},{"le_ms":30000,"count":2},{"le_ms":60000,"count":2}],"count":3,"sum_ms":90753}},"last_source_error":{"class":"server_error","at":"2026-09-02T10:00:00Z"},"inflight_pulls":1,"queue_depth":1,"served_by_source_ratio":null,"updated_at":"2026-09-02T10:00:00Z"} diff --git a/crates/madmin/src/on_demand_migration.rs b/crates/madmin/src/on_demand_migration.rs index ffbf60ee9..3d213098a 100644 --- a/crates/madmin/src/on_demand_migration.rs +++ b/crates/madmin/src/on_demand_migration.rs @@ -26,6 +26,7 @@ use crate::client::{AdminClient, AdminClientError, percent_encode_path_segment}; use http::Method; use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; use std::fmt; /// Config schema version this client speaks. @@ -295,12 +296,93 @@ pub struct OnDemandMigrationGetResponse { pub updated_at: String, } -/// `GET .../status` response. Runtime counters are added by later ODM tasks. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// `GET .../status` response: the switch state plus this node's runtime +/// snapshot of the bucket. The runtime fields are `null` while the bucket has +/// no live state on the answering node (module off, config absent or +/// disabled); `provider` and `endpoint_host` then still describe the saved +/// config, if any. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct OnDemandMigrationStatus { pub configured: bool, pub enabled: bool, pub module_enabled: bool, + #[serde(default)] + pub provider: Option, + /// Host of the source endpoint, without scheme or port. + #[serde(default)] + pub endpoint_host: Option, + #[serde(default)] + pub breaker: Option, + #[serde(default)] + pub counters: Option, + #[serde(default)] + pub last_source_error: Option, + #[serde(default)] + pub inflight_pulls: u64, + #[serde(default)] + pub queue_depth: u64, + /// `source_hit / (source_hit + local GETs)`; `None` when the server has + /// no per-bucket GET total to divide by (it never reports a made-up 0). + #[serde(default)] + pub served_by_source_ratio: Option, + /// RFC 3339 save time of the config; `None` when not configured. + #[serde(default)] + pub updated_at: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OnDemandMigrationBreakerState { + Closed, + Open, + HalfOpen, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationBreaker { + pub state: OnDemandMigrationBreakerState, + /// RFC 3339 time the breaker last opened; `None` while closed or when + /// the server does not report it. + #[serde(default)] + pub opened_at: Option, +} + +/// Lifetime counters of the bucket's runtime on the answering node. Keys are +/// the fixed label values of the Prometheus series with the same names. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationCounters { + /// `op -> outcome -> count`. + pub requests_total: BTreeMap>, + pub pulled_bytes_total: u64, + /// `path -> count`. + pub pulled_objects_total: BTreeMap, + /// `reason -> count`. + pub pull_failures_total: BTreeMap, + pub source_latency: OnDemandMigrationSourceLatency, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSourceLatency { + pub buckets: Vec, + /// Total observations, including those above the last bound. + pub count: u64, + pub sum_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationLatencyBucket { + /// Upper bound of the bucket in milliseconds. + pub le_ms: u64, + /// Cumulative observations at or below `le_ms`. + pub count: u64, +} + +/// The most recent source failure: class only, never the key or message. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSourceError { + pub class: String, + /// RFC 3339. + pub at: String, } fn config_path(bucket: &str) -> String { @@ -401,7 +483,42 @@ mod tests { let response: OnDemandMigrationGetResponse = round_trip(GET_RESPONSE_FIXTURE); assert_eq!(response.updated_at, "2026-09-02T10:00:00Z"); let status: OnDemandMigrationStatus = round_trip(STATUS_FIXTURE); - assert!(status.configured && status.enabled && !status.module_enabled); + assert!(status.configured && status.enabled && status.module_enabled); + assert_eq!(status.provider.as_deref(), Some("minio")); + assert_eq!(status.endpoint_host.as_deref(), Some("source.example.com")); + let breaker = status.breaker.expect("breaker present"); + assert_eq!(breaker.state, OnDemandMigrationBreakerState::HalfOpen); + assert_eq!(breaker.opened_at, None); + let counters = status.counters.expect("counters present"); + assert_eq!(counters.requests_total["get"]["source_hit"], 2); + assert_eq!(counters.requests_total["head"]["negative_cached"], 1); + assert_eq!(counters.pulled_bytes_total, 4096); + assert_eq!(counters.pulled_objects_total["inline"], 1); + assert_eq!(counters.pull_failures_total["source_timeout"], 1); + assert_eq!(counters.source_latency.buckets.len(), 14); + assert_eq!(counters.source_latency.count, 3); + assert_eq!(counters.source_latency.sum_ms, 90_753); + let last_error = status.last_source_error.expect("last source error present"); + assert_eq!(last_error.class, "server_error"); + assert_eq!(last_error.at, "2026-09-02T10:00:00Z"); + assert_eq!(status.inflight_pulls, 1); + assert_eq!(status.queue_depth, 1); + assert_eq!(status.served_by_source_ratio, None, "the ratio is null, never a fabricated 0"); + assert_eq!(status.updated_at.as_deref(), Some("2026-09-02T10:00:00Z")); + } + + #[test] + fn status_without_runtime_state_decodes_with_null_runtime_fields() { + let status: OnDemandMigrationStatus = serde_json::from_str( + r#"{"configured":false,"enabled":false,"module_enabled":false,"provider":null,"endpoint_host":null,"breaker":null,"counters":null,"last_source_error":null,"inflight_pulls":0,"queue_depth":0,"served_by_source_ratio":null,"updated_at":null}"#, + ) + .expect("status decodes"); + assert!(!status.configured); + assert_eq!(status.provider, None); + assert_eq!(status.breaker, None); + assert_eq!(status.counters, None); + assert_eq!(status.served_by_source_ratio, None); + assert_eq!(status.updated_at, None); } #[test] diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index c4f849a4d..bdb1d8210 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -27,6 +27,7 @@ pub mod ilm; pub mod node; pub mod notification; pub mod notification_target; +pub mod on_demand_migration; pub mod replication; pub(crate) mod request; pub mod resource; @@ -68,6 +69,9 @@ pub(crate) use notification::collect_notification_runtime_metrics; pub use notification::{NotificationStats, collect_notification_metrics}; pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics}; pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics}; +pub use on_demand_migration::{ + OnDemandMigrationBreakerState, OnDemandMigrationBucketStats, collect_on_demand_migration_metrics, source_latency_le_label, +}; pub use replication::{ReplicationMetricsSnapshot, collect_replication_metrics}; pub(crate) use replication::{ReplicationRuntimeStats, collect_replication_runtime_metrics}; pub(crate) use request::{ApiRequestMetricSupport, ApiRequestStats, collect_request_metrics}; diff --git a/crates/obs/src/metrics/collectors/on_demand_migration.rs b/crates/obs/src/metrics/collectors/on_demand_migration.rs new file mode 100644 index 000000000..b854a0b3c --- /dev/null +++ b/crates/obs/src/metrics/collectors/on_demand_migration.rs @@ -0,0 +1,357 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! On-Demand Migration metrics collector (rustfs/backlog#2157). +//! +//! [`OnDemandMigrationBucketStats`] is the obs-side projection of one +//! bucket's runtime snapshot; the storage boundary fills it and this module +//! turns it into Prometheus series. Label values come from the snapshot +//! itself, so the collector never invents a label the runtime did not count. + +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::on_demand_migration::{ + BREAKER_STATE_CLOSED, BREAKER_STATE_HALF_OPEN, BREAKER_STATE_OPEN, BUCKET_L, LE_L, ODM_BREAKER_STATE_MD, + ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD, ODM_PULLED_BYTES_TOTAL_MD, ODM_PULLED_OBJECTS_TOTAL_MD, + ODM_QUEUE_DEPTH_MD, ODM_REQUESTS_TOTAL_MD, ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, + ODM_SOURCE_LATENCY_SECONDS_SUM_MD, OP_L, OUTCOME_L, PATH_L, REASON_L, +}; +use std::borrow::Cow; +use std::collections::BTreeMap; + +/// Bucket-scoped series that do not depend on a label set: +/// `pulled_bytes_total`, `inflight_pulls`, `queue_depth`, `breaker_state`, +/// `source_latency_seconds_sum`, `source_latency_seconds_count`. +const FIXED_METRICS_PER_BUCKET: usize = 6; + +/// Breaker state as the runtime reports it. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum OnDemandMigrationBreakerState { + #[default] + Closed, + HalfOpen, + Open, +} + +impl OnDemandMigrationBreakerState { + /// The `breaker_state` gauge value. + pub fn gauge_value(self) -> f64 { + match self { + Self::Closed => BREAKER_STATE_CLOSED, + Self::HalfOpen => BREAKER_STATE_HALF_OPEN, + Self::Open => BREAKER_STATE_OPEN, + } + } +} + +/// One bucket of the runtime snapshot. Maps are keyed by the runtime's +/// label values (`op -> outcome -> count`, `path -> count`, +/// `reason -> count`); `BTreeMap` keeps the emitted series order stable. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct OnDemandMigrationBucketStats { + pub bucket: String, + pub requests_total: BTreeMap>, + pub pulled_bytes_total: u64, + pub pulled_objects_total: BTreeMap, + pub pull_failures_total: BTreeMap, + pub inflight_pulls: u64, + pub queue_depth: u64, + /// `(upper bound in milliseconds, cumulative observations at or below it)`, + /// ascending; observations above the last bound are only in `count`. + pub source_latency_buckets: Vec<(u64, u64)>, + pub source_latency_count: u64, + pub source_latency_sum_ms: u64, + pub breaker_state: OnDemandMigrationBreakerState, +} + +/// Renders a millisecond bound as the `le` label in seconds (`5` -> `0.005`, +/// `1000` -> `1`). +pub fn source_latency_le_label(le_ms: u64) -> String { + let secs = le_ms as f64 / 1000.0; + if secs.fract() == 0.0 { + format!("{}", secs as u64) + } else { + format!("{secs}") + } +} + +pub fn collect_on_demand_migration_metrics(stats: &[OnDemandMigrationBucketStats]) -> Vec { + if stats.is_empty() { + return Vec::new(); + } + + let metric_count = stats + .iter() + .map(|stat| { + FIXED_METRICS_PER_BUCKET + + stat.requests_total.values().map(BTreeMap::len).sum::() + + stat.pulled_objects_total.len() + + stat.pull_failures_total.len() + + stat.source_latency_buckets.len() + + 1 + }) + .sum(); + let mut metrics = Vec::with_capacity(metric_count); + for stat in stats { + let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); + + for (op, by_outcome) in &stat.requests_total { + for (outcome, count) in by_outcome { + metrics.push( + PrometheusMetric::from_descriptor(&ODM_REQUESTS_TOTAL_MD, *count as f64) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label_owned(OP_L, op.clone()) + .with_label_owned(OUTCOME_L, outcome.clone()), + ); + } + } + metrics.push( + PrometheusMetric::from_descriptor(&ODM_PULLED_BYTES_TOTAL_MD, stat.pulled_bytes_total as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + for (path, count) in &stat.pulled_objects_total { + metrics.push( + PrometheusMetric::from_descriptor(&ODM_PULLED_OBJECTS_TOTAL_MD, *count as f64) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label_owned(PATH_L, path.clone()), + ); + } + for (reason, count) in &stat.pull_failures_total { + metrics.push( + PrometheusMetric::from_descriptor(&ODM_PULL_FAILURES_TOTAL_MD, *count as f64) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label_owned(REASON_L, reason.clone()), + ); + } + metrics.push( + PrometheusMetric::from_descriptor(&ODM_INFLIGHT_PULLS_MD, stat.inflight_pulls as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&ODM_QUEUE_DEPTH_MD, stat.queue_depth as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + for (le_ms, cumulative) in &stat.source_latency_buckets { + metrics.push( + PrometheusMetric::from_descriptor(&ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, *cumulative as f64) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label_owned(LE_L, source_latency_le_label(*le_ms)), + ); + } + metrics.push( + PrometheusMetric::from_descriptor(&ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, stat.source_latency_count as f64) + .with_label(BUCKET_L, bucket_label.clone()) + .with_label(LE_L, Cow::Borrowed("+Inf")), + ); + metrics.push( + PrometheusMetric::from_descriptor(&ODM_SOURCE_LATENCY_SECONDS_SUM_MD, stat.source_latency_sum_ms as f64 / 1000.0) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, stat.source_latency_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&ODM_BREAKER_STATE_MD, stat.breaker_state.gauge_value()) + .with_label(BUCKET_L, bucket_label), + ); + } + + metrics +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use crate::metrics::schema::on_demand_migration::{PULL_FAILURE_REASONS, PULL_PATHS, REQUEST_OPS, REQUEST_OUTCOMES}; + + /// The bucket the runtime golden snapshot describes (ecstore + /// `snapshot_matches_golden_json`): two GET source hits, one negative + /// cached HEAD, one inline pull of 4096 bytes, one timeout failure, three + /// source calls of 3 ms, 750 ms and 90 s, one inflight and one queued pull. + pub(crate) fn golden_stats(bucket: &str) -> OnDemandMigrationBucketStats { + let mut requests_total = BTreeMap::new(); + for op in REQUEST_OPS { + let by_outcome: BTreeMap = REQUEST_OUTCOMES + .iter() + .map(|outcome| { + let count = match (op, *outcome) { + ("get", "source_hit") => 2, + ("head", "negative_cached") => 1, + _ => 0, + }; + (outcome.to_string(), count) + }) + .collect(); + requests_total.insert(op.to_string(), by_outcome); + } + let pulled_objects_total = PULL_PATHS + .iter() + .map(|path| (path.to_string(), u64::from(*path == "inline"))) + .collect(); + let pull_failures_total = PULL_FAILURE_REASONS + .iter() + .map(|reason| (reason.to_string(), u64::from(*reason == "source_timeout"))) + .collect(); + let bounds_ms = [ + 5, 10, 20, 50, 100, 200, 500, 1_000, 2_000, 5_000, 10_000, 20_000, 30_000, 60_000, + ]; + let source_latency_buckets = bounds_ms + .iter() + .map(|bound| (*bound, if *bound < 1_000 { 1 } else { 2 })) + .collect(); + OnDemandMigrationBucketStats { + bucket: bucket.to_string(), + requests_total, + pulled_bytes_total: 4096, + pulled_objects_total, + pull_failures_total, + inflight_pulls: 1, + queue_depth: 1, + source_latency_buckets, + source_latency_count: 3, + source_latency_sum_ms: 90_753, + breaker_state: OnDemandMigrationBreakerState::HalfOpen, + } + } + + /// `(name, labels, value)` of one emitted series. + type Rendered = (String, Vec<(String, String)>, f64); + + /// Emitted series in emission order. + fn rendered(metrics: &[PrometheusMetric]) -> Vec { + metrics + .iter() + .map(|metric| { + ( + metric.name.to_string(), + metric + .labels + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect(), + metric.value, + ) + }) + .collect() + } + + fn series(name: &str, labels: &[(&str, &str)], value: f64) -> Rendered { + ( + format!("rustfs_on_demand_migration_{name}"), + labels.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(), + value, + ) + } + + #[test] + fn golden_snapshot_renders_every_series_with_fixed_labels() { + let metrics = collect_on_demand_migration_metrics(&[golden_stats("photos")]); + let actual = rendered(&metrics); + + let mut expected = Vec::new(); + for op in REQUEST_OPS { + for outcome in REQUEST_OUTCOMES { + let value = match (op, outcome) { + ("get", "source_hit") => 2.0, + ("head", "negative_cached") => 1.0, + _ => 0.0, + }; + expected.push(series("requests_total", &[("bucket", "photos"), ("op", op), ("outcome", outcome)], value)); + } + } + // BTreeMap order: the golden fixture's outcomes sort alphabetically. + expected.sort_by(|a, b| a.1.cmp(&b.1)); + expected.push(series("pulled_bytes_total", &[("bucket", "photos")], 4096.0)); + for path in ["backfill", "background", "inline"] { + expected.push(series( + "pulled_objects_total", + &[("bucket", "photos"), ("path", path)], + f64::from(u8::from(path == "inline")), + )); + } + let mut reasons = PULL_FAILURE_REASONS; + reasons.sort_unstable(); + for reason in reasons { + expected.push(series( + "pull_failures_total", + &[("bucket", "photos"), ("reason", reason)], + f64::from(u8::from(reason == "source_timeout")), + )); + } + expected.push(series("inflight_pulls", &[("bucket", "photos")], 1.0)); + expected.push(series("queue_depth", &[("bucket", "photos")], 1.0)); + for (le, value) in [ + ("0.005", 1.0), + ("0.01", 1.0), + ("0.02", 1.0), + ("0.05", 1.0), + ("0.1", 1.0), + ("0.2", 1.0), + ("0.5", 1.0), + ("1", 2.0), + ("2", 2.0), + ("5", 2.0), + ("10", 2.0), + ("20", 2.0), + ("30", 2.0), + ("60", 2.0), + ("+Inf", 3.0), + ] { + expected.push(series("source_latency_seconds_distribution", &[("bucket", "photos"), ("le", le)], value)); + } + expected.push(series("source_latency_seconds_sum", &[("bucket", "photos")], 90.753)); + expected.push(series("source_latency_seconds_count", &[("bucket", "photos")], 3.0)); + expected.push(series("breaker_state", &[("bucket", "photos")], 1.0)); + + assert_eq!(actual.len(), expected.len()); + for (actual, expected) in actual.iter().zip(&expected) { + assert_eq!(actual, expected); + } + } + + #[test] + fn breaker_state_gauge_encodes_the_three_states() { + for (state, value) in [ + (OnDemandMigrationBreakerState::Closed, 0.0), + (OnDemandMigrationBreakerState::HalfOpen, 1.0), + (OnDemandMigrationBreakerState::Open, 2.0), + ] { + let stats = OnDemandMigrationBucketStats { + bucket: "b".to_string(), + breaker_state: state, + ..Default::default() + }; + let metrics = collect_on_demand_migration_metrics(&[stats]); + let breaker = metrics + .iter() + .find(|metric| metric.name == ODM_BREAKER_STATE_MD.get_full_metric_name()) + .expect("breaker gauge"); + assert_eq!(breaker.value, value); + } + } + + #[test] + fn le_labels_render_bounds_in_seconds() { + assert_eq!(source_latency_le_label(5), "0.005"); + assert_eq!(source_latency_le_label(20), "0.02"); + assert_eq!(source_latency_le_label(500), "0.5"); + assert_eq!(source_latency_le_label(1_000), "1"); + assert_eq!(source_latency_le_label(60_000), "60"); + } + + #[test] + fn empty_snapshot_emits_nothing() { + assert!(collect_on_demand_migration_metrics(&[]).is_empty()); + } +} diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index af2c4b8ab..529a66b15 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -34,6 +34,6 @@ pub(crate) use storage_api::metrics::{ BucketOperations, BucketOptions, ObsBucketBandwidthMonitor, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, StorageAdminApi, obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, - obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, - obs_resolve_object_store_handle, obs_transition_state_handle, + obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_snapshot, + obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle, }; diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 29734268f..39a1a37fd 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -33,6 +33,7 @@ use crate::metrics::collectors::{ NotificationStats, NotificationTargetRuntimeStats, NotificationTargetStats, + OnDemandMigrationBucketStats, // System monitoring collectors (migrated from rustfs-obs::system) ProcessAttributeError, ProcessCpuStats, @@ -64,6 +65,7 @@ use crate::metrics::collectors::{ collect_node_metrics, collect_notification_runtime_metrics, collect_notification_target_runtime_metrics, + collect_on_demand_migration_metrics, collect_process_attributes, collect_process_cpu_metrics, collect_process_disk_metrics, @@ -118,6 +120,14 @@ use crate::metrics::schema::notification_target::{ NOTIFICATION_TARGET_TOTAL_MESSAGES_BY_SERVER_MD, NOTIFICATION_TARGET_TOTAL_MESSAGES_MD, SERVER as NOTIFICATION_SERVER_LABEL, TARGET_ID as NOTIFICATION_TARGET_ID_LABEL, TARGET_TYPE as NOTIFICATION_TARGET_TYPE_LABEL, }; +use crate::metrics::schema::on_demand_migration::{ + BUCKET_L as ODM_BUCKET_L, LE_L as ODM_LE_L, ODM_BREAKER_STATE_MD, ODM_INFLIGHT_PULLS_MD, ODM_PULL_FAILURES_TOTAL_MD, + ODM_PULLED_BYTES_TOTAL_MD, ODM_PULLED_OBJECTS_TOTAL_MD, ODM_QUEUE_DEPTH_MD, ODM_REQUESTS_TOTAL_MD, + ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, ODM_SOURCE_LATENCY_SECONDS_SUM_MD, + OP_L as ODM_OP_L, OUTCOME_L as ODM_OUTCOME_L, PATH_L as ODM_PATH_L, PULL_FAILURE_REASONS as ODM_PULL_FAILURE_REASONS, + PULL_PATHS as ODM_PULL_PATHS, REASON_L as ODM_REASON_L, REQUEST_OPS as ODM_REQUEST_OPS, + REQUEST_OUTCOMES as ODM_REQUEST_OUTCOMES, SOURCE_LATENCY_LE as ODM_SOURCE_LATENCY_LE, +}; use crate::metrics::schema::scanner::{ BUCKET_LABEL as SCANNER_BUCKET_LABEL, CYCLE_SCOPE_LABEL as SCANNER_CYCLE_SCOPE_LABEL, DRIVE_LABEL as SCANNER_DRIVE_LABEL, RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD, SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD, @@ -134,8 +144,9 @@ use crate::metrics::stats_collector::{ collect_bucket_replication_stats_bundle, collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats, collect_compression_cluster_stats, collect_disk_and_system_drive_runtime_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, - collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_process_metric_bundle_with, - collect_replication_stats, collect_scanner_runtime_metric_stats, collect_system_cpu_and_memory_stats_with, + collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_on_demand_migration_stats, + collect_process_metric_bundle_with, collect_replication_stats, collect_scanner_runtime_metric_stats, + collect_system_cpu_and_memory_stats_with, }; use crate::node_identity::{SERVER_LABEL, current_local_node_identity}; use crate::telemetry::retire_metric_series; @@ -298,6 +309,8 @@ static METRICS_RUNTIME_COLLECTOR_HEALTH: OnceLock type ReplBwKey = (String, String); // (bucket, target_arn) type BucketKey = String; +/// `(full metric name, labels)` of one series. +type MetricSeriesKey = (String, Vec<(&'static str, Cow<'static, str>)>); type BucketRangeKey = (String, String); // (bucket, range) type AuditLegacyTargetKey = String; type AuditTargetKey = (String, String); // (server, target_id) @@ -897,6 +910,10 @@ fn repl_proxy_bucket_live_keys(stats: &[BucketReplicationRuntimeStats]) -> HashS stats.iter().map(|stat| stat.stats.bucket.clone()).collect() } +fn on_demand_migration_bucket_live_keys(stats: &[OnDemandMigrationBucketStats]) -> HashSet { + stats.iter().map(|stat| stat.bucket.clone()).collect() +} + fn update_series_zero_tombstones( has_seen_valid_snapshot: &mut bool, prev_live_keys: &mut HashSet, @@ -1577,6 +1594,58 @@ fn retire_bucket_replication_proxy_request_metric_series(bucket: &str) -> usize retired } +/// Every on-demand migration series a bucket can own, as `(name, labels)`: +/// the fixed label sets of the runtime snapshot, in the collector's label +/// order. Retirement walks this list once the bucket's config is gone. +fn on_demand_migration_metric_series(bucket: &str) -> Vec { + let bucket_label = || (ODM_BUCKET_L, Cow::Owned(bucket.to_string())); + let mut series = Vec::new(); + for op in ODM_REQUEST_OPS { + for outcome in ODM_REQUEST_OUTCOMES { + series.push(( + ODM_REQUESTS_TOTAL_MD.get_full_metric_name(), + vec![ + bucket_label(), + (ODM_OP_L, Cow::Borrowed(op)), + (ODM_OUTCOME_L, Cow::Borrowed(outcome)), + ], + )); + } + } + series.push((ODM_PULLED_BYTES_TOTAL_MD.get_full_metric_name(), vec![bucket_label()])); + for path in ODM_PULL_PATHS { + series.push(( + ODM_PULLED_OBJECTS_TOTAL_MD.get_full_metric_name(), + vec![bucket_label(), (ODM_PATH_L, Cow::Borrowed(path))], + )); + } + for reason in ODM_PULL_FAILURE_REASONS { + series.push(( + ODM_PULL_FAILURES_TOTAL_MD.get_full_metric_name(), + vec![bucket_label(), (ODM_REASON_L, Cow::Borrowed(reason))], + )); + } + series.push((ODM_INFLIGHT_PULLS_MD.get_full_metric_name(), vec![bucket_label()])); + series.push((ODM_QUEUE_DEPTH_MD.get_full_metric_name(), vec![bucket_label()])); + for le in ODM_SOURCE_LATENCY_LE { + series.push(( + ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD.get_full_metric_name(), + vec![bucket_label(), (ODM_LE_L, Cow::Borrowed(le))], + )); + } + series.push((ODM_SOURCE_LATENCY_SECONDS_SUM_MD.get_full_metric_name(), vec![bucket_label()])); + series.push((ODM_SOURCE_LATENCY_SECONDS_COUNT_MD.get_full_metric_name(), vec![bucket_label()])); + series.push((ODM_BREAKER_STATE_MD.get_full_metric_name(), vec![bucket_label()])); + series +} + +fn retire_on_demand_migration_metric_series(bucket: &str) -> usize { + on_demand_migration_metric_series(bucket) + .iter() + .map(|(name, labels)| retire_metric_series(name, labels)) + .sum() +} + fn retire_repl_backlog_target_metric_series(bucket: &str, target_arn: &str) -> usize { let labels = [ (BUCKET_L, Cow::Owned(bucket.to_string())), @@ -1966,6 +2035,8 @@ pub fn init_metrics_runtime(token: CancellationToken) { let mut has_seen_valid_flow_snapshot = false; let mut prev_proxy_bucket_live_keys: HashSet = HashSet::new(); let mut has_seen_proxy_bucket_snapshot = false; + let mut prev_on_demand_migration_live_keys: HashSet = HashSet::new(); + let mut has_seen_on_demand_migration_snapshot = false; loop { tokio::select! { _ = interval.tick() => { @@ -2039,6 +2110,21 @@ pub fn init_metrics_runtime(token: CancellationToken) { metrics.extend(collect_repl_backlog_zero_tombstone_metrics(&backlog_zero_tombstones)); metrics.extend(collect_repl_backlog_target_zero_tombstone_metrics(&backlog_target_zero_tombstones)); metrics.extend(collect_repl_flow_zero_tombstone_metrics(&flow_zero_tombstones)); + // A bucket whose on-demand migration config is gone leaves the + // snapshot; its series are retired after this cycle's report. + let on_demand_migration = collect_on_demand_migration_stats(); + let current_on_demand_migration_live_keys = on_demand_migration_bucket_live_keys(&on_demand_migration); + let retire_on_demand_migration_buckets = if has_seen_on_demand_migration_snapshot { + prev_on_demand_migration_live_keys + .difference(¤t_on_demand_migration_live_keys) + .cloned() + .collect::>() + } else { + Vec::new() + }; + prev_on_demand_migration_live_keys = current_on_demand_migration_live_keys; + has_seen_on_demand_migration_snapshot = true; + metrics.extend(collect_on_demand_migration_metrics(&on_demand_migration)); let replication = collect_replication_stats().await; metrics.extend(collect_replication_runtime_metrics(&ReplicationRuntimeStats { server: current_local_node_identity(), @@ -2065,6 +2151,9 @@ pub fn init_metrics_runtime(token: CancellationToken) { for bucket in retire_proxy_buckets { let _ = retire_bucket_replication_proxy_request_metric_series(&bucket); } + for bucket in retire_on_demand_migration_buckets { + let _ = retire_on_demand_migration_metric_series(&bucket); + } }, ).await; } @@ -2830,6 +2919,67 @@ mod tests { assert_eq!(current, bucket_keys(&["logs"])); } + #[test] + fn on_demand_migration_bucket_keys_detect_removed_buckets() { + let previous = on_demand_migration_bucket_live_keys(&[ + OnDemandMigrationBucketStats { + bucket: "photos".to_string(), + ..Default::default() + }, + OnDemandMigrationBucketStats { + bucket: "logs".to_string(), + ..Default::default() + }, + ]); + let current = on_demand_migration_bucket_live_keys(&[OnDemandMigrationBucketStats { + bucket: "logs".to_string(), + ..Default::default() + }]); + let retired = previous.difference(¤t).cloned().collect::>(); + + assert_eq!(retired, bucket_keys(&["photos"])); + assert_eq!(current, bucket_keys(&["logs"])); + assert!(on_demand_migration_bucket_live_keys(&[]).is_empty()); + } + + /// Retirement must name exactly the series the collector emitted for the + /// bucket, with identical label order, or the recorder keeps them alive. + #[test] + fn on_demand_migration_retirement_covers_every_emitted_series() { + let stats = crate::metrics::collectors::on_demand_migration::tests::golden_stats("photos"); + let emitted = collect_on_demand_migration_metrics(&[stats]) + .into_iter() + .map(|metric| { + ( + metric.name.to_string(), + metric + .labels + .into_iter() + .map(|(key, value)| (key, value.to_string())) + .collect::>(), + ) + }) + .collect::>(); + let retired = on_demand_migration_metric_series("photos") + .into_iter() + .map(|(name, labels)| { + ( + name, + labels + .into_iter() + .map(|(key, value)| (key, value.to_string())) + .collect::>(), + ) + }) + .collect::>(); + + assert_eq!(retired, emitted); + // Without a process-global recorder there is nothing to retire; the + // walk itself must still cover every series. + assert_eq!(retire_on_demand_migration_metric_series("photos"), 0); + assert_eq!(on_demand_migration_metric_series("photos").len(), emitted.len()); + } + #[test] fn metrics_runtime_status_reports_disabled_state() { let snapshot = build_metrics_runtime_status_snapshot(false, false, fixed_metrics_runtime_config(), false); diff --git a/crates/obs/src/metrics/schema/entry/subsystem.rs b/crates/obs/src/metrics/schema/entry/subsystem.rs index 7b0606890..255fe6892 100644 --- a/crates/obs/src/metrics/schema/entry/subsystem.rs +++ b/crates/obs/src/metrics/schema/entry/subsystem.rs @@ -52,6 +52,7 @@ pub enum MetricSubsystem { Notification, Scanner, Compression, + OnDemandMigration, // Custom paths Custom(String), @@ -95,6 +96,7 @@ impl MetricSubsystem { Self::Notification => "/notification", Self::Scanner => "/scanner", Self::Compression => "/compression", + Self::OnDemandMigration => "/on-demand-migration", // Custom paths Self::Custom(path) => path, @@ -143,6 +145,7 @@ impl MetricSubsystem { "/notification" => Self::Notification, "/scanner" => Self::Scanner, "/compression" => Self::Compression, + "/on-demand-migration" => Self::OnDemandMigration, // Treat other paths as custom subsystems _ => Self::Custom(path.to_string()), @@ -204,6 +207,7 @@ pub mod subsystems { pub const NOTIFICATION: MetricSubsystem = MetricSubsystem::Notification; pub const SCANNER: MetricSubsystem = MetricSubsystem::Scanner; pub const COMPRESSION: MetricSubsystem = MetricSubsystem::Compression; + pub const ON_DEMAND_MIGRATION: MetricSubsystem = MetricSubsystem::OnDemandMigration; } #[cfg(test)] @@ -218,6 +222,8 @@ mod tests { assert_eq!(MetricSubsystem::SystemNetworkHost.as_str(), "system_network_host"); assert_eq!(MetricSubsystem::BucketApi.as_str(), "bucket_api"); assert_eq!(MetricSubsystem::ClusterHealth.as_str(), "cluster_health"); + assert_eq!(MetricSubsystem::OnDemandMigration.as_str(), "on_demand_migration"); + assert_eq!(MetricSubsystem::from_path("/on-demand-migration"), MetricSubsystem::OnDemandMigration); // Test custom paths let custom = MetricSubsystem::new("/custom/path-test"); diff --git a/crates/obs/src/metrics/schema/mod.rs b/crates/obs/src/metrics/schema/mod.rs index 4effef7ff..878fad62a 100644 --- a/crates/obs/src/metrics/schema/mod.rs +++ b/crates/obs/src/metrics/schema/mod.rs @@ -28,6 +28,7 @@ pub mod ilm; pub mod node_bucket; pub mod node_disk; pub mod notification_target; +pub mod on_demand_migration; pub mod process_resource; pub mod replication; pub(crate) mod request; diff --git a/crates/obs/src/metrics/schema/on_demand_migration.rs b/crates/obs/src/metrics/schema/on_demand_migration.rs new file mode 100644 index 000000000..62266683c --- /dev/null +++ b/crates/obs/src/metrics/schema/on_demand_migration.rs @@ -0,0 +1,275 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! On-Demand Migration metric descriptors (rustfs/backlog#2157). +//! +//! Every series is bucket-scoped and mirrors one counter of the per-bucket +//! runtime snapshot (`OdmStatsSnapshot` in ecstore). The label value lists +//! below are the runtime's fixed label sets; they are what series retirement +//! enumerates when a bucket's config disappears, so they must stay in sync +//! with the runtime's golden JSON. + +use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems}; +use std::sync::LazyLock; + +/// Bucket the series belongs to. +pub const BUCKET_L: &str = "bucket"; +/// Request operation that entered ODM (`get` | `head`). +pub const OP_L: &str = "op"; +/// How a request that entered ODM ended. +pub const OUTCOME_L: &str = "outcome"; +/// Which pipeline stored a pulled object locally. +pub const PATH_L: &str = "path"; +/// Why a pull did not produce a local object. +pub const REASON_L: &str = "reason"; +/// Upper bound (seconds) of a source latency bucket. +pub const LE_L: &str = "le"; + +/// Fixed `op` label values. +pub const REQUEST_OPS: [&str; 2] = ["get", "head"]; +/// Fixed `outcome` label values; `local_hit` is absent because locally +/// served requests never reach the runtime. +pub const REQUEST_OUTCOMES: [&str; 7] = [ + "source_hit", + "source_miss", + "source_error", + "breaker_open", + "negative_cached", + "filtered", + "unsupported", +]; +/// Fixed `path` label values. +pub const PULL_PATHS: [&str; 3] = ["inline", "background", "backfill"]; +/// Fixed `reason` label values. +pub const PULL_FAILURE_REASONS: [&str; 12] = [ + "source_not_found", + "source_access_denied", + "source_throttled", + "source_timeout", + "source_connect", + "source_server_error", + "source_unsupported", + "source_other", + "etag_mismatch", + "local_write", + "canceled", + "queue_full", +]; +/// Fixed `le` label values of the source latency distribution: the runtime's +/// 14 millisecond bounds rendered in seconds, plus the overflow bucket. +pub const SOURCE_LATENCY_LE: [&str; 15] = [ + "0.005", "0.01", "0.02", "0.05", "0.1", "0.2", "0.5", "1", "2", "5", "10", "20", "30", "60", "+Inf", +]; + +/// `breaker_state` gauge value: the breaker admits every request. +pub const BREAKER_STATE_CLOSED: f64 = 0.0; +/// `breaker_state` gauge value: the breaker admits a single probe. +pub const BREAKER_STATE_HALF_OPEN: f64 = 1.0; +/// `breaker_state` gauge value: the breaker rejects every request. +pub const BREAKER_STATE_OPEN: f64 = 2.0; + +const REQUESTS_TOTAL: &str = "requests_total"; +const PULLED_BYTES_TOTAL: &str = "pulled_bytes_total"; +const PULLED_OBJECTS_TOTAL: &str = "pulled_objects_total"; +const PULL_FAILURES_TOTAL: &str = "pull_failures_total"; +const INFLIGHT_PULLS: &str = "inflight_pulls"; +const QUEUE_DEPTH: &str = "queue_depth"; +const SOURCE_LATENCY_SECONDS_DISTRIBUTION: &str = "source_latency_seconds_distribution"; +const SOURCE_LATENCY_SECONDS_SUM: &str = "source_latency_seconds_sum"; +const SOURCE_LATENCY_SECONDS_COUNT: &str = "source_latency_seconds_count"; +const BREAKER_STATE: &str = "breaker_state"; + +pub static ODM_REQUESTS_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(REQUESTS_TOTAL), + "Total number of requests that entered on-demand migration for a bucket by operation and outcome", + &[BUCKET_L, OP_L, OUTCOME_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_PULLED_BYTES_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(PULLED_BYTES_TOTAL), + "Total number of bytes pulled from the on-demand migration source for a bucket", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_PULLED_OBJECTS_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(PULLED_OBJECTS_TOTAL), + "Total number of objects pulled from the on-demand migration source and stored locally for a bucket by pull path", + &[BUCKET_L, PATH_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_PULL_FAILURES_TOTAL_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(PULL_FAILURES_TOTAL), + "Total number of on-demand migration pulls that did not produce a local object for a bucket by reason", + &[BUCKET_L, REASON_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_INFLIGHT_PULLS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(INFLIGHT_PULLS), + "Current number of on-demand migration pulls holding a pull slot for a bucket", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_QUEUE_DEPTH_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(QUEUE_DEPTH), + "Current number of on-demand migration pulls waiting for a pull slot for a bucket", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +/// Source latency is exported in the cumulative `le` counter layout used by +/// the API TTFB distributions: the runtime only exposes pre-aggregated +/// bucket counts, which the recorder cannot replay as histogram samples. +pub static ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(SOURCE_LATENCY_SECONDS_DISTRIBUTION), + "Cumulative number of on-demand migration source calls for a bucket whose latency was at most le seconds", + &[BUCKET_L, LE_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_SOURCE_LATENCY_SECONDS_SUM_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(SOURCE_LATENCY_SECONDS_SUM), + "Total latency in seconds of on-demand migration source calls for a bucket", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_SOURCE_LATENCY_SECONDS_COUNT_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(SOURCE_LATENCY_SECONDS_COUNT), + "Total number of on-demand migration source calls observed for a bucket", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +pub static ODM_BREAKER_STATE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(BREAKER_STATE), + "State of the on-demand migration source breaker for a bucket: 0 closed, 1 half-open, 2 open", + &[BUCKET_L], + subsystems::ON_DEMAND_MIGRATION, + ) +}); + +// backfill_* descriptors (`listed_total`, `pulled_total`, `skipped_total`, +// `failed_total`, `state`) are added here by ODM-12 (rustfs/backlog#2159). + +#[cfg(test)] +mod tests { + use super::*; + use crate::MetricType; + + fn labels(descriptor: &MetricDescriptor) -> Vec<&str> { + descriptor.variable_labels.iter().map(String::as_str).collect() + } + + #[test] + fn descriptors_use_the_on_demand_migration_prefix() { + for (descriptor, suffix) in [ + (&*ODM_REQUESTS_TOTAL_MD, "requests_total"), + (&*ODM_PULLED_BYTES_TOTAL_MD, "pulled_bytes_total"), + (&*ODM_PULLED_OBJECTS_TOTAL_MD, "pulled_objects_total"), + (&*ODM_PULL_FAILURES_TOTAL_MD, "pull_failures_total"), + (&*ODM_INFLIGHT_PULLS_MD, "inflight_pulls"), + (&*ODM_QUEUE_DEPTH_MD, "queue_depth"), + (&*ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD, "source_latency_seconds_distribution"), + (&*ODM_SOURCE_LATENCY_SECONDS_SUM_MD, "source_latency_seconds_sum"), + (&*ODM_SOURCE_LATENCY_SECONDS_COUNT_MD, "source_latency_seconds_count"), + (&*ODM_BREAKER_STATE_MD, "breaker_state"), + ] { + assert_eq!(descriptor.get_full_metric_name(), format!("rustfs_on_demand_migration_{suffix}")); + assert_eq!(descriptor.subsystem, subsystems::ON_DEMAND_MIGRATION); + assert!(!descriptor.help.is_empty(), "{suffix} needs help text"); + } + } + + #[test] + fn counters_and_gauges_carry_the_documented_label_sets() { + assert_eq!(ODM_REQUESTS_TOTAL_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_REQUESTS_TOTAL_MD), vec!["bucket", "op", "outcome"]); + assert_eq!( + ODM_REQUESTS_TOTAL_MD.help, + "Total number of requests that entered on-demand migration for a bucket by operation and outcome" + ); + + assert_eq!(ODM_PULLED_BYTES_TOTAL_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_PULLED_BYTES_TOTAL_MD), vec!["bucket"]); + + assert_eq!(ODM_PULLED_OBJECTS_TOTAL_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_PULLED_OBJECTS_TOTAL_MD), vec!["bucket", "path"]); + + assert_eq!(ODM_PULL_FAILURES_TOTAL_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_PULL_FAILURES_TOTAL_MD), vec!["bucket", "reason"]); + + assert_eq!(ODM_INFLIGHT_PULLS_MD.metric_type, MetricType::Gauge); + assert_eq!(labels(&ODM_INFLIGHT_PULLS_MD), vec!["bucket"]); + + assert_eq!(ODM_QUEUE_DEPTH_MD.metric_type, MetricType::Gauge); + assert_eq!(labels(&ODM_QUEUE_DEPTH_MD), vec!["bucket"]); + + assert_eq!(ODM_BREAKER_STATE_MD.metric_type, MetricType::Gauge); + assert_eq!(labels(&ODM_BREAKER_STATE_MD), vec!["bucket"]); + assert_eq!( + ODM_BREAKER_STATE_MD.help, + "State of the on-demand migration source breaker for a bucket: 0 closed, 1 half-open, 2 open" + ); + } + + #[test] + fn source_latency_uses_the_counter_bucket_contract() { + assert_eq!(ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_SOURCE_LATENCY_SECONDS_DISTRIBUTION_MD), vec!["bucket", "le"]); + assert_eq!(ODM_SOURCE_LATENCY_SECONDS_SUM_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_SOURCE_LATENCY_SECONDS_SUM_MD), vec!["bucket"]); + assert_eq!(ODM_SOURCE_LATENCY_SECONDS_COUNT_MD.metric_type, MetricType::Counter); + assert_eq!(labels(&ODM_SOURCE_LATENCY_SECONDS_COUNT_MD), vec!["bucket"]); + assert_eq!(SOURCE_LATENCY_LE.len(), 15); + assert_eq!(SOURCE_LATENCY_LE[14], "+Inf"); + } + + #[test] + fn fixed_label_values_are_unique() { + for values in [ + REQUEST_OPS.as_slice(), + REQUEST_OUTCOMES.as_slice(), + PULL_PATHS.as_slice(), + PULL_FAILURE_REASONS.as_slice(), + SOURCE_LATENCY_LE.as_slice(), + ] { + let unique: std::collections::BTreeSet<_> = values.iter().collect(); + assert_eq!(unique.len(), values.len(), "{values:?}"); + } + } +} diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 8ee95ed47..ecc7f7b23 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -26,14 +26,15 @@ use crate::metrics::collectors::{ ClusterHealthStats, ClusterStats, ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats, IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, - ProcessStats, ProcessStatusType, ReplicationMetricsSnapshot, ResourceStats, ScannerRuntimeStats, ScannerStats, + OnDemandMigrationBucketStats, ProcessStats, ProcessStatusType, ReplicationMetricsSnapshot, ResourceStats, + ScannerRuntimeStats, ScannerStats, }; use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot}; use crate::metrics::{ BucketOperations, BucketOptions, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, StorageAdminApi, obs_bucket_replication_stats_snapshot, obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, - obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, + obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, }; use crate::node_identity::current_local_node_identity; use jiff::Timestamp; @@ -661,6 +662,11 @@ pub(crate) async fn collect_bucket_replication_stats_bundle() obs_bucket_replication_stats_bundle().await } +/// Collect per-bucket on-demand migration stats from the global runtime. +pub fn collect_on_demand_migration_stats() -> Vec { + obs_on_demand_migration_snapshot() +} + /// Collect site-level replication stats from the global replication runtime. pub async fn collect_replication_stats() -> ReplicationMetricsSnapshot { obs_site_replication_stats().await diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index 073bfb488..30145f770 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -17,6 +17,10 @@ use std::time::Duration; pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config; +use rustfs_ecstore::api::bucket::on_demand_migration::{ + BreakerState as SourceOdmBreakerState, OdmBucketSnapshot as SourceOdmBucketSnapshot, + OnDemandMigrationSys as SourceOnDemandMigrationSys, +}; use rustfs_ecstore::api::bucket::replication::{ BucketReplicationStats as SourceBucketReplicationStats, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBucketBacklogObservability, RuntimeReplicationTargetBacklog, durable_mrf_backlog_summary_snapshot, @@ -37,6 +41,8 @@ pub(crate) use rustfs_ecstore::api::runtime::{ pub(crate) use rustfs_ecstore::api::storage::ECStore as ObsStore; use rustfs_storage_api as storage_contracts; +use crate::metrics::collectors::{OnDemandMigrationBreakerState, OnDemandMigrationBucketStats}; + #[derive(Debug, Clone, PartialEq)] pub(crate) struct ObsBucketReplicationTargetStatsSnapshot { pub(crate) target_arn: String, @@ -454,6 +460,42 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec OnDemandMigrationBucketStats { + let stats = snapshot.stats; + OnDemandMigrationBucketStats { + bucket: snapshot.bucket, + requests_total: stats.requests_total, + pulled_bytes_total: stats.pulled_bytes_total, + pulled_objects_total: stats.pulled_objects_total, + pull_failures_total: stats.pull_failures_total, + inflight_pulls: stats.inflight_pulls, + queue_depth: stats.queue_depth, + source_latency_buckets: stats + .source_latency + .buckets + .into_iter() + .map(|bucket| (bucket.le_ms, bucket.count)) + .collect(), + source_latency_count: stats.source_latency.count, + source_latency_sum_ms: stats.source_latency.sum_ms, + breaker_state: match stats.breaker_state { + SourceOdmBreakerState::Closed => OnDemandMigrationBreakerState::Closed, + SourceOdmBreakerState::HalfOpen => OnDemandMigrationBreakerState::HalfOpen, + SourceOdmBreakerState::Open => OnDemandMigrationBreakerState::Open, + }, + } +} + +/// Every bucket with live on-demand migration state on this node, sorted by +/// name. Empty while the module switch is off. +pub(crate) fn obs_on_demand_migration_snapshot() -> Vec { + SourceOnDemandMigrationSys::get() + .snapshot() + .into_iter() + .map(on_demand_migration_stats_from_snapshot) + .collect() +} + pub(crate) async fn obs_replication_site_stats_snapshot(current_data_transfer_rate: f64) -> ObsReplicationSiteStatsSnapshot { let Some(stats) = get_global_replication_stats() else { return ObsReplicationSiteStatsSnapshot::default(); @@ -693,6 +735,51 @@ mod tests { assert_eq!(snapshot.mrf_last_flush_duration_millis, 4); } + #[test] + fn on_demand_migration_snapshot_projects_counters_and_breaker_state() { + // Built from JSON: the snapshot's timestamps use `time`, which obs does not depend on. + let snapshot: SourceOdmBucketSnapshot = serde_json::from_value(serde_json::json!({ + "bucket": "photos", + "provider": "minio", + "endpoint_host": "source.example.com", + "applied_at": "2026-09-02T10:00:00Z", + "client_error": null, + "negative_cache_entries": 0, + "inflight_keys": 1, + "max_concurrent_pulls": 8, + "stats": { + "requests_total": {"get": {"source_hit": 2}}, + "pulled_bytes_total": 4096, + "pulled_objects_total": {"inline": 1}, + "pull_failures_total": {"source_timeout": 1}, + "inflight_pulls": 1, + "queue_depth": 2, + "source_latency": { + "buckets": [{"le_ms": 5, "count": 1}, {"le_ms": 10, "count": 2}], + "count": 3, + "sum_ms": 90753 + }, + "last_source_error": {"class": "server_error", "at": "2026-09-02T10:00:00Z"}, + "breaker_state": "open" + } + })) + .expect("runtime snapshot decodes"); + + let stats = on_demand_migration_stats_from_snapshot(snapshot); + + assert_eq!(stats.bucket, "photos"); + assert_eq!(stats.requests_total["get"]["source_hit"], 2); + assert_eq!(stats.pulled_bytes_total, 4096); + assert_eq!(stats.pulled_objects_total["inline"], 1); + assert_eq!(stats.pull_failures_total["source_timeout"], 1); + assert_eq!(stats.inflight_pulls, 1); + assert_eq!(stats.queue_depth, 2); + assert_eq!(stats.source_latency_buckets, vec![(5, 1), (10, 2)]); + assert_eq!(stats.source_latency_count, 3); + assert_eq!(stats.source_latency_sum_ms, 90_753); + assert_eq!(stats.breaker_state, OnDemandMigrationBreakerState::Open); + } + #[test] fn bucket_replication_snapshot_preserves_durable_mrf_unavailable_state() { let snapshot = bucket_replication_stats_snapshot_from_parts( @@ -721,7 +808,7 @@ pub(crate) mod metrics { ObsBucketBandwidthMonitor, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, obs_bucket_replication_stats_snapshot, obs_expiry_state_handle, obs_get_global_bucket_monitor, obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free, obs_is_disk_compression_enabled, - obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot, - obs_resolve_object_store_handle, obs_transition_state_handle, + obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_on_demand_migration_snapshot, + obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle, }; } diff --git a/rustfs/src/admin/handlers/on_demand_migration.rs b/rustfs/src/admin/handlers/on_demand_migration.rs index c6a521ab4..16de974fc 100644 --- a/rustfs/src/admin/handlers/on_demand_migration.rs +++ b/rustfs/src/admin/handlers/on_demand_migration.rs @@ -18,8 +18,9 @@ //! pulled on first access. This module is the management plane only: //! `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}` configure, read and //! clear the source, `?dry-run=true` validates and probes without saving, and -//! `GET .../status` reports the switch state. The data plane, counters and -//! backfill live in later ODM tasks and extend the same routes. +//! `GET .../status` reports the switch state plus this node's runtime +//! snapshot of the bucket (breaker, counters, last source error). The data +//! plane and backfill live in other ODM tasks. //! //! Credentials in the request body are never echoed: every response carries //! the `redacted()` config, probe failures name only the error class, and no @@ -38,7 +39,7 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{ SourceClient, SourceClientSpec, SourceError, SourceProbe, SourceProvider, SourceTimeouts, }; use crate::admin::storage_api::bucket::on_demand_migration::{ - OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, ValidationContext, + OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext, }; use crate::admin::storage_api::bucket::remote_s3_client::{PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError}; use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketOptions}; @@ -54,6 +55,7 @@ use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; use serde::Serialize; +use std::collections::BTreeMap; use std::num::NonZeroU64; use std::sync::Arc; use std::time::Duration; @@ -123,11 +125,138 @@ pub(crate) struct GetBucketOnDemandMigrationResponse { pub updated_at: String, } +/// `GET .../status` body. Field order and `null` handling are pinned by the +/// `rustfs-madmin` fixture. Runtime fields are `null` while the bucket has no +/// live state on this node; `provider` and `endpoint_host` then fall back to +/// the saved config so a disabled module still shows what is configured. #[derive(Debug, Serialize)] pub(crate) struct BucketOnDemandMigrationStatus { pub configured: bool, pub enabled: bool, pub module_enabled: bool, + pub provider: Option, + pub endpoint_host: Option, + pub breaker: Option, + pub counters: Option, + pub last_source_error: Option, + pub inflight_pulls: u64, + pub queue_depth: u64, + /// `source_hit / (source_hit + local GETs)`. The API request metrics + /// count per operation, not per bucket, and the runtime only sees + /// misses, so there is no per-bucket GET total to divide by: this stays + /// `null` rather than reporting a made-up 0. + pub served_by_source_ratio: Option, + /// RFC 3339 save time of the config; `null` when not configured. + pub updated_at: Option, +} + +#[derive(Debug, Serialize)] +pub(crate) struct BreakerStatus { + pub state: &'static str, + /// The runtime snapshot does not carry the breaker's open instant yet + /// (it is a monotonic clock reading inside ecstore), so this is `null`. + pub opened_at: Option, +} + +/// Lifetime counters of the bucket's runtime on this node, keyed by the +/// fixed label values of the Prometheus series with the same names. +#[derive(Debug, Serialize)] +pub(crate) struct RuntimeCounters { + pub requests_total: BTreeMap>, + pub pulled_bytes_total: u64, + pub pulled_objects_total: BTreeMap, + pub pull_failures_total: BTreeMap, + pub source_latency: SourceLatencyStatus, +} + +#[derive(Debug, Serialize)] +pub(crate) struct SourceLatencyStatus { + pub buckets: Vec, + pub count: u64, + pub sum_ms: u64, +} + +#[derive(Debug, Serialize)] +pub(crate) struct LatencyBucketStatus { + pub le_ms: u64, + pub count: u64, +} + +#[derive(Debug, Serialize)] +pub(crate) struct LastSourceErrorStatus { + pub class: String, + pub at: String, +} + +/// Host of the configured source endpoint, matching the runtime's +/// `endpoint_host` so the status reads the same with or without live state. +fn config_endpoint_host(config: &OnDemandMigrationConfig) -> Option { + url::Url::parse(&config.source.effective_endpoint()) + .ok() + .and_then(|url| url.host_str().map(str::to_ascii_lowercase)) +} + +fn bucket_status( + config: Option<(&OnDemandMigrationConfig, OffsetDateTime)>, + runtime: Option, + module_enabled: bool, +) -> S3Result { + let updated_at = config.map(|(_, updated_at)| format_updated_at(updated_at)).transpose()?; + let mut status = BucketOnDemandMigrationStatus { + configured: config.is_some(), + enabled: config.is_some_and(|(config, _)| config.enabled), + module_enabled, + provider: config.map(|(config, _)| config.source.provider.as_str().to_string()), + endpoint_host: config.and_then(|(config, _)| config_endpoint_host(config)), + breaker: None, + counters: None, + last_source_error: None, + inflight_pulls: 0, + queue_depth: 0, + served_by_source_ratio: None, + updated_at, + }; + let Some(runtime) = runtime else { + return Ok(status); + }; + let stats = runtime.stats; + status.provider = Some(runtime.provider); + status.endpoint_host = Some(runtime.endpoint_host); + status.breaker = Some(BreakerStatus { + state: stats.breaker_state.as_str(), + opened_at: None, + }); + status.counters = Some(RuntimeCounters { + requests_total: stats.requests_total, + pulled_bytes_total: stats.pulled_bytes_total, + pulled_objects_total: stats.pulled_objects_total, + pull_failures_total: stats.pull_failures_total, + source_latency: SourceLatencyStatus { + buckets: stats + .source_latency + .buckets + .into_iter() + .map(|bucket| LatencyBucketStatus { + le_ms: bucket.le_ms, + count: bucket.count, + }) + .collect(), + count: stats.source_latency.count, + sum_ms: stats.source_latency.sum_ms, + }, + }); + status.last_source_error = stats + .last_source_error + .map(|error| { + Ok::<_, S3Error>(LastSourceErrorStatus { + class: error.class, + at: format_updated_at(error.at)?, + }) + }) + .transpose()?; + status.inflight_pulls = stats.inflight_pulls; + status.queue_depth = stats.queue_depth; + Ok(status) } pub struct SetBucketOnDemandMigrationHandler; @@ -540,12 +669,13 @@ impl Operation for GetBucketOnDemandMigrationStatusHandler { let config = metadata_sys::get_on_demand_migration_config(&bucket).await.map_err(|err| { admin_s3_error(S3ErrorCode::InternalError, format!("failed to read on-demand migration config: {err}")) })?; + let runtime = OnDemandMigrationSys::get().bucket_snapshot(&bucket); - let status = BucketOnDemandMigrationStatus { - configured: config.is_some(), - enabled: config.is_some_and(|(config, _)| config.enabled), - module_enabled: module_enabled(), - }; + let status = bucket_status( + config.as_ref().map(|(config, updated_at)| (config, *updated_at)), + runtime, + module_enabled(), + )?; admin_json_response(req.uri.path(), &cred.secret_key, StatusCode::OK, &status) } } @@ -633,14 +763,94 @@ mod tests { assert_eq!(serde_json::to_string(&response).expect("serialize"), GET_RESPONSE_FIXTURE.trim()); } + /// The runtime snapshot the ecstore golden test (`snapshot_matches_golden_json`) + /// produces, as this node would hand it to the status route. + fn fixture_runtime_snapshot() -> OdmBucketSnapshot { + let fixture: serde_json::Value = serde_json::from_str(STATUS_FIXTURE.trim()).expect("status fixture parses"); + let counters = &fixture["counters"]; + let snapshot = serde_json::json!({ + "bucket": "photos", + "provider": fixture["provider"], + "endpoint_host": fixture["endpoint_host"], + "applied_at": FIXTURE_UPDATED_AT, + "client_error": null, + "negative_cache_entries": 0, + "inflight_keys": 1, + "max_concurrent_pulls": 8, + "stats": { + "requests_total": counters["requests_total"], + "pulled_bytes_total": counters["pulled_bytes_total"], + "pulled_objects_total": counters["pulled_objects_total"], + "pull_failures_total": counters["pull_failures_total"], + "inflight_pulls": fixture["inflight_pulls"], + "queue_depth": fixture["queue_depth"], + "source_latency": counters["source_latency"], + "last_source_error": fixture["last_source_error"], + "breaker_state": fixture["breaker"]["state"], + } + }); + serde_json::from_value(snapshot).expect("runtime snapshot decodes") + } + #[test] fn status_matches_madmin_golden_fixture() { - let status = BucketOnDemandMigrationStatus { - configured: true, - enabled: true, - module_enabled: false, - }; - assert_eq!(serde_json::to_string(&status).expect("serialize"), STATUS_FIXTURE.trim()); + let config = fixture_config(); + let updated_at = OffsetDateTime::from_unix_timestamp(1_788_343_200).expect("timestamp"); + let status = bucket_status(Some((&config, updated_at)), Some(fixture_runtime_snapshot()), true).expect("status"); + let json = serde_json::to_string(&status).expect("serialize"); + assert_eq!(json, STATUS_FIXTURE.trim()); + assert!(json.contains(r#""served_by_source_ratio":null"#), "the ratio field is present as null"); + } + + #[test] + fn status_without_runtime_state_describes_the_config_and_nulls_the_runtime() { + let config = fixture_config(); + let updated_at = OffsetDateTime::from_unix_timestamp(1_788_343_200).expect("timestamp"); + let status = bucket_status(Some((&config, updated_at)), None, false).expect("status"); + assert_eq!( + serde_json::to_value(&status).expect("serialize"), + serde_json::json!({ + "configured": true, + "enabled": true, + "module_enabled": false, + "provider": "minio", + "endpoint_host": "source.example.com", + "breaker": null, + "counters": null, + "last_source_error": null, + "inflight_pulls": 0, + "queue_depth": 0, + "served_by_source_ratio": null, + "updated_at": FIXTURE_UPDATED_AT, + }) + ); + + let status = bucket_status(None, None, true).expect("status"); + assert_eq!( + serde_json::to_value(&status).expect("serialize"), + serde_json::json!({ + "configured": false, + "enabled": false, + "module_enabled": true, + "provider": null, + "endpoint_host": null, + "breaker": null, + "counters": null, + "last_source_error": null, + "inflight_pulls": 0, + "queue_depth": 0, + "served_by_source_ratio": null, + "updated_at": null, + }) + ); + } + + #[test] + fn config_endpoint_host_matches_the_runtime_host_rule() { + let mut config = fixture_config(); + assert_eq!(config_endpoint_host(&config).as_deref(), Some("source.example.com")); + config.source.endpoint = Some("https://Bucket.S3.Example:9000/base".to_string()); + assert_eq!(config_endpoint_host(&config).as_deref(), Some("bucket.s3.example")); } #[test] @@ -903,6 +1113,12 @@ mod store_tests { Ok(response_json(response).await) } + fn assert_status_switches(status: &Value, configured: bool, enabled: bool, module_enabled: bool) { + assert_eq!(status["configured"], Value::Bool(configured), "{status}"); + assert_eq!(status["enabled"], Value::Bool(enabled), "{status}"); + assert_eq!(status["module_enabled"], Value::Bool(module_enabled), "{status}"); + } + async fn status() -> Value { let router = bucket_router(); let response = GetBucketOnDemandMigrationStatusHandler {} @@ -1057,10 +1273,11 @@ mod store_tests { let err = get_config().await.expect_err("nothing is configured yet"); assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_NO_SUCH_CONFIGURATION.into())); assert_eq!(err.status_code(), Some(StatusCode::NOT_FOUND)); - assert_eq!( - status().await, - serde_json::json!({"configured": false, "enabled": false, "module_enabled": false}) - ); + let body = status().await; + assert_status_switches(&body, false, false, false); + assert_eq!(body["provider"], Value::Null); + assert_eq!(body["counters"], Value::Null); + assert_eq!(body["updated_at"], Value::Null); }) .await; @@ -1147,10 +1364,14 @@ mod store_tests { assert_eq!(body["config"]["source"]["credentials"]["secret_key"], Value::String("REDACTED".into())); assert_eq!(body["config"]["source"]["credentials"]["access_key"], Value::String("AKIASOURCE".into())); assert_eq!(body["updated_at"], Value::String(first_updated_at.clone())); - assert_eq!( - status().await, - serde_json::json!({"configured": true, "enabled": true, "module_enabled": true}) - ); + let body = status().await; + assert_status_switches(&body, true, true, true); + assert_eq!(body["provider"], Value::String("minio".into())); + assert_eq!(body["endpoint_host"], Value::String("127.0.0.1".into())); + assert_eq!(body["updated_at"], Value::String(first_updated_at.clone())); + assert_eq!(body["served_by_source_ratio"], Value::Null, "no per-bucket GET total exists"); + assert_eq!(body["inflight_pulls"], Value::from(0)); + assert_eq!(body["queue_depth"], Value::from(0)); // The peer fan-out ran: the single unreachable peer is reported. let context = crate::admin::runtime_sources::current_app_context(); @@ -1215,10 +1436,11 @@ mod store_tests { let metadata = metadata_sys::get(BUCKET).await.expect("bucket metadata"); assert!(metadata.on_demand_migration_config_json.is_empty()); assert!(metadata.on_demand_migration_config_updated_at > OffsetDateTime::UNIX_EPOCH); - assert_eq!( - status().await, - serde_json::json!({"configured": false, "enabled": false, "module_enabled": true}) - ); + let body = status().await; + assert_status_switches(&body, false, false, true); + assert_eq!(body["provider"], Value::Null); + assert_eq!(body["breaker"], Value::Null); + assert_eq!(body["updated_at"], Value::Null); let response = DeleteBucketOnDemandMigrationHandler {} .call(root_request(Method::DELETE, config_uri(""), Vec::new()), bucket_params(&router)) diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index c423bddae..143dd2714 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -284,8 +284,10 @@ pub(crate) mod durability { } pub(crate) mod on_demand_migration { + pub(crate) type OdmBucketSnapshot = super::ecstore_bucket::on_demand_migration::OdmBucketSnapshot; pub(crate) type OnDemandMigrationConfig = super::ecstore_bucket::on_demand_migration::OnDemandMigrationConfig; pub(crate) type OnDemandMigrationConfigError = super::ecstore_bucket::on_demand_migration::OnDemandMigrationConfigError; + pub(crate) type OnDemandMigrationSys = super::ecstore_bucket::on_demand_migration::OnDemandMigrationSys; pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle; pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider; pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>;