refactor: centralize owner server config reads (#3793)

crates/obs/src/metrics/mod.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
     GLOBAL_ExpiryState as OBS_GLOBAL_EXPIRY_STATE, GLOBAL_TransitionState as OBS_GLOBAL_TRANSITION_STATE,
 };
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config;
-pub(crate) use rustfs_ecstore::api::bucket::replication::GLOBAL_REPLICATION_STATS as OBS_GLOBAL_REPLICATION_STATS;
+pub(crate) use rustfs_ecstore::api::bucket::replication::{
+    GLOBAL_REPLICATION_STATS as OBS_GLOBAL_REPLICATION_STATS, ReplicationStats as ObsReplicationStats,
+};
 pub(crate) use rustfs_ecstore::api::capacity::{
     get_total_usable_capacity as obs_get_total_usable_capacity,
     get_total_usable_capacity_free as obs_get_total_usable_capacity_free,
@@ -31,6 +34,7 @@ pub(crate) use rustfs_ecstore::api::storage::ECStore as ObsStore;
 pub mod collectors;
 pub mod config;
 pub mod report;
+mod runtime_sources;
 pub mod scheduler;
 pub mod schema;
 pub mod stats_collector;

crates/obs/src/metrics/runtime_sources.rs

@@ -0,0 +1,102 @@
+// 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.
+
+use crate::metrics::{
+    OBS_GLOBAL_EXPIRY_STATE, OBS_GLOBAL_REPLICATION_STATS, OBS_GLOBAL_TRANSITION_STATE, ObsBucketBandwidthMonitor,
+    ObsReplicationStats, obs_get_global_bucket_monitor,
+};
+use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot};
+use std::sync::Arc;
+
+pub(crate) type ObsReplicationStatsHandle = Arc<ObsReplicationStats>;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct ObsIlmRuntimeSnapshot {
+    pub(crate) expiry_pending_tasks: u64,
+    pub(crate) transition_active_tasks: u64,
+    pub(crate) transition_pending_tasks: u64,
+    pub(crate) transition_missed_immediate_tasks: u64,
+    pub(crate) transition_queue_full_tasks: u64,
+    pub(crate) transition_queue_send_timeout_tasks: u64,
+    pub(crate) transition_compensation_scheduled_tasks: u64,
+    pub(crate) transition_compensation_running_tasks: u64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct ObsIamMetricsSnapshot {
+    pub(crate) last_sync_duration_millis: u64,
+    pub(crate) plugin_authn_service_failed_requests_minute: u64,
+    pub(crate) plugin_authn_service_last_fail_seconds: u64,
+    pub(crate) plugin_authn_service_last_succ_seconds: u64,
+    pub(crate) plugin_authn_service_succ_avg_rtt_ms_minute: u64,
+    pub(crate) plugin_authn_service_succ_max_rtt_ms_minute: u64,
+    pub(crate) plugin_authn_service_total_requests_minute: u64,
+    pub(crate) since_last_sync_millis: u64,
+    pub(crate) sync_failures: u64,
+    pub(crate) sync_successes: u64,
+}
+
+fn usize_to_u64_saturating(value: usize) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
+fn i64_to_u64_floor_zero(value: i64) -> u64 {
+    u64::try_from(value.max(0)).unwrap_or_default()
+}
+
+pub(crate) fn bucket_monitor_handle() -> Option<Arc<ObsBucketBandwidthMonitor>> {
+    obs_get_global_bucket_monitor()
+}
+
+pub(crate) fn bucket_monitor_available() -> bool {
+    bucket_monitor_handle().is_some()
+}
+
+pub(crate) fn replication_stats_handle() -> Option<ObsReplicationStatsHandle> {
+    OBS_GLOBAL_REPLICATION_STATS.get().cloned()
+}
+
+pub(crate) async fn ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot {
+    ObsIlmRuntimeSnapshot {
+        expiry_pending_tasks: usize_to_u64_saturating(OBS_GLOBAL_EXPIRY_STATE.read().await.pending_tasks()),
+        transition_active_tasks: i64_to_u64_floor_zero(OBS_GLOBAL_TRANSITION_STATE.active_tasks()),
+        transition_pending_tasks: usize_to_u64_saturating(OBS_GLOBAL_TRANSITION_STATE.pending_tasks()),
+        transition_missed_immediate_tasks: i64_to_u64_floor_zero(OBS_GLOBAL_TRANSITION_STATE.missed_immediate_tasks()),
+        transition_queue_full_tasks: i64_to_u64_floor_zero(OBS_GLOBAL_TRANSITION_STATE.queue_full_tasks()),
+        transition_queue_send_timeout_tasks: i64_to_u64_floor_zero(OBS_GLOBAL_TRANSITION_STATE.queue_send_timeout_tasks()),
+        transition_compensation_scheduled_tasks: i64_to_u64_floor_zero(
+            OBS_GLOBAL_TRANSITION_STATE.compensation_scheduled_tasks(),
+        ),
+        transition_compensation_running_tasks: i64_to_u64_floor_zero(OBS_GLOBAL_TRANSITION_STATE.compensation_running_tasks()),
+    }
+}
+
+pub(crate) fn iam_metrics_snapshot() -> Option<ObsIamMetricsSnapshot> {
+    let iam_sys = get_global_iam_sys()?;
+    let sync = iam_sys.sync_metrics_snapshot();
+    let oidc = oidc_plugin_authn_metrics_snapshot();
+
+    Some(ObsIamMetricsSnapshot {
+        last_sync_duration_millis: sync.last_sync_duration_millis,
+        plugin_authn_service_failed_requests_minute: oidc.failed_requests_minute,
+        plugin_authn_service_last_fail_seconds: oidc.last_fail_seconds,
+        plugin_authn_service_last_succ_seconds: oidc.last_succ_seconds,
+        plugin_authn_service_succ_avg_rtt_ms_minute: oidc.succ_avg_rtt_ms_minute,
+        plugin_authn_service_succ_max_rtt_ms_minute: oidc.succ_max_rtt_ms_minute,
+        plugin_authn_service_total_requests_minute: oidc.total_requests_minute,
+        since_last_sync_millis: sync.since_last_sync_millis,
+        sync_failures: sync.sync_failures,
+        sync_successes: sync.sync_successes,
+    })
+}

crates/obs/src/metrics/scheduler.rs

@@ -70,8 +70,8 @@ use crate::metrics::config::{
     ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, ENV_CLUSTER_METRICS_INTERVAL, ENV_DEFAULT_METRICS_INTERVAL,
     ENV_NODE_METRICS_INTERVAL, ENV_NOTIFICATION_METRICS_INTERVAL, ENV_RESOURCE_METRICS_INTERVAL,
 };
-use crate::metrics::obs_get_global_bucket_monitor;
 use crate::metrics::report::{PrometheusMetric, report_metrics};
+use crate::metrics::runtime_sources::bucket_monitor_available;
 use crate::metrics::schema::bucket_replication::{
     BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, TARGET_ARN_L,
 };
@@ -599,7 +599,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
         loop {
             tokio::select! {
                 _ = interval.tick() => {
-                    let monitor_available = obs_get_global_bucket_monitor().is_some();
+                    let monitor_available = bucket_monitor_available();
                     let stats = collect_bucket_replication_bandwidth_stats();
 
                     let current_live_keys = repl_bw_live_keys(&stats);

crates/obs/src/metrics/stats_collector.rs

@@ -26,15 +26,16 @@ use crate::metrics::collectors::{
     DriveDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats,
     ProcessStatusType, ReplicationStats, ResourceStats, ScannerStats,
 };
+use crate::metrics::runtime_sources::{
+    ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot, replication_stats_handle,
+};
 use crate::metrics::{
-    OBS_GLOBAL_EXPIRY_STATE, OBS_GLOBAL_REPLICATION_STATS, OBS_GLOBAL_TRANSITION_STATE, ObsEcstoreResult, ObsStore,
-    obs_get_global_bucket_monitor, obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free,
+    ObsEcstoreResult, ObsStore, obs_get_quota_config, obs_get_total_usable_capacity, obs_get_total_usable_capacity_free,
     obs_load_data_usage_from_backend, obs_resolve_object_store_handle,
 };
 use chrono::Utc;
 use rustfs_common::heal_channel::HealScanMode;
 use rustfs_common::metrics::global_metrics;
-use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot};
 use rustfs_io_metrics::internode_metrics::global_internode_metrics;
 use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system};
 use rustfs_storage_api::{BucketOperations, BucketOptions, StorageAdminApi};
@@ -74,18 +75,6 @@ struct ObsBucketReplicationBandwidthStats {
     current_bandwidth_bytes_per_sec: f64,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-struct ObsIlmRuntimeSnapshot {
-    expiry_pending_tasks: u64,
-    transition_active_tasks: u64,
-    transition_pending_tasks: u64,
-    transition_missed_immediate_tasks: u64,
-    transition_queue_full_tasks: u64,
-    transition_queue_send_timeout_tasks: u64,
-    transition_compensation_scheduled_tasks: u64,
-    transition_compensation_running_tasks: u64,
-}
-
 fn usize_to_u64_saturating(value: usize) -> u64 {
     u64::try_from(value).unwrap_or(u64::MAX)
 }
@@ -148,7 +137,7 @@ async fn obs_bucket_quota_limit_bytes(bucket: &str) -> u64 {
 }
 
 fn obs_bucket_replication_bandwidth_stats() -> Option<Vec<ObsBucketReplicationBandwidthStats>> {
-    let monitor = obs_get_global_bucket_monitor()?;
+    let monitor = bucket_monitor_handle()?;
     Some(
         monitor
             .get_report(|_| true)
@@ -165,26 +154,11 @@ fn obs_bucket_replication_bandwidth_stats() -> Option<Vec<ObsBucketReplicationBa
 }
 
 async fn obs_ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot {
-    let expiry_pending_tasks = {
-        let expiry = OBS_GLOBAL_EXPIRY_STATE.read().await;
-        usize_to_u64_saturating(expiry.pending_tasks())
-    };
-    let transition = &OBS_GLOBAL_TRANSITION_STATE;
-
-    ObsIlmRuntimeSnapshot {
-        expiry_pending_tasks,
-        transition_active_tasks: i64_to_u64_floor_zero(transition.active_tasks()),
-        transition_pending_tasks: usize_to_u64_saturating(transition.pending_tasks()),
-        transition_missed_immediate_tasks: i64_to_u64_floor_zero(transition.missed_immediate_tasks()),
-        transition_queue_full_tasks: i64_to_u64_floor_zero(transition.queue_full_tasks()),
-        transition_queue_send_timeout_tasks: i64_to_u64_floor_zero(transition.queue_send_timeout_tasks()),
-        transition_compensation_scheduled_tasks: i64_to_u64_floor_zero(transition.compensation_scheduled_tasks()),
-        transition_compensation_running_tasks: i64_to_u64_floor_zero(transition.compensation_running_tasks()),
-    }
+    ilm_runtime_snapshot().await
 }
 
 async fn obs_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
-    let Some(stats) = OBS_GLOBAL_REPLICATION_STATS.get() else {
+    let Some(stats) = replication_stats_handle() else {
         return Vec::new();
     };
 
@@ -256,7 +230,7 @@ async fn obs_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
 }
 
 async fn obs_site_replication_stats() -> ReplicationStats {
-    let Some(stats) = OBS_GLOBAL_REPLICATION_STATS.get() else {
+    let Some(stats) = replication_stats_handle() else {
         return ReplicationStats::default();
     };
 
@@ -904,21 +878,19 @@ pub async fn collect_erasure_set_stats() -> Vec<ErasureSetStats> {
 }
 
 pub async fn collect_iam_stats() -> Option<IamStats> {
-    let iam_sys = get_global_iam_sys()?;
-    let sync = iam_sys.sync_metrics_snapshot();
-    let oidc = oidc_plugin_authn_metrics_snapshot();
+    let snapshot = iam_metrics_snapshot()?;
 
     Some(IamStats {
-        last_sync_duration_millis: sync.last_sync_duration_millis,
-        plugin_authn_service_failed_requests_minute: oidc.failed_requests_minute,
-        plugin_authn_service_last_fail_seconds: oidc.last_fail_seconds,
-        plugin_authn_service_last_succ_seconds: oidc.last_succ_seconds,
-        plugin_authn_service_succ_avg_rtt_ms_minute: oidc.succ_avg_rtt_ms_minute,
-        plugin_authn_service_succ_max_rtt_ms_minute: oidc.succ_max_rtt_ms_minute,
-        plugin_authn_service_total_requests_minute: oidc.total_requests_minute,
-        since_last_sync_millis: sync.since_last_sync_millis,
-        sync_failures: sync.sync_failures,
-        sync_successes: sync.sync_successes,
+        last_sync_duration_millis: snapshot.last_sync_duration_millis,
+        plugin_authn_service_failed_requests_minute: snapshot.plugin_authn_service_failed_requests_minute,
+        plugin_authn_service_last_fail_seconds: snapshot.plugin_authn_service_last_fail_seconds,
+        plugin_authn_service_last_succ_seconds: snapshot.plugin_authn_service_last_succ_seconds,
+        plugin_authn_service_succ_avg_rtt_ms_minute: snapshot.plugin_authn_service_succ_avg_rtt_ms_minute,
+        plugin_authn_service_succ_max_rtt_ms_minute: snapshot.plugin_authn_service_succ_max_rtt_ms_minute,
+        plugin_authn_service_total_requests_minute: snapshot.plugin_authn_service_total_requests_minute,
+        since_last_sync_millis: snapshot.since_last_sync_millis,
+        sync_failures: snapshot.sync_failures,
+        sync_successes: snapshot.sync_successes,
     })
 }