refactor: centralize ecstore data-plane runtime sources (#3797)

crates/ecstore/src/erasure_coding/encode.rs

@@ -18,6 +18,7 @@ use crate::disk::error_reduce::{
 };
 use crate::erasure_coding::BitrotWriterWrapper;
 use crate::erasure_coding::Erasure;
+use crate::runtime_sources;
 use bytes::Bytes;
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
@@ -138,8 +139,7 @@ impl<'a> MultiWriter<'a> {
         if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
             let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
             let summary_text = format_write_quorum_failure(&summary);
-            rustfs_io_metrics::internode_metrics::global_internode_metrics()
-                .record_erasure_write_quorum_failure("write", quorum_dominant_error_metric_label(&summary));
+            runtime_sources::record_erasure_write_quorum_failure("write", quorum_dominant_error_metric_label(&summary));
             error!("reduce_write_quorum_errs: {:?}, {}, errs={:?}", write_err, summary_text, self.errs);
             return Err(std::io::Error::other(format!("Failed to write data: {summary_text}")));
         }
@@ -193,8 +193,7 @@ impl<'a> MultiWriter<'a> {
         if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) {
             let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum);
             let summary_text = format_write_quorum_failure(&summary);
-            rustfs_io_metrics::internode_metrics::global_internode_metrics()
-                .record_erasure_write_quorum_failure("shutdown", quorum_dominant_error_metric_label(&summary));
+            runtime_sources::record_erasure_write_quorum_failure("shutdown", quorum_dominant_error_metric_label(&summary));
             error!(
                 "reduce_write_quorum_errs during shutdown: {:?}, {}, errs={:?}",
                 write_err, summary_text, self.errs

crates/ecstore/src/lib.rs

@@ -40,6 +40,7 @@ mod pools;
 mod rebalance;
 mod rio;
 mod rpc;
+mod runtime_sources;
 mod set_disk;
 mod sets;
 mod storage_api_contracts;

crates/ecstore/src/object_api/readers.rs

@@ -11,7 +11,7 @@ use chacha20poly1305::ChaCha20Poly1305;
 #[cfg(feature = "rio-v2")]
 use hmac::{Hmac, Mac};
 use md5::{Digest, Md5};
-use rustfs_kms::{service_manager::get_global_encryption_service, types::ObjectEncryptionContext};
+use rustfs_kms::types::ObjectEncryptionContext;
 use rustfs_utils::http::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
 use rustfs_utils::path::path_join_buf;
 #[cfg(feature = "rio-v2")]
@@ -1308,7 +1308,7 @@ async fn resolve_managed_material(bucket: &str, object: &str, metadata: &HashMap
     let kms_context: Option<HashMap<String, String>> = None;
     let object_context = build_object_encryption_context(bucket, object, kms_context.as_ref());
 
-    let decrypted_key = if let Some(service) = get_global_encryption_service().await {
+    let decrypted_key = if let Some(service) = crate::runtime_sources::object_encryption_service().await {
         #[cfg(feature = "rio-v2")]
         let data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
             service.decrypt_legacy_data_key(&encrypted_dek).await

crates/ecstore/src/runtime_sources.rs

@@ -0,0 +1,44 @@
+// 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 std::sync::Arc;
+
+use crate::{config::get_global_storage_class, global::get_global_deployment_id};
+use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+use rustfs_kms::{ObjectEncryptionService, get_global_encryption_service};
+
+pub(crate) fn record_erasure_write_quorum_failure(stage: &'static str, dominant_error: &'static str) {
+    global_internode_metrics().record_erasure_write_quorum_failure(stage, dominant_error);
+}
+
+pub(crate) async fn object_encryption_service() -> Option<Arc<ObjectEncryptionService>> {
+    get_global_encryption_service().await
+}
+
+pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize> {
+    get_global_storage_class().and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default()))
+}
+
+pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool {
+    get_global_storage_class().is_some_and(|sc| sc.should_inline(shard_size, versioned))
+}
+
+pub(crate) fn deployment_upload_id(upload_id: &str) -> String {
+    base64_simd::URL_SAFE_NO_PAD
+        .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())
+}
+
+pub(crate) fn global_lock_manager() -> Arc<rustfs_lock::GlobalLockManager> {
+    rustfs_lock::get_global_lock_manager()
+}

crates/ecstore/src/set_disk.rs

@@ -38,13 +38,14 @@ use crate::error::{GenericError, ObjectApiError, is_err_object_not_found};
 use crate::global::{GLOBAL_LocalNodeName, GLOBAL_TierConfigMgr};
 use crate::object_api::ObjectOptions;
 use crate::rpc::heal_bucket_local_on_disks;
+use crate::runtime_sources;
 use crate::store_utils::is_reserved_or_invalid_bucket;
 use crate::{
     bucket::lifecycle::bucket_lifecycle_ops::{
         LifecycleOps, gen_transition_objname, get_transitioned_object_reader, put_restore_opts,
     },
     cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
-    config::{get_global_storage_class, storageclass},
+    config::storageclass,
     disk::{
         CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions,
         RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
@@ -53,7 +54,7 @@ use crate::{
     error::{StorageError, to_object_err},
     // event::name::EventName,
     event_notification::{EventArgs, send_event},
-    global::{GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_deployment_id, is_dist_erasure},
+    global::{GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, is_dist_erasure},
     object_api::{GetObjectReader, ObjectInfo, PutObjReader},
     store_init::{get_format_erasure_in_quorum, load_format_erasure, load_format_erasure_all, save_format_file},
 };
@@ -396,9 +397,7 @@ fn build_tiered_decommission_file_info(
     default_parity_count: usize,
     storage_class: Option<&str>,
 ) -> (FileInfo, usize) {
-    let parity_drives = get_global_storage_class()
-        .and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default()))
-        .unwrap_or(default_parity_count);
+    let parity_drives = runtime_sources::storage_class_parity(storage_class).unwrap_or(default_parity_count);
     let data_drives = disk_count - parity_drives;
     let mut write_quorum = data_drives;
     if data_drives == parity_drives {
@@ -562,7 +561,7 @@ impl SetDisks {
             set_endpoints,
             disk_health_cache: Arc::new(RwLock::new(Vec::new())),
             lockers,
-            local_lock_manager: rustfs_lock::get_global_lock_manager(),
+            local_lock_manager: runtime_sources::global_lock_manager(),
         })
     }
 
@@ -1028,13 +1027,7 @@ impl rustfs_storage_api::ObjectIO for SetDisks {
                 user_defined.insert(key.clone(), value.clone());
             }
         }
-        let sc_parity_drives = {
-            if let Some(sc) = get_global_storage_class() {
-                sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str())
-            } else {
-                None
-            }
-        };
+        let sc_parity_drives = runtime_sources::storage_class_parity(user_defined.get(AMZ_STORAGE_CLASS).map(String::as_str));
 
         let mut parity_drives = sc_parity_drives.unwrap_or(self.default_parity_count);
         if opts.max_parity {
@@ -1082,13 +1075,8 @@ impl rustfs_storage_api::ObjectIO for SetDisks {
         let result: Result<ObjectInfo> = async {
             let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
 
-            let is_inline_buffer = {
-                if let Some(sc) = get_global_storage_class() {
-                    sc.should_inline(erasure.shard_file_size(data.size()), opts.versioned)
-                } else {
-                    false
-                }
-            };
+            let is_inline_buffer =
+                runtime_sources::storage_class_should_inline(erasure.shard_file_size(data.size()), opts.versioned);
 
             let shard_file_size = erasure.shard_file_size(data.size());
             let shard_size = erasure.shard_size();
@@ -3729,8 +3717,7 @@ impl rustfs_storage_api::MultipartOperations for SetDisks {
             uploads.push(MultipartInfo {
                 bucket: bucket.to_owned(),
                 object: object.to_owned(),
-                upload_id: base64_simd::URL_SAFE_NO_PAD
-                    .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes()),
+                upload_id: runtime_sources::deployment_upload_id(&upload_id),
                 initiated: Some(start_time),
                 ..Default::default()
             });
@@ -3820,13 +3807,7 @@ impl rustfs_storage_api::MultipartOperations for SetDisks {
             let _ = user_defined.remove(AMZ_STORAGE_CLASS);
         }
 
-        let sc_parity_drives = {
-            if let Some(sc) = get_global_storage_class() {
-                sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str())
-            } else {
-                None
-            }
-        };
+        let sc_parity_drives = runtime_sources::storage_class_parity(user_defined.get(AMZ_STORAGE_CLASS).map(String::as_str));
 
         let mut parity_drives = sc_parity_drives.unwrap_or(self.default_parity_count);
         if opts.max_parity {
@@ -3897,8 +3878,7 @@ impl rustfs_storage_api::MultipartOperations for SetDisks {
 
         let upload_uuid = format!("{}x{}", Uuid::new_v4(), mod_time.unix_timestamp_nanos());
 
-        let upload_id = base64_simd::URL_SAFE_NO_PAD
-            .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_uuid).as_bytes());
+        let upload_id = runtime_sources::deployment_upload_id(&upload_uuid);
 
         let upload_path = Self::get_upload_id_dir(bucket, object, upload_uuid.as_str());
 

crates/ecstore/src/set_disk/heal.rs

@@ -414,13 +414,10 @@ impl SetDisks {
                                     }
                                 }
 
-                                let is_inline_buffer = {
-                                    if let Some(sc) = get_global_storage_class() {
-                                        sc.should_inline(erasure.shard_file_size(latest_meta.size), false)
-                                    } else {
-                                        false
-                                    }
-                                };
+                                let is_inline_buffer = runtime_sources::storage_class_should_inline(
+                                    erasure.shard_file_size(latest_meta.size),
+                                    false,
+                                );
                                 // create writers for all disk positions, but only for outdated disks
                                 for (index, disk_op) in out_dated_disks.iter().enumerate() {
                                     if let Some(outdated_disk) = disk_op {