From 8fc1c9281e80487382af00373fa33e08ca835fff Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 6 Sep 2026 01:31:58 +0800 Subject: [PATCH] refactor(odm): move migration orchestration into application (#7226) * refactor(odm): move migration orchestration into application * style(odm): format relocated listing test imports --- Cargo.lock | 4 +- crates/ecstore/Cargo.toml | 1 - crates/ecstore/src/api/mod.rs | 72 ++------ crates/ecstore/src/bucket/metadata.rs | 89 +++------- crates/ecstore/src/bucket/metadata_sys.rs | 152 ++++++---------- crates/ecstore/src/bucket/mod.rs | 1 - crates/ecstore/src/bucket/remote_s3_client.rs | 6 +- crates/obs/src/metrics/mod.rs | 1 + crates/obs/src/metrics/storage_api.rs | 168 ++++++------------ .../background-services-inventory.md | 6 +- docs/architecture/crate-boundaries.md | 25 +++ .../ecstore-api-facade-inventory.md | 7 + .../remote-credential-sealing-adr.md | 8 +- docs/operations/on-demand-migration.md | 6 +- rustfs/Cargo.toml | 5 +- .../src/admin/handlers/on_demand_migration.rs | 39 ++-- rustfs/src/admin/storage_api.rs | 40 +---- rustfs/src/app/bucket_list_through.rs | 18 +- rustfs/src/app/object/get.rs | 8 +- rustfs/src/app/object/head.rs | 12 +- .../src/app/object/on_demand_migration_put.rs | 6 +- rustfs/src/app/object/shared.rs | 4 +- rustfs/src/app/storage_api.rs | 33 ---- rustfs/src/lib.rs | 1 + .../src}/on_demand_migration/azure.rs | 10 +- .../on_demand_migration/backend_contract.rs | 2 +- .../src}/on_demand_migration/backfill.rs | 32 ++-- .../src}/on_demand_migration/breaker.rs | 0 .../src}/on_demand_migration/config.rs | 83 +++++++-- .../src}/on_demand_migration/gcs.rs | 8 +- .../src}/on_demand_migration/list_through.rs | 0 rustfs/src/on_demand_migration/metrics.rs | 147 +++++++++++++++ .../src}/on_demand_migration/mod.rs | 15 +- .../src}/on_demand_migration/native_http.rs | 2 +- .../on_demand_migration/negative_cache.rs | 0 .../src}/on_demand_migration/pull.rs | 4 +- .../src}/on_demand_migration/source_client.rs | 6 +- .../src}/on_demand_migration/stats.rs | 0 rustfs/src/on_demand_migration/storage_api.rs | 28 +++ .../src}/on_demand_migration/sys.rs | 51 +++++- .../on_demand_migration/test_http_fixture.rs | 0 rustfs/src/startup_background.rs | 7 +- rustfs/src/startup_bucket_metadata.rs | 6 +- rustfs/src/storage/storage_api.rs | 14 +- rustfs/src/storage_api.rs | 32 +++- 45 files changed, 619 insertions(+), 540 deletions(-) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/azure.rs (99%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/backend_contract.rs (99%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/backfill.rs (98%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/breaker.rs (100%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/config.rs (94%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/gcs.rs (98%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/list_through.rs (100%) create mode 100644 rustfs/src/on_demand_migration/metrics.rs rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/mod.rs (84%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/native_http.rs (99%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/negative_cache.rs (100%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/pull.rs (99%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/source_client.rs (99%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/stats.rs (100%) create mode 100644 rustfs/src/on_demand_migration/storage_api.rs rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/sys.rs (96%) rename {crates/ecstore/src/bucket => rustfs/src}/on_demand_migration/test_http_fixture.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index 76072f5da..c6d279229 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9493,6 +9493,8 @@ dependencies = [ "atomic_enum", "aws-config", "aws-sdk-s3", + "aws-smithy-runtime-api", + "aws-smithy-types", "axum", "base64-simd", "bytes", @@ -9506,6 +9508,7 @@ dependencies = [ "futures", "futures-lite", "futures-util", + "google-cloud-auth", "hashbrown 0.17.1", "hex-simd", "hmac 0.13.0", @@ -9792,7 +9795,6 @@ dependencies = [ "path-absolutize", "pin-project-lite", "proptest", - "quick-xml", "rand 0.10.2", "ratelimit", "rcgen", diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 8dbfec7c5..4925b283e 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -216,7 +216,6 @@ serde_urlencoded.workspace = true google-cloud-storage = { workspace = true, optional = true } google-cloud-auth = { workspace = true, optional = true } faster-hex = { workspace = true } -quick-xml = { workspace = true } ratelimit = { workspace = true } aws-smithy-http-client = { workspace = true, default-features = false, features = ["rustls-aws-lc"] } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index e752713ff..b19397a7f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -146,69 +146,23 @@ pub mod bucket { }; } - pub mod on_demand_migration { - pub use crate::bucket::on_demand_migration::{ - ApplyOutcome, BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, - Breaker, BreakerState, BreakerTransition, BreakerVerdict, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, GaugeGuard, - LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup, - OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason, - PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS, - SourceLatencySnapshot, source_backend_spec, source_client_spec, - }; - pub use crate::bucket::on_demand_migration::{ - AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, - ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, - Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, - ValidationContext, - }; - pub use crate::bucket::on_demand_migration::{ - EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS, - PullCompletion, PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, SourceIdleGuard, WriteBackBody, - WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, - idle_guarded_body, - }; - pub use crate::bucket::on_demand_migration::{ - FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, - ListThroughToken, ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MAX_LIST_NO_PROGRESS_PAGES, MergeOutcome, - MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter, - decode_continuation_token, source_list_plan, - }; - pub mod backfill { - pub use crate::bucket::on_demand_migration::backfill::{ - BACKFILL_CHECKPOINT_FILE, BACKFILL_CHECKPOINT_FORMAT_VERSION, BACKFILL_FAILED_KEYS_CAPACITY, BACKFILL_LEASE, - BACKFILL_LEASE_LOCK_PREFIX, BACKFILL_LIST_PAGE_SIZE, BACKFILL_RECOVERY_INTERVAL, BACKFILL_SAVE_EVERY_KEYS, - BACKFILL_SAVE_INTERVAL, BackfillCheckpoint, BackfillContext, BackfillContextFactory, BackfillError, - BackfillLastError, BackfillOwner, BackfillRecoveryStats, BackfillRequest, BackfillRunner, BackfillState, - BucketBackfillContext, LocalBackfillObject, PriorityPullPermits, PullPermit, PullPriority, SkipExisting, - StoredCheckpoint, SysBackfillContexts, global_backfill_runner, install_global_backfill_runner, key_hash, - read_checkpoint, run_backfill_recovery_loop, spawn_backfill_recovery_loop, - }; - } - pub mod source_client { - pub use crate::bucket::on_demand_migration::source_client::{ - AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError, - SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse, - SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style, - }; - } - } - pub mod metadata_sys { - #[cfg(feature = "test-util")] - pub use crate::bucket::metadata_sys::ConfigWriteLockProbe; pub use crate::bucket::metadata_sys::{ - BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock, + BUCKET_CONFIG_PUBLISH_HOOK, BucketConfigPublishHook, BucketMetadataMutationGuard, BucketMetadataSys, + ObjectLockConfigState, acquire_bucket_metadata_transaction_lock, acquire_bucket_metadata_transaction_lock_for_incarnation, acquire_scanner_bucket_incarnation_fence, capture_bucket_metadata_incarnation, delete, delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config, get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, - get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, - get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, - reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, update, - update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_quota_if_incarnation, - update_under_transaction_lock, + get_on_demand_migration_config_in, get_public_access_block_config, get_quota_config, get_replication_config, + get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config, + init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, + update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, + update_quota_if_incarnation, update_under_transaction_lock, }; + #[cfg(feature = "test-util")] + pub use crate::bucket::metadata_sys::{ConfigWriteLockProbe, test_support}; } pub mod migration { @@ -251,7 +205,7 @@ pub mod bucket { pub mod remote_s3_client { pub use crate::bucket::remote_s3_client::{ PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_client, - validate_remote_endpoint, + build_remote_s3_config, validate_remote_endpoint, validate_target_ca_pem, }; } @@ -497,9 +451,9 @@ pub mod object { ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission, RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitStartError, - ScannerPublicationCommitState, StreamConsumer, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, - register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook, - unregister_object_mutation_hook, + ScannerPublicationCommitState, StreamConsumer, WriteCompletion, get_object_body_cache_plaintext_len, + lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook, + unregister_get_object_body_cache_hook, unregister_object_mutation_hook, }; pub use crate::store::{ PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError, diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index 41b2afdc5..09030ba12 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -489,28 +489,17 @@ impl BucketMetadata { !self.bucket_targets_config_json.is_empty() && self.bucket_target_config.is_none() } - /// Parsed per-bucket durability override, if a valid one is stored. - /// - /// Absent/empty/unparsable payloads all mean "no override" (the bucket - /// follows the global durability mode); a parse failure is logged so a - /// corrupted entry cannot silently change fsync behavior. - /// Parsed on-demand migration config, if one is stored. - /// - /// `Ok(None)` means no config (absent or cleared). A stored payload that - /// does not parse is an error, never a default: the runtime must not - /// pull from a source it cannot describe. - pub fn on_demand_migration_config( - &self, - ) -> std::result::Result< - Option, - super::on_demand_migration::OnDemandMigrationConfigError, - > { - if self.on_demand_migration_config_json.is_empty() { - return Ok(None); - } - super::on_demand_migration::OnDemandMigrationConfig::from_json(&self.on_demand_migration_config_json).map(Some) + /// Opaque application-owned configuration with its persisted update time. + /// Empty bytes mean absent or cleared; decoding belongs to the consumer. + pub fn on_demand_migration_config(&self) -> Option<(&[u8], OffsetDateTime)> { + (!self.on_demand_migration_config_json.is_empty()).then_some(( + self.on_demand_migration_config_json.as_slice(), + self.on_demand_migration_config_updated_at, + )) } + /// Parsed per-bucket durability override, if a valid one is stored. + /// Invalid payloads follow the global mode after logging a parse failure. pub fn durability_config(&self) -> Option { if self.durability_config_json.is_empty() { return None; @@ -916,13 +905,6 @@ impl BucketMetadata { self.durability_config_updated_at = updated; } BUCKET_ON_DEMAND_MIGRATION_CONFIG => { - // Structural check only (shape, unknown fields); the - // deployment-relative rules run in the admin handler with a - // `ValidationContext`. A blob this build cannot read must not - // be persisted for every later reader to trip over. - if !data.is_empty() { - super::on_demand_migration::OnDemandMigrationConfig::from_json(&data).map_err(Error::other)?; - } self.on_demand_migration_config_json = data; self.on_demand_migration_config_updated_at = updated; } @@ -1978,51 +1960,30 @@ mod test { const ODM_JSON: &[u8] = br#"{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#; - /// rustfs/backlog#2148: the on-demand migration config is a RustFS - /// extension entry that round-trips through `update_config` and the - /// msgpack codec, clears on delete, and never parses corruption into a - /// default. + /// The metadata codec preserves application-owned bytes and timestamps. #[test] fn on_demand_migration_config_round_trips_and_tracks_updates() { - use crate::bucket::on_demand_migration::{OnDemandMigrationConfig, OnDemandMigrationConfigError}; - let mut bm = BucketMetadata::new("odm-bucket"); - assert_eq!(bm.on_demand_migration_config(), Ok(None), "fresh metadata carries no config"); - - let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap(); + assert_eq!(bm.on_demand_migration_config(), None, "fresh metadata carries no config"); bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec()) - .expect("valid config is accepted"); - assert_ne!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH); - assert_eq!(bm.on_demand_migration_config(), Ok(Some(expected.clone()))); - + .expect("opaque config is accepted"); + let stamped = bm.on_demand_migration_config_updated_at; + assert_ne!(stamped, OffsetDateTime::UNIX_EPOCH); + assert_eq!(bm.on_demand_migration_config(), Some((ODM_JSON, stamped))); let back = BucketMetadata::unmarshal(&bm.marshal_msg().unwrap()).unwrap(); assert_eq!(back.on_demand_migration_config_json, bm.on_demand_migration_config_json); - assert_eq!( - back.on_demand_migration_config_updated_at.unix_timestamp(), - bm.on_demand_migration_config_updated_at.unix_timestamp() - ); - assert_eq!(back.on_demand_migration_config(), Ok(Some(expected))); - - // A blob this build cannot read is rejected at the write boundary - // rather than persisted for every reader to trip over. - let before = bm.on_demand_migration_config_json.clone(); - assert!( - bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec()) - .is_err() - ); - assert_eq!(bm.on_demand_migration_config_json, before, "a rejected update leaves the blob untouched"); - - // Delete clears the entry. - let stamped = bm.on_demand_migration_config_updated_at; + assert_eq!(back.on_demand_migration_config_updated_at.unix_timestamp(), stamped.unix_timestamp()); bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, Vec::new()).unwrap(); assert!(bm.on_demand_migration_config_json.is_empty()); - assert_eq!(bm.on_demand_migration_config(), Ok(None)); + assert_eq!(bm.on_demand_migration_config(), None); assert!(bm.on_demand_migration_config_updated_at >= stamped); - - // Corruption that bypassed `update_config` (disk, another writer) - // is a typed error, never a default. - bm.on_demand_migration_config_json = b"not-json".to_vec(); - assert!(matches!(bm.on_demand_migration_config(), Err(OnDemandMigrationConfigError::Malformed(_)))); + bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, b"not-json".to_vec()) + .unwrap(); + let back = BucketMetadata::unmarshal(&bm.marshal_msg().unwrap()).unwrap(); + assert_eq!( + back.on_demand_migration_config_json, b"not-json", + "metadata must not reinterpret application bytes" + ); } /// rustfs/backlog#2148: a `.metadata.bin` written before the on-demand @@ -2034,7 +1995,7 @@ mod test { let mut bm = BucketMetadata::unmarshal(&blob[4..]).expect("unmarshal MinIO bucket metadata"); assert!(bm.on_demand_migration_config_json.is_empty()); assert_eq!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH); - assert_eq!(bm.on_demand_migration_config(), Ok(None)); + assert_eq!(bm.on_demand_migration_config(), None); bm.default_timestamps(); assert_ne!(bm.created, OffsetDateTime::UNIX_EPOCH, "fixture must carry a real creation time"); diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 4a53927f9..5e30841ad 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -19,7 +19,6 @@ use super::quota::BucketQuota; use super::target::BucketTargets; use crate::bucket::bucket_target_sys::BucketTargetSys; use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_parse_with_presence}; -use crate::bucket::on_demand_migration::{ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig}; use crate::bucket::utils::is_meta_bucketname; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result, is_err_bucket_not_found, is_err_strict_volume_not_found}; @@ -49,6 +48,11 @@ use tokio_util::sync::CancellationToken; use tracing::{error, warn}; use uuid::Uuid; +/// Opaque bucket configuration notifications for application-owned services. +/// `None` withdraws a configuration; consumers validate nonempty bytes. +pub type BucketConfigPublishHook = Box) + Send + Sync>; +pub static BUCKET_CONFIG_PUBLISH_HOOK: std::sync::OnceLock = std::sync::OnceLock::new(); + const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60); #[cfg(any(test, feature = "test-util"))] @@ -395,39 +399,20 @@ fn clear_bucket_durability(bucket: &str) { crate::disk::local::bucket_durability::set(bucket, None); } -/// Publish the bucket's on-demand migration config (or its absence) to the -/// runtime registered in `ON_DEMAND_MIGRATION_CONFIG_HOOK`. -/// -/// Called from the same five cache-install paths as -/// [`sync_bucket_durability`]. A stored payload this build cannot parse is -/// published as `None`: the runtime must stop pulling for that bucket rather -/// than keep an older config or guess. +/// Publish application-owned bytes on every cache install path. fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) { - let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() else { - return; - }; - match bm.on_demand_migration_config() { - Ok(config) => hook(bucket, config.as_ref()), - Err(err) => { - warn!( - event = "bucket_metadata_parse_failed", - component = "ecstore", - subsystem = "bucket_metadata", - bucket = %bucket, - config = "on_demand_migration", - error = %err, - "Failed to parse bucket metadata config" - ); - hook(bucket, None); - } + if let Some(hook) = BUCKET_CONFIG_PUBLISH_HOOK.get() { + hook( + bucket, + super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, + bm.on_demand_migration_config(), + ); } } -/// Withdraw a bucket's on-demand migration config when its metadata leaves -/// the cache. fn clear_on_demand_migration(bucket: &str) { - if let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() { - hook(bucket, None); + if let Some(hook) = BUCKET_CONFIG_PUBLISH_HOOK.get() { + hook(bucket, super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, None); } } @@ -1049,15 +1034,24 @@ pub async fn get_durability_config( } /// The bucket's on-demand migration config with its update time, or -/// `Ok(None)` when the bucket has none. A stored payload that does not parse -/// is a typed error (`OnDemandMigrationConfigError` inside `Error::Io`). -pub async fn get_on_demand_migration_config(bucket: &str) -> Result> { +/// `Ok(None)` when the bucket has none. Bytes are opaque to the metadata owner. +pub async fn get_on_demand_migration_config(bucket: &str) -> Result, OffsetDateTime)>> { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let bucket_meta_sys = bucket_meta_sys_lock.read().await; bucket_meta_sys.get_on_demand_migration_config(bucket).await } +/// Resolve opaque configuration from the store's own metadata system. +pub async fn get_on_demand_migration_config_in( + ctx: &crate::runtime::instance::InstanceContext, + bucket: &str, +) -> Result, OffsetDateTime)>> { + let sys = bucket_metadata_sys_of(ctx)?; + let lock = sys.read().await; + lock.get_on_demand_migration_config(bucket).await +} + pub async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let bucket_meta_sys = bucket_meta_sys_lock.read().await; @@ -2579,29 +2573,27 @@ impl BucketMetadataSys { } /// See [`get_on_demand_migration_config`]. - pub async fn get_on_demand_migration_config( - &self, - bucket: &str, - ) -> Result> { + pub async fn get_on_demand_migration_config(&self, bucket: &str) -> Result, OffsetDateTime)>> { let (bm, _) = self.get_config(bucket).await?; - let config = bm.on_demand_migration_config().map_err(Error::other)?; - Ok(config.map(|config| (config, bm.on_demand_migration_config_updated_at))) + Ok(bm + .on_demand_migration_config() + .map(|(bytes, updated_at)| (bytes.to_vec(), updated_at))) } } /// Test-only fixture shared with sibling modules (e.g. the quota checker /// tests): a 4-disk `ECStore` on an isolated instance context, so tests /// exercising the metadata system never touch ambient process state. -#[cfg(test)] -pub(crate) mod test_support { +#[cfg(any(test, feature = "test-util"))] +pub mod test_support { use super::*; use crate::disk::endpoint::Endpoint; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::runtime::instance::InstanceContext; use crate::store::init_local_disks_with_instance_ctx; - pub(crate) async fn isolated_store_over_temp_disks() -> (Vec, Arc) { + pub async fn isolated_store_over_temp_disks() -> (Vec, Arc) { let mut dirs = Vec::with_capacity(4); let mut endpoints = Vec::with_capacity(4); for disk_idx in 0..4 { @@ -4387,17 +4379,21 @@ mod tests { /// Every `(bucket, config)` the recording hook has seen. Tests filter by /// their own bucket name; the hook is process-wide and set once. - static ODM_HOOK_CALLS: std::sync::Mutex)>> = std::sync::Mutex::new(Vec::new()); + static ODM_HOOK_CALLS: std::sync::Mutex, OffsetDateTime)>)>> = std::sync::Mutex::new(Vec::new()); fn install_recording_odm_hook() { - ON_DEMAND_MIGRATION_CONFIG_HOOK.get_or_init(|| { - Box::new(|bucket, config| { - ODM_HOOK_CALLS.lock().unwrap().push((bucket.to_string(), config.cloned())); + BUCKET_CONFIG_PUBLISH_HOOK.get_or_init(|| { + Box::new(|bucket, config_file, config| { + assert_eq!(config_file, super::super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG); + ODM_HOOK_CALLS + .lock() + .unwrap() + .push((bucket.to_string(), config.map(|(bytes, stamp)| (bytes.to_vec(), stamp)))); }) }); } - fn odm_hook_calls(bucket: &str) -> Vec> { + fn odm_hook_calls(bucket: &str) -> Vec, OffsetDateTime)>> { ODM_HOOK_CALLS .lock() .unwrap() @@ -4407,54 +4403,6 @@ mod tests { .collect() } - /// rustfs/backlog#2148: the accessor reports absence as `Ok(None)` and a - /// stored payload it cannot parse as a typed error, never as a default - /// and never as `ConfigNotFound`. - #[tokio::test] - async fn get_on_demand_migration_config_distinguishes_absent_from_corrupt() { - use crate::bucket::on_demand_migration::OnDemandMigrationConfigError; - - let (_dirs, ecstore) = isolated_store_over_temp_disks().await; - let sys = BucketMetadataSys::new(ecstore); - let bucket = "odm-accessor"; - - sys.set(bucket.to_string(), Arc::new(BucketMetadata::new(bucket))).await; - assert_eq!(sys.get_on_demand_migration_config(bucket).await.unwrap(), None); - - let mut corrupt = BucketMetadata::new(bucket); - corrupt.on_demand_migration_config_json = br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec(); - sys.set(bucket.to_string(), Arc::new(corrupt)).await; - let err = sys - .get_on_demand_migration_config(bucket) - .await - .expect_err("corrupt config must not read as a default"); - assert_ne!(err, Error::ConfigNotFound, "corruption must not be reported as absence"); - let typed = match &err { - Error::Io(io) => io - .get_ref() - .and_then(|source| source.downcast_ref::()), - _ => None, - }; - assert!( - matches!(typed, Some(OnDemandMigrationConfigError::Malformed(_))), - "typed parse error must survive the Result boundary, got: {err:?}" - ); - - let mut valid = BucketMetadata::new(bucket); - valid - .update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec()) - .unwrap(); - let stamped = valid.on_demand_migration_config_updated_at; - sys.set(bucket.to_string(), Arc::new(valid)).await; - let (config, updated_at) = sys - .get_on_demand_migration_config(bucket) - .await - .unwrap() - .expect("stored config is returned"); - assert_eq!(config, OnDemandMigrationConfig::from_json(ODM_JSON).unwrap()); - assert_eq!(updated_at, stamped); - } - /// rustfs/backlog#2148: the publish hook fires on every path that /// installs bucket metadata into the cache (set, initial load, peer /// reload, refresh loop, lazy load) and withdraws on removal, mirroring @@ -4468,11 +4416,15 @@ mod tests { for dir in &dirs { std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist"); } - let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap(); + let expect_publish = |before: usize, label: &str| { let calls = odm_hook_calls(bucket); assert_eq!(calls.len(), before + 1, "{label} must publish exactly once"); - assert_eq!(calls.last().unwrap().as_ref(), Some(&expected), "{label} must publish the stored config"); + assert_eq!( + calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()), + Some(ODM_JSON), + "{label} must publish the stored bytes" + ); }; // set (via persist_new_and_set, which installs through `set`). @@ -4518,14 +4470,18 @@ mod tests { assert_eq!(calls.len(), before + 1, "remove must withdraw exactly once"); assert_eq!(calls.last().unwrap(), &None); - // A corrupt payload is withdrawn, never published as a config. + // Opaque bytes reach the application even if they are not valid JSON. let mut corrupt = BucketMetadata::new(bucket); corrupt.on_demand_migration_config_json = b"not-json".to_vec(); let before = odm_hook_calls(bucket).len(); lazy.set(bucket.to_string(), Arc::new(corrupt)).await; let calls = odm_hook_calls(bucket); assert_eq!(calls.len(), before + 1); - assert_eq!(calls.last().unwrap(), &None, "unreadable config must publish absence"); + assert_eq!( + calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()), + Some(b"not-json".as_slice()), + "the application validates opaque config bytes" + ); } #[tokio::test] diff --git a/crates/ecstore/src/bucket/mod.rs b/crates/ecstore/src/bucket/mod.rs index e93419cf2..da42fc43e 100644 --- a/crates/ecstore/src/bucket/mod.rs +++ b/crates/ecstore/src/bucket/mod.rs @@ -26,7 +26,6 @@ mod metadata_test; pub mod migration; mod msgp_decode; pub mod object_lock; -pub mod on_demand_migration; pub mod policy_sys; pub mod quota; pub mod remote_s3_client; diff --git a/crates/ecstore/src/bucket/remote_s3_client.rs b/crates/ecstore/src/bucket/remote_s3_client.rs index 6389aa67e..1385cea80 100644 --- a/crates/ecstore/src/bucket/remote_s3_client.rs +++ b/crates/ecstore/src/bucket/remote_s3_client.rs @@ -283,9 +283,7 @@ impl Intercept for UserAgentSuffixInterceptor { /// Builds the SDK config for `spec` without finalizing it, so callers can add /// interceptors or (in tests) swap the HTTP client before `build()`. -pub(crate) async fn build_remote_s3_config( - spec: &RemoteS3EndpointSpec, -) -> Result { +pub async fn build_remote_s3_config(spec: &RemoteS3EndpointSpec) -> Result { let Some(credentials) = &spec.credentials else { return Err(RemoteS3ClientError::MissingCredentials); }; @@ -525,7 +523,7 @@ fn validate_ca_pem_bundle(ca_cert_pem: &[u8]) -> Result<(), String> { Ok(()) } -pub(crate) fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), RemoteS3ClientError> { +pub fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), RemoteS3ClientError> { validate_ca_pem_bundle(ca_cert_pem.as_bytes()).map_err(RemoteS3ClientError::InvalidCaPem) } diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index aea11381d..009cc94f2 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -38,3 +38,4 @@ pub(crate) use storage_api::metrics::{ obs_on_demand_migration_snapshot, obs_replication_site_stats_snapshot, obs_resolve_object_store_handle, obs_transition_state_handle, }; +pub use storage_api::register_on_demand_migration_metrics_source; diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index e69cc64d8..44b1f0834 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -17,13 +17,6 @@ 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::backfill::{ - BackfillCheckpoint as SourceBackfillCheckpoint, global_backfill_runner as source_global_backfill_runner, -}; -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, @@ -44,9 +37,7 @@ 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::{ - OdmBackfillBucketStats, OdmBackfillRuntimeStats, OnDemandMigrationBreakerState, OnDemandMigrationBucketStats, -}; +use crate::metrics::collectors::{OdmBackfillBucketStats, OdmBackfillRuntimeStats, OnDemandMigrationBucketStats}; #[derive(Debug, Clone, PartialEq)] pub(crate) struct ObsBucketReplicationTargetStatsSnapshot { @@ -465,70 +456,37 @@ 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, - }, - } +struct OnDemandMigrationMetricsSource { + snapshot: fn() -> Vec, + backfill_snapshot: fn() -> Vec, } -/// 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() -} +static ON_DEMAND_MIGRATION_METRICS_SOURCE: std::sync::OnceLock = std::sync::OnceLock::new(); -fn on_demand_migration_backfill_stats_from_checkpoint( - bucket: String, - checkpoint: SourceBackfillCheckpoint, -) -> OdmBackfillBucketStats { - OdmBackfillBucketStats { - bucket, - state: checkpoint.state.as_str().to_string(), - listed: checkpoint.listed, - enqueued: checkpoint.enqueued, - pulled: checkpoint.pulled, - skipped_existing: checkpoint.skipped_existing, - failed: checkpoint.failed, - bytes: checkpoint.bytes, - } -} - -/// Backfill jobs running on this node, sorted by bucket. Empty until the -/// runner is installed, and empty again once a job finishes: the series are -/// per-node job progress, not a cluster-wide history. -pub(crate) fn obs_on_demand_migration_backfill_snapshot(server: String) -> OdmBackfillRuntimeStats { - let buckets = source_global_backfill_runner() - .map(|runner| { - runner - .local_job_snapshots() - .into_iter() - .map(|(bucket, checkpoint)| on_demand_migration_backfill_stats_from_checkpoint(bucket, checkpoint)) - .collect() +/// Register the application-owned ODM snapshots before starting the collector. +pub fn register_on_demand_migration_metrics_source( + snapshot: fn() -> Vec, + backfill_snapshot: fn() -> Vec, +) -> bool { + ON_DEMAND_MIGRATION_METRICS_SOURCE + .set(OnDemandMigrationMetricsSource { + snapshot, + backfill_snapshot, }) + .is_ok() +} + +pub(crate) fn obs_on_demand_migration_snapshot() -> Vec { + ON_DEMAND_MIGRATION_METRICS_SOURCE + .get() + .map(|source| (source.snapshot)()) + .unwrap_or_default() +} + +pub(crate) fn obs_on_demand_migration_backfill_snapshot(server: String) -> OdmBackfillRuntimeStats { + let buckets = ON_DEMAND_MIGRATION_METRICS_SOURCE + .get() + .map(|source| (source.backfill_snapshot)()) .unwrap_or_default(); OdmBackfillRuntimeStats { server, buckets } } @@ -580,6 +538,31 @@ pub(crate) async fn obs_replication_site_stats_snapshot(current_data_transfer_ra mod tests { use super::*; + #[test] + fn on_demand_migration_callbacks_supply_runtime_snapshots() { + assert!(register_on_demand_migration_metrics_source( + || vec![OnDemandMigrationBucketStats { + bucket: "configured".into(), + pulled_bytes_total: 4096, + ..Default::default() + }], + || vec![OdmBackfillBucketStats { + bucket: "backfill".into(), + pulled: 3, + ..Default::default() + }], + )); + let snapshot = obs_on_demand_migration_snapshot(); + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].bucket, "configured"); + assert_eq!(snapshot[0].pulled_bytes_total, 4096); + let backfill = obs_on_demand_migration_backfill_snapshot("node-a".into()); + assert_eq!(backfill.server, "node-a"); + assert_eq!(backfill.buckets.len(), 1); + assert_eq!(backfill.buckets[0].bucket, "backfill"); + assert_eq!(backfill.buckets[0].pulled, 3); + } + #[test] fn obs_replication_numeric_conversions_floor_negative_values() { assert_eq!(i64_to_u64_floor_zero(-1), 0); @@ -772,51 +755,6 @@ 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( diff --git a/docs/architecture/background-services-inventory.md b/docs/architecture/background-services-inventory.md index 772a761c0..01b612a3a 100644 --- a/docs/architecture/background-services-inventory.md +++ b/docs/architecture/background-services-inventory.md @@ -13,8 +13,8 @@ Operator-facing behaviour, configuration, and troubleshooting for these services | Service | Desired source | Current-status inputs | Status surface | Side effects | |---|---|---|---|---| -| Write-back pull pipeline (`crates/ecstore/src/bucket/on_demand_migration/pull.rs`; the local write is delegated to the app layer in `rustfs/src/app/object/on_demand_migration_put.rs`) | The bucket's `on-demand-migration.json` (`enabled`, `policy.max_concurrent_pulls`, `pull_queue_capacity`, `multipart_part_size_bytes`, `bandwidth_limit_bytes_per_sec`) together with the process switch `RUSTFS_ON_DEMAND_MIGRATION_ENABLED` | Per-bucket runtime state in `crates/ecstore/src/bucket/on_demand_migration/sys.rs`: whether a state is installed, whether its source client built, its cancellation token, queue depth, in-flight pull permits | `GET /rustfs/admin/v3/on-demand-migration/{bucket}/status` (`inflight_pulls`, `queue_depth`, `counters.pulled_*`, `counters.pull_failures_total`) and the `rustfs_on_demand_migration_*` series | Source GET/HEAD/GetObjectTagging traffic; local object writes through the internal put path, hence quota consumption, bucket default SSE, versioning, Object Lock defaults, `ObjectCreated` notifications, and outbound replication scheduling | -| Backfill job (module under `crates/ecstore/src/bucket/on_demand_migration/`, rustfs/backlog#2159 — not yet in the tree) | An admin `start` request plus the bucket config; invalidated when the config's `updated_at` changes or the config is deleted | The persisted checkpoint under the bucket's metadata prefix, its `state` field, and the owner lease | The backfill section of the bucket status endpoint and the `rustfs_on_demand_migration_backfill_*` series | Source `ListObjectsV2` paging; queue admission into the write-back pipeline (and therefore all of its side effects); checkpoint writes | -| Backfill recovery loop (registered from `rustfs/src/startup_background.rs`, rustfs/backlog#2159 — not yet in the tree) | The set of persisted checkpoints in `state = running`; runs on every node | Checkpoint owner lease expiry | Takeover is reported through the same backfill status; a takeover emits a warn-level lease event | Claims the lease and resumes the backfill job, inheriting its side effects. Scanning checkpoints is read-only | +| Write-back pull pipeline (`rustfs/src/on_demand_migration/pull.rs`; the local write is delegated to the app layer in `rustfs/src/app/object/on_demand_migration_put.rs`) | The bucket's `on-demand-migration.json` (`enabled`, `policy.max_concurrent_pulls`, `pull_queue_capacity`, `multipart_part_size_bytes`, `bandwidth_limit_bytes_per_sec`) together with the process switch `RUSTFS_ON_DEMAND_MIGRATION_ENABLED` | Per-bucket runtime state in `rustfs/src/on_demand_migration/sys.rs`: whether a state is installed, whether its source client built, its cancellation token, queue depth, in-flight pull permits | `GET /rustfs/admin/v3/on-demand-migration/{bucket}/status` (`inflight_pulls`, `queue_depth`, `counters.pulled_*`, `counters.pull_failures_total`) and the `rustfs_on_demand_migration_*` series | Source GET/HEAD/GetObjectTagging traffic; local object writes through the internal put path, hence quota consumption, bucket default SSE, versioning, Object Lock defaults, `ObjectCreated` notifications, and outbound replication scheduling | +| Backfill job (module under `rustfs/src/on_demand_migration/`, rustfs/backlog#2159) | An admin `start` request plus the bucket config; invalidated when the config's `updated_at` changes or the config is deleted | The persisted checkpoint under the bucket's metadata prefix, its `state` field, and the owner lease | The backfill section of the bucket status endpoint and the `rustfs_on_demand_migration_backfill_*` series | Source `ListObjectsV2` paging; queue admission into the write-back pipeline (and therefore all of its side effects); checkpoint writes | +| Backfill recovery loop (registered from `rustfs/src/startup_background.rs`, rustfs/backlog#2159) | The set of persisted checkpoints in `state = running`; runs on every node | Checkpoint owner lease expiry | Takeover is reported through the same backfill status; a takeover emits a warn-level lease event | Claims the lease and resumes the backfill job, inheriting its side effects. Scanning checkpoints is read-only | The pull pipeline has no separate loop of its own: a bucket's queue dispatcher starts lazily on the first background pull and is cancelled when the bucket's state is rebuilt or removed, and each inline pull commits in a task that outlives its request so a client disconnect cannot truncate the stored object. Neither the switch nor the config is re-read by the workers: the bucket-metadata publish hook rebuilds the state, which is the only desired-state path. diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index fb94ed0aa..6fdca19ab 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -84,3 +84,28 @@ The server-config model (`Config`, `KV`, `KVS`) and the global server-config sna ## Required Architecture Documents The guard requires the documents and section headings listed in its `require_source_contains` entries (`scripts/check_architecture_migration_rules.sh`); the directory index is [README.md](README.md). + +## On-Demand Migration Service + +`rustfs/src/on_demand_migration/` owns source clients, pull scheduling, list +merging, runtime state and backfill orchestration. Its `storage_api.rs` is the +only ECStore facade boundary. Object write-back still enters the application's +internal PUT and multipart use cases, including the atomic create-only commit, +delete-marker protection, encryption, quota and notification rules. + +ECStore stores the existing ODM bytes and update timestamp without interpreting +the JSON. Every metadata cache install or removal publishes those bytes through +`BUCKET_CONFIG_PUBLISH_HOOK`; the application decodes them and synchronously +withdraws corrupt configurations. Configuration writes validate structure and +deployment constraints in the admin use case before the incarnation-fenced +metadata update. Backfill reads metadata from its store's instance context and +preserves the checkpoint ETag compare-and-set, lease and tail-drained writes. + +Observability owns its metric DTOs and accepts application snapshot callbacks; +it does not depend on the ODM runtime. The application registers both bucket +and backfill snapshots during startup, before metadata and metric collection. + +This boundary does not change `.metadata.bin`, the ODM wire format or the +backfill checkpoint format. An older binary may still discard unknown metadata +fields when it rewrites a bucket; service relocation does not make mixed-version +configuration writes or rollback preserve ODM configuration. diff --git a/docs/architecture/ecstore-api-facade-inventory.md b/docs/architecture/ecstore-api-facade-inventory.md index 7fbd79c52..71b9dcbb8 100644 --- a/docs/architecture/ecstore-api-facade-inventory.md +++ b/docs/architecture/ecstore-api-facade-inventory.md @@ -63,3 +63,10 @@ Lifecycle, replication, and `SetDisks` split blockers, extracted contracts, and 4. Do not replace `SetDisks` with multiple runtime structs in one change; move one operation family only after contracts and focused tests exist. 5. Remove or narrow one facade group per change so rollback preserves object IO, quorum, lifecycle/replication queues, scanner repair, notification/audit events, and metadata compatibility. 6. Keep `api::bucket`, `api::config`, `api::disk`, and `api::tier` on explicit submodules and symbol lists; do not restore `pub use crate::::{...}` whole-module passthroughs for those groups. + +### On-Demand Migration + +`rustfs/src/on_demand_migration/storage_api.rs` owns the service's storage facade +imports: opaque bucket configuration, shared remote S3 client construction, +namespace locking, object options and metadata-object persistence. ODM types +are owned by the application and are no longer exported through ECStore. diff --git a/docs/architecture/remote-credential-sealing-adr.md b/docs/architecture/remote-credential-sealing-adr.md index 618011fec..4a21029b3 100644 --- a/docs/architecture/remote-credential-sealing-adr.md +++ b/docs/architecture/remote-credential-sealing-adr.md @@ -1,7 +1,7 @@ # Remote Credential Sealing ADR **Use this when:** you add, read, or persist a stored remote credential — a replication target, a remote tier, or an on-demand migration source — or you need the sealed-envelope format, the mixed-version rules, or the reason this is worth doing in one deployment and not in another. -**Source of truth:** the three stores that hold remote credentials — `BUCKET_TARGETS_FILE` and `BUCKET_ON_DEMAND_MIGRATION_CONFIG` in `crates/ecstore/src/bucket/metadata.rs`, `TIER_CONFIG_FILE` in `crates/ecstore/src/services/tier/tier.rs` — the shared envelope in `crates/ecstore/src/bucket/sealed_credentials.rs`, the consumers `crates/ecstore/src/bucket/bucket_target_sys.rs`, `crates/ecstore/src/services/tier/tier_config.rs` and `crates/ecstore/src/bucket/on_demand_migration/config.rs`, and the backend properties in [../operations/kms-backend-security.md](../operations/kms-backend-security.md). +**Source of truth:** the three stores that hold remote credentials — `BUCKET_TARGETS_FILE` and `BUCKET_ON_DEMAND_MIGRATION_CONFIG` in `crates/ecstore/src/bucket/metadata.rs`, `TIER_CONFIG_FILE` in `crates/ecstore/src/services/tier/tier.rs` — the shared envelope in `crates/ecstore/src/bucket/sealed_credentials.rs`, the consumers `crates/ecstore/src/bucket/bucket_target_sys.rs`, `crates/ecstore/src/services/tier/tier_config.rs` and `rustfs/src/on_demand_migration/config.rs`, and the backend properties in [../operations/kms-backend-security.md](../operations/kms-backend-security.md). ## Recommendation @@ -37,7 +37,7 @@ Two of the three are not files at all. `bucket-targets.json` and `on-demand-migr | Store | Reached as | Actually persisted at | Written by | Container | |---|---|---|---|---| | Replication and ILM targets | `BUCKET_TARGETS_FILE` | `BucketMetadata::bucket_targets_config_json`, msgpack field `BucketTargetsConfigJSON` | `BucketMetadata::update_config`, then `BucketMetadata::save_with_store`; `crates/ecstore/src/bucket/metadata_sys.rs` serializes the update under a transaction lock | `{BUCKET_META_PREFIX}/{bucket}/{BUCKET_METADATA_FILE}` in `RUSTFS_META_BUCKET` (`crates/ecstore/src/disk/mod.rs`) | -| On-demand migration source | `BUCKET_ON_DEMAND_MIGRATION_CONFIG` | `BucketMetadata::on_demand_migration_config_json`, msgpack field `OnDemandMigrationConfigJSON` | same path; `update_config` additionally refuses a blob this build cannot parse | same blob as above | +| On-demand migration source | `BUCKET_ON_DEMAND_MIGRATION_CONFIG` | `BucketMetadata::on_demand_migration_config_json`, msgpack field `OnDemandMigrationConfigJSON` | same path; the application validates structure and deployment constraints before persistence | same blob as above | | Remote tiers | `TIER_CONFIG_FILE` | its own object, a four-byte `TIER_CONFIG_FORMAT` / `TIER_CONFIG_VERSION` header followed by an `rmp_serde` payload of `ExternalTierConfigMgr` | `TierConfigMgr` through `encode_external_tiering_config_blob`, under `tier_config_lock_path` | `tier_config_path` under `CONFIG_PREFIX` in `RUSTFS_META_BUCKET` | The consequence of the first two sharing a blob is that any change to how that blob parses has a blast radius covering policy, lifecycle, versioning, object lock and everything else in `BucketMetadata` — not just credentials. @@ -48,7 +48,7 @@ Three things hold the line today, and all three keep working whether or not seal - **The reserved bucket.** `RUSTFS_META_BUCKET` is `.rustfs.sys`; `is_reserved_or_invalid_bucket` keeps it off the S3 surface, and the admin inspect archive in `rustfs/src/admin/handlers/inspect_archive.rs` runs its request through a strict bucket-name check that a dot-prefixed reserved name does not pass. - **Admin authorization** on every route that can read or write one of the three configurations. -- **Redaction on every read path.** `BucketTarget::redacted_credentials` and the `Debug` for `Credentials` in `crates/ecstore/src/bucket/target/bucket_target.rs`, used by the remote-target listing in `rustfs/src/admin/handlers/replication.rs` and by the bucket-metadata export in `rustfs/src/admin/handlers/bucket_meta.rs`; `TierConfig::redacted` in `crates/ecstore/src/services/tier/tier_config.rs`, which is also what that type's `Clone` and `Debug` do; and `SourceCredentials::redacted` in `crates/ecstore/src/bucket/on_demand_migration/config.rs`, used by `rustfs/src/admin/handlers/on_demand_migration.rs`. +- **Redaction on every read path.** `BucketTarget::redacted_credentials` and the `Debug` for `Credentials` in `crates/ecstore/src/bucket/target/bucket_target.rs`, used by the remote-target listing in `rustfs/src/admin/handlers/replication.rs` and by the bucket-metadata export in `rustfs/src/admin/handlers/bucket_meta.rs`; `TierConfig::redacted` in `crates/ecstore/src/services/tier/tier_config.rs`, which is also what that type's `Clone` and `Debug` do; and `SourceCredentials::redacted` in `rustfs/src/on_demand_migration/config.rs`, used by `rustfs/src/admin/handlers/on_demand_migration.rs`. So no API returns a stored secret. The bytes are reachable by reading the drives, and that is the boundary sealing is proposed to move. @@ -74,7 +74,7 @@ The envelope deliberately does **not** carry its own scope. A scope read out of ## Why a hook instead of a dependency -`crates/ecstore/Cargo.toml` has no `rustfs-kms` dependency, and adding one would invert the crate layering described in [crate-boundaries.md](crate-boundaries.md). The established shape is an `OnceLock` hook that ECStore defines and the binary installs at startup, as `EVENT_DISPATCH_HOOK` in `crates/ecstore/src/services/event_notification.rs` and `ON_DEMAND_MIGRATION_CONFIG_HOOK` in `crates/ecstore/src/bucket/on_demand_migration/config.rs` already do. `install_credential_sealer` follows it, and the binary supplies an implementation backed by `crates/kms/src/service_manager.rs`. +`crates/ecstore/Cargo.toml` has no `rustfs-kms` dependency, and adding one would invert the crate layering described in [crate-boundaries.md](crate-boundaries.md). The established shape is an `OnceLock` hook that ECStore defines and the binary installs at startup, as `EVENT_DISPATCH_HOOK` in `crates/ecstore/src/services/event_notification.rs` and `BUCKET_CONFIG_PUBLISH_HOOK` in `crates/ecstore/src/bucket/metadata_sys.rs` already do. `install_credential_sealer` follows it, and the binary supplies an implementation backed by `crates/kms/src/service_manager.rs`. ## Compatibility, per store, because the three differ diff --git a/docs/operations/on-demand-migration.md b/docs/operations/on-demand-migration.md index e9beb27d7..2932ecb3f 100644 --- a/docs/operations/on-demand-migration.md +++ b/docs/operations/on-demand-migration.md @@ -1,7 +1,7 @@ # On-Demand Migration **Use this when:** you are moving an existing S3-compatible bucket into RustFS without a stop-the-world copy, or you are debugging a bucket that serves reads from an external source (424 `SourceUnavailable`, an open circuit breaker, missing pulled objects, a source 403). -**Source of truth:** `crates/ecstore/src/bucket/on_demand_migration/` (`config.rs` for the wire model and its bounds, `sys.rs` for the per-node runtime, `source_client.rs` for the outbound client, `pull.rs` for the write-back pipeline, `breaker.rs` and `negative_cache.rs` for the protections), `rustfs/src/app/object/get.rs` and `head.rs` for the read paths, `rustfs/src/app/object/on_demand_migration_put.rs` for the local write, `rustfs/src/admin/handlers/on_demand_migration.rs` for the admin API, and `crates/obs/src/metrics/schema/on_demand_migration.rs` for the metric contract. +**Source of truth:** `rustfs/src/on_demand_migration/` (`config.rs` for the wire model and its bounds, `sys.rs` for the per-node runtime, `source_client.rs` for the outbound client, `pull.rs` for the write-back pipeline, `breaker.rs` and `negative_cache.rs` for the protections), `rustfs/src/app/object/get.rs` and `head.rs` for the read paths, `rustfs/src/app/object/on_demand_migration_put.rs` for the local write, `rustfs/src/admin/handlers/on_demand_migration.rs` for the admin API, and `crates/obs/src/metrics/schema/on_demand_migration.rs` for the metric contract. On-Demand Migration (ODM) attaches an external S3-compatible **source bucket** to a local RustFS bucket. When a client GETs a key that does not exist locally, RustFS fetches it from the source, streams it to the client, and stores it locally in the same pass; every later read is served locally. It is a pull-style, lazy migration path — the RustFS equivalent of Cloudflare R2 Sippy, Tigris shadow buckets, and Alibaba Cloud OSS / Tencent COS mirror-back-to-origin. @@ -111,7 +111,7 @@ Read-through only migrates what clients touch. The background backfill job walks ## Configuration reference -The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unknown fields are rejected rather than dropped, so a config written by a newer build fails loudly on an older one. Every default and bound below comes from `crates/ecstore/src/bucket/on_demand_migration/config.rs`. +The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unknown fields are rejected rather than dropped, so a config written by a newer build fails loudly on an older one. Every default and bound below comes from `rustfs/src/on_demand_migration/config.rs`. | Field | Type | Default | Bounds / rules | |---|---|---|---| @@ -168,7 +168,7 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket | `azure` | Optional; derived as `https://.blob.core.windows.net` | Native Blob REST, not S3 | Unused; write `auto` | Needs `source.azure`; the container is `source.bucket`. Reads need `Read` on the blob and `List` on the container, plus `Tags` when `policy.copy_tags` is on | None yet: no interop job covers Azure | | `gcs_native` | Optional; derived as `https://storage.googleapis.com` | Native GCS API, not S3 | Unused; write `auto` | Needs `source.gcs`. Reads use the XML API for objects and `objects.list` for listings, both with an OAuth token minted from the service-account key; the key needs `storage.objects.get` and `storage.objects.list` | None yet: no interop job covers native GCS | -Every backend answers the same trait contract, pinned by `backend_contract.rs` in `crates/ecstore/src/bucket/on_demand_migration/`, and the three differences that contract allows are the ones documented here. +Every backend answers the same trait contract, pinned by `backend_contract.rs` in `rustfs/src/on_demand_migration/`, and the three differences that contract allows are the ones documented here. `azure` differs in two of them. Its ETag is a concurrency token rather than a digest of the bytes, so it is stored as `odm-source-etag` provenance and never used as the expected MD5 of a pulled object — the write-back integrity check falls back to the local digest. And its listing paginates only with an opaque marker: there is no "start after this key" form, so a caller that asks for one gets `Unsupported` instead of a listing that silently starts over. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 063ce400e..03b6b378e 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -58,7 +58,7 @@ required-features = ["swift"] [features] default = ["ftps", "webdav", "gcs"] -gcs = ["rustfs-ecstore/gcs"] +gcs = ["rustfs-ecstore/gcs", "dep:google-cloud-auth"] metrics-gpu = ["rustfs-obs/gpu"] ftps = ["rustfs-protocols/ftps"] swift = ["rustfs-protocols/swift"] @@ -288,6 +288,9 @@ reqwest = { workspace = true, features = ["json", "stream"] } socket2 = { workspace = true, features = ["all"] } tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "signal", "process", "io-util", "fs"] } tokio-rustls = { workspace = true, default-features = false, features = ["logging", "tls12", "aws-lc-rs"] } +aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] } +aws-smithy-types = { workspace = true } +google-cloud-auth = { workspace = true, optional = true } aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } tokio-stream.workspace = true tokio-util = { workspace = true, features = ["io", "compat", "time"] } diff --git a/rustfs/src/admin/handlers/on_demand_migration.rs b/rustfs/src/admin/handlers/on_demand_migration.rs index d8bc9d713..3e331b26f 100644 --- a/rustfs/src/admin/handlers/on_demand_migration.rs +++ b/rustfs/src/admin/handlers/on_demand_migration.rs @@ -38,16 +38,6 @@ use crate::admin::runtime_sources::{ }; use crate::admin::storage_api::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG; use crate::admin::storage_api::bucket::metadata_sys; -use crate::admin::storage_api::bucket::on_demand_migration::backfill::{ - BackfillCheckpoint, BackfillError, BackfillRequest, BackfillState, SkipExisting, global_backfill_runner, -}; -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::{ - OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext, - source_backend_spec, -}; use crate::admin::storage_api::bucket::remote_s3_client::{ PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy, }; @@ -58,6 +48,16 @@ use crate::admin::utils::{extract_query_params, read_compatible_admin_body}; use crate::error::ApiError; use crate::license::license_check; use crate::module_switches::{ENV_ON_DEMAND_MIGRATION_ENABLED, on_demand_migration_enabled_from_env}; +use crate::on_demand_migration::backfill::{ + BackfillCheckpoint, BackfillError, BackfillRequest, BackfillState, SkipExisting, global_backfill_runner, +}; +use crate::on_demand_migration::source_client::{ + SourceClient, SourceClientSpec, SourceError, SourceProbe, SourceProvider, SourceTimeouts, +}; +use crate::on_demand_migration::{ + OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext, + source_backend_spec, +}; use crate::server::ADMIN_PREFIX; use hyper::{Method, StatusCode}; use matchit::Params; @@ -580,7 +580,7 @@ async fn validate_config(bucket: &str, config: &OnDemandMigrationConfig) -> S3Re } fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider { - use crate::admin::storage_api::bucket::on_demand_migration::Provider; + use crate::on_demand_migration::Provider; match config.source.provider { Provider::S3 => SourceProvider::S3, Provider::Aws => SourceProvider::Aws, @@ -787,7 +787,7 @@ impl Operation for GetBucketOnDemandMigrationHandler { let bucket = bucket_from_params(¶ms)?; let cred = authorize_for_bucket(&req, AdminAction::GetBucketOnDemandMigrationAction, &bucket).await?; - let Some((config, updated_at)) = metadata_sys::get_on_demand_migration_config(&bucket).await.map_err(|err| { + let Some((config, updated_at)) = crate::on_demand_migration::config::get_config(&bucket).await.map_err(|err| { admin_s3_error(S3ErrorCode::InternalError, format!("failed to read on-demand migration config: {err}")) })? else { @@ -846,7 +846,7 @@ impl Operation for GetBucketOnDemandMigrationStatusHandler { let bucket = bucket_from_params(¶ms)?; let cred = authorize_for_bucket(&req, AdminAction::GetBucketOnDemandMigrationAction, &bucket).await?; - let config = metadata_sys::get_on_demand_migration_config(&bucket).await.map_err(|err| { + let config = crate::on_demand_migration::config::get_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); @@ -1763,6 +1763,19 @@ mod store_tests { assert_eq!(body["inflight_pulls"], Value::from(0)); assert_eq!(body["queue_depth"], Value::from(0)); + // A malformed replacement cannot overwrite the saved source. + let before = metadata_sys::get(BUCKET).await.expect("saved metadata"); + for invalid in [b"not-json".to_vec(), br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec()] { + let err = SetBucketOnDemandMigrationHandler {} + .call(root_request(Method::PUT, config_uri(""), invalid), bucket_params(&router)) + .await + .expect_err("malformed replacement must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + let after = metadata_sys::get(BUCKET).await.expect("saved metadata remains readable"); + assert_eq!(after.on_demand_migration_config_json, before.on_demand_migration_config_json); + assert_eq!(after.on_demand_migration_config_updated_at, before.on_demand_migration_config_updated_at); + } + // The peer fan-out ran: the single unreachable peer is reported. let context = crate::admin::runtime_sources::current_app_context(); let err = reload_peers(context.as_deref(), BUCKET) diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index ea7ae56b5..01caaa95f 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -20,8 +20,8 @@ use time::OffsetDateTime; mod ecstore_bucket { pub(crate) use crate::storage::storage_api::ecstore_bucket::{ - bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, object_lock, on_demand_migration, quota, - remote_s3_client, replication, target, utils, versioning, versioning_sys, + bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, object_lock, quota, remote_s3_client, + replication, target, utils, versioning, versioning_sys, }; } @@ -284,35 +284,6 @@ pub(crate) mod durability { pub(crate) type BucketDurabilityConfig = super::ecstore_bucket::durability::BucketDurabilityConfig; } -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>; - pub(crate) use super::ecstore_bucket::on_demand_migration::source_backend_spec; - - pub(crate) mod backfill { - pub(crate) type BackfillCheckpoint = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillCheckpoint; - pub(crate) type BackfillError = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillError; - pub(crate) type BackfillRequest = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillRequest; - pub(crate) type BackfillState = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillState; - pub(crate) type SkipExisting = super::super::ecstore_bucket::on_demand_migration::backfill::SkipExisting; - pub(crate) use super::super::ecstore_bucket::on_demand_migration::backfill::global_backfill_runner; - } - - pub(crate) mod source_client { - pub(crate) type SourceClient = super::super::ecstore_bucket::on_demand_migration::source_client::SourceClient; - pub(crate) type SourceClientSpec = super::super::ecstore_bucket::on_demand_migration::source_client::SourceClientSpec; - pub(crate) type SourceError = super::super::ecstore_bucket::on_demand_migration::source_client::SourceError; - pub(crate) type SourceProbe = super::super::ecstore_bucket::on_demand_migration::source_client::SourceProbe; - pub(crate) type SourceProvider = super::super::ecstore_bucket::on_demand_migration::source_client::SourceProvider; - pub(crate) type SourceTimeouts = super::super::ecstore_bucket::on_demand_migration::source_client::SourceTimeouts; - } -} - pub(crate) mod remote_s3_client { pub(crate) type PathStyle = super::ecstore_bucket::remote_s3_client::PathStyle; pub(crate) type RemoteCredentials = super::ecstore_bucket::remote_s3_client::RemoteCredentials; @@ -455,12 +426,6 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::get_durability_config(bucket).await } - pub(crate) async fn get_on_demand_migration_config( - bucket: &str, - ) -> Result> { - super::ecstore_bucket::metadata_sys::get_on_demand_migration_config(bucket).await - } - pub(crate) async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> { super::ecstore_bucket::metadata_sys::get_quota_config(bucket).await } @@ -913,7 +878,6 @@ pub(crate) mod bucket { pub(crate) use super::lifecycle; pub(crate) use super::metadata; pub(crate) use super::metadata_sys; - pub(crate) use super::on_demand_migration; pub(crate) use super::quota; pub(crate) use super::remote_s3_client; pub(crate) use super::replication; diff --git a/rustfs/src/app/bucket_list_through.rs b/rustfs/src/app/bucket_list_through.rs index ce09fc990..c385fdb9c 100644 --- a/rustfs/src/app/bucket_list_through.rs +++ b/rustfs/src/app/bucket_list_through.rs @@ -25,11 +25,6 @@ use super::storage_api::bucket_usecase::ECStore; use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo; use super::storage_api::bucket_usecase::StorageObjectOptions; -use super::storage_api::bucket_usecase::bucket::on_demand_migration::{ - BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, - MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan, - SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan, -}; use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys; use super::storage_api::bucket_usecase::contract::list::{ListObjectsV2Info as StorageListObjectsV2Info, ListOperations as _}; use super::storage_api::bucket_usecase::contract::object::ObjectOperations as _; @@ -37,6 +32,11 @@ use super::storage_api::bucket_usecase::s3::{S3Error, S3ErrorCode, S3Result}; use super::storage_api::bucket_usecase::s3_api::bucket::ListObjectsV2Params; use crate::app::object::shared::{odm_source_unavailable_error, odm_state_error_class}; use crate::error::ApiError; +use crate::on_demand_migration::{ + BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, + MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan, + SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan, +}; use futures::StreamExt; use http::HeaderMap; use rustfs_utils::http::{SUFFIX_SOURCE_PROXY_REQUEST, get_header}; @@ -443,14 +443,14 @@ mod tests { use super::*; use crate::app::bucket_usecase::DefaultBucketUsecase; use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore}; - use crate::app::storage_api::bucket_usecase::bucket::on_demand_migration::{ - FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, - SourceCredentials, TlsConfig, - }; use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response}; use crate::app::storage_api::test::StoragePutObjReader; use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; use crate::app::storage_api::test::contract::object::ObjectIO as _; + use crate::on_demand_migration::{ + FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, + SourceCredentials, TlsConfig, + }; use s3s::dto::ListObjectsInput; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index 65039350c..591c4b4c6 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -15,11 +15,11 @@ //! GetObject / GetObjectAttributes read path: cold fill, resume, stream tuning. use super::*; -use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ +use crate::on_demand_migration::WriteBackBody; +use crate::on_demand_migration::{ BucketOdmState, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, PullError, PullLeader, PullOutcome, PullReason, PullSlot, RangeGetPolicy, SourceBody, SourceClient, SourceError, SourceGet, SourceHead, commit_inline, idle_guarded_body, }; -use crate::app::storage_api::object_usecase::on_demand_migration::WriteBackBody; use rustfs_rio::{TeeOptions, TeePrimary, tee_reader_with_options}; use tokio_stream::wrappers::ReceiverStream; @@ -4798,11 +4798,11 @@ pub(super) async fn odm_get_from_source( #[cfg(test)] mod on_demand_migration_tests { use super::*; - use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + use crate::on_demand_migration::{ BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OdmStateError, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, SourceErrorPolicy, TlsConfig, }; - use crate::app::storage_api::object_usecase::on_demand_migration::{ + use crate::on_demand_migration::{ LocalObject, OdmWriteBack, WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, }; use async_trait::async_trait; diff --git a/rustfs/src/app/object/head.rs b/rustfs/src/app/object/head.rs index 6d877af8a..7cb38bbbc 100644 --- a/rustfs/src/app/object/head.rs +++ b/rustfs/src/app/object/head.rs @@ -15,7 +15,7 @@ //! HeadObject path. use super::*; -use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ +use crate::on_demand_migration::{ BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, SourceClient, SourceError, SourceHead, }; @@ -638,7 +638,7 @@ impl DefaultObjectUsecase { #[cfg(test)] mod tests { use super::*; - use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + use crate::on_demand_migration::{ BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OdmStateError, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, SourceErrorPolicy, TlsConfig, }; @@ -739,11 +739,9 @@ mod tests { user_metadata: HashMap::from([("owner".to_string(), "alice".to_string())]), version_id: Some("v1".to_string()), storage_class: Some("STANDARD_IA".to_string()), - sse: Some( - crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::SourceSse::Kms { - key_id: Some("key-1".to_string()), - }, - ), + sse: Some(crate::on_demand_migration::source_client::SourceSse::Kms { + key_id: Some("key-1".to_string()), + }), is_multipart_etag: true, etag_is_opaque: false, } diff --git a/rustfs/src/app/object/on_demand_migration_put.rs b/rustfs/src/app/object/on_demand_migration_put.rs index 7edea71f8..50e87b5cf 100644 --- a/rustfs/src/app/object/on_demand_migration_put.rs +++ b/rustfs/src/app/object/on_demand_migration_put.rs @@ -13,7 +13,7 @@ // limitations under the License. //! On-demand migration write-back (rustfs/backlog#2153): the app-layer -//! [`OdmWriteBack`] the ecstore pull pipeline stores source objects with. +//! [`OdmWriteBack`] the migration service stores source objects with. //! //! Every write goes through the internal put entry points, so a pulled //! object is indistinguishable from a client PUT: bucket default SSE, quota, @@ -34,7 +34,7 @@ use super::*; use crate::app::storage_api::multipart_usecase::contract::multipart::CompletePart; -use crate::app::storage_api::object_usecase::on_demand_migration::{ +use crate::on_demand_migration::{ LocalObject, OdmWriteBack, SourceHead, WriteBackBody, WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, is_multipart_etag, }; @@ -297,7 +297,6 @@ impl OdmWriteBack for OnDemandMigrationWriteBack { mod tests { use super::*; use crate::app::storage_api::multipart_usecase::contract::multipart::MultipartOperations as _; - use crate::app::storage_api::object_usecase::on_demand_migration::{PullFailureReason, SourceSse}; use crate::app::storage_api::s3::{ BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault, @@ -306,6 +305,7 @@ mod tests { use crate::app::storage_api::test::bucket::utils::serialize; use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata}; + use crate::on_demand_migration::{PullFailureReason, SourceSse}; use http::Method; use rustfs_utils::http::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, contains_key_str, get_str}; use sha2::{Digest as Sha256Digest, Sha256}; diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index 8f0808956..8df039d14 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -15,9 +15,7 @@ //! Cross-cutting helpers shared by the object use-case modules. use super::*; -use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ - OdmStateError, PolicyConfig, SourceErrorPolicy, SourceHead, -}; +use crate::on_demand_migration::{OdmStateError, PolicyConfig, SourceErrorPolicy, SourceHead}; pub(super) const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id"; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 383df36a1..6c810e213 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -628,26 +628,6 @@ pub(crate) mod bucket { } } - pub(crate) mod on_demand_migration { - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::{ - SourceClient, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, - }; - #[cfg(test)] - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ - BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, - PathStyle, Provider, SourceConfig, SourceCredentials, TlsConfig, - }; - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ - BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig, - PullError, PullLeader, PullOutcome, PullReason, PullSlot, RangeGetPolicy, SourceBody, SourceErrorPolicy, - commit_inline, idle_guarded_body, - }; - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ - ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, - MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan, - }; - } - pub(crate) mod policy_sys { pub(crate) type PolicySys = crate::storage::storage_api::ecstore_bucket::policy_sys::PolicySys; } @@ -1181,19 +1161,6 @@ pub(crate) mod bucket_usecase { pub(crate) mod object_usecase { pub(crate) use super::storage_contracts::BUCKET_LIFECYCLE_LOCK_OBJECT; - pub(crate) mod on_demand_migration { - #[cfg(test)] - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::PullFailureReason; - #[cfg(test)] - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::SourceSse; - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::{ - SourceHead, is_multipart_etag, - }; - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ - LocalObject, OdmWriteBack, WriteBackBody, WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, - }; - } - pub(crate) mod object_cache { #[cfg(test)] pub(crate) use crate::storage::storage_api::ecstore_object::GetObjectBodySource; diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 95b0c0fba..77c387412 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -93,6 +93,7 @@ pub(crate) mod kms_rekey; pub mod license; pub mod memory_observability; pub mod module_switches; +pub mod on_demand_migration; pub mod profiling; #[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))] pub mod protocols; diff --git a/crates/ecstore/src/bucket/on_demand_migration/azure.rs b/rustfs/src/on_demand_migration/azure.rs similarity index 99% rename from crates/ecstore/src/bucket/on_demand_migration/azure.rs rename to rustfs/src/on_demand_migration/azure.rs index 08e687797..25d4cb0a1 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/azure.rs +++ b/rustfs/src/on_demand_migration/azure.rs @@ -40,8 +40,8 @@ use super::source_client::{ AzureAuth, AzureSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceTimeouts, range_header_value, }; -use crate::bucket::remote_s3_client::RemoteS3ClientError; -use crate::storage_api_contracts::range::HTTPRangeSpec; +use super::storage_api::HTTPRangeSpec; +use super::storage_api::remote_s3_client::RemoteS3ClientError; use hmac::{Hmac, Mac, digest::KeyInit}; use http::{HeaderMap, HeaderValue, Method}; use quick_xml::Reader; @@ -549,9 +549,9 @@ fn leaf_text(reader: &mut Reader<&[u8]>, end: quick_xml::name::QName<'_>) -> Res #[cfg(test)] mod tests { use super::*; - use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract}; - use crate::bucket::on_demand_migration::source_client::SourceError; - use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; + use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract}; + use crate::on_demand_migration::source_client::SourceError; + use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; const LIST_PAGE: &str = r#" diff --git a/crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs b/rustfs/src/on_demand_migration/backend_contract.rs similarity index 99% rename from crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs rename to rustfs/src/on_demand_migration/backend_contract.rs index a8e1b1337..3a99cd549 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs +++ b/rustfs/src/on_demand_migration/backend_contract.rs @@ -27,7 +27,7 @@ //! and whether the provider can resume a listing from a key. use super::source_client::{SourceBackend, SourceError, SourceListRequest}; -use crate::storage_api_contracts::range::HTTPRangeSpec; +use super::storage_api::HTTPRangeSpec; use std::collections::HashMap; /// The single object every fixture serves. diff --git a/crates/ecstore/src/bucket/on_demand_migration/backfill.rs b/rustfs/src/on_demand_migration/backfill.rs similarity index 98% rename from crates/ecstore/src/bucket/on_demand_migration/backfill.rs rename to rustfs/src/on_demand_migration/backfill.rs index 3a0ccfd69..8cc384ec8 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/backfill.rs +++ b/rustfs/src/on_demand_migration/backfill.rs @@ -45,16 +45,12 @@ use super::pull::{EnqueueOutcome, PullReason, QueuedPullOutcome}; use super::source_client::{SourceError, SourcePage}; +use super::storage_api::{ + BUCKET_META_PREFIX, ECStore, HTTPPreconditions, NamespaceLocking as _, ObjectOperations as _, ObjectOptions, + RUSTFS_META_BUCKET, StorageError, WriteCompletion, get_lock_acquire_timeout, get_on_demand_migration_config_in, + local_node_name, read_config_with_metadata, save_config_with_opts, +}; use super::sys::{BucketOdmState, OnDemandMigrationSys}; -use crate::bucket::metadata_sys::bucket_metadata_sys_of; -use crate::config::com::{read_config_with_metadata, save_config_with_opts}; -use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; -use crate::error::Error as StorageError; -use crate::object_api::ObjectOptions; -use crate::runtime::sources::local_node_name; -use crate::set_disk::get_lock_acquire_timeout; -use crate::storage_api_contracts::{namespace::NamespaceLocking as _, object::HTTPPreconditions, object::ObjectOperations as _}; -use crate::store::ECStore; use async_trait::async_trait; use futures::StreamExt; use futures::stream::FuturesUnordered; @@ -474,12 +470,10 @@ impl BackfillContext for BucketBackfillContext { } async fn config_updated_at(&self) -> Result, StorageError> { - let sys = bucket_metadata_sys_of(&self.api.ctx)?; - let guard = sys.read().await; - Ok(guard - .get_on_demand_migration_config(self.state.bucket()) - .await? - .map(|(_, updated_at)| updated_at)) + Ok( + super::config::decode_stored_config(get_on_demand_migration_config_in(&self.api.ctx, self.state.bucket()).await?)? + .map(|(_, updated_at)| updated_at), + ) } } @@ -683,7 +677,7 @@ async fn write_checkpoint( }; let opts = ObjectOptions { max_parity: true, - write_completion: crate::object_api::WriteCompletion::TailDrained, + write_completion: WriteCompletion::TailDrained, http_preconditions: Some(preconditions), ..Default::default() }; @@ -1506,10 +1500,10 @@ pub async fn run_backfill_recovery_loop(runner: Arc, cancel: Can #[cfg(test)] mod tests { + use super::super::storage_api::test_support::isolated_store_over_temp_disks; use super::*; - use crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks; - use crate::bucket::on_demand_migration::source_client::SourceObject; - use crate::bucket::on_demand_migration::sys::PullError; + use crate::on_demand_migration::source_client::SourceObject; + use crate::on_demand_migration::sys::PullError; use std::collections::{BTreeSet, HashSet}; use std::sync::atomic::AtomicBool; diff --git a/crates/ecstore/src/bucket/on_demand_migration/breaker.rs b/rustfs/src/on_demand_migration/breaker.rs similarity index 100% rename from crates/ecstore/src/bucket/on_demand_migration/breaker.rs rename to rustfs/src/on_demand_migration/breaker.rs diff --git a/crates/ecstore/src/bucket/on_demand_migration/config.rs b/rustfs/src/on_demand_migration/config.rs similarity index 94% rename from crates/ecstore/src/bucket/on_demand_migration/config.rs rename to rustfs/src/on_demand_migration/config.rs index d00f08c2c..28b67819c 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/config.rs +++ b/rustfs/src/on_demand_migration/config.rs @@ -14,16 +14,34 @@ //! Bucket-level On-Demand Migration configuration: wire model (JSON stored //! under `on-demand-migration.json`), pure validation, credential redaction, -//! and the publish hook the runtime registers into (rustfs/backlog#2148). +//! and persisted-config decoding (rustfs/backlog#2148). //! //! The persisted blob is not encrypted; it shares the trust boundary of //! `bucket-targets.json` and `tier-config.bin`. use serde::{Deserialize, Serialize}; use std::fmt; -use std::sync::OnceLock; use url::Url; +/// Decode bytes only at the service boundary, preserving typed corruption errors. +pub(super) fn decode_stored_config( + stored: Option<(Vec, time::OffsetDateTime)>, +) -> Result, super::storage_api::StorageError> { + stored + .map(|(bytes, updated_at)| { + OnDemandMigrationConfig::from_json(&bytes) + .map(|config| (config, updated_at)) + .map_err(super::storage_api::StorageError::other) + }) + .transpose() +} + +pub(crate) async fn get_config( + bucket: &str, +) -> Result, super::storage_api::StorageError> { + decode_stored_config(super::storage_api::get_on_demand_migration_config(bucket).await?) +} + /// The only wire version this build reads and writes. pub const ON_DEMAND_MIGRATION_CONFIG_VERSION: u32 = 1; @@ -786,16 +804,6 @@ impl EndpointKey { } } -/// Signature of the runtime publish hook: called with the bucket name and -/// its parsed config (`None` when absent, cleared, or unreadable) every time -/// the bucket's metadata is installed into or removed from the cache. -pub type ConfigPublishHook = Box) + Send + Sync>; - -/// Registration point for the runtime (`OnDemandMigrationSys`). Until it is -/// set, metadata publishes are no-ops for ODM, so this crate carries no -/// runtime dependency and the config layer stays inert. -pub static ON_DEMAND_MIGRATION_CONFIG_HOOK: OnceLock = OnceLock::new(); - #[cfg(test)] mod tests { use super::*; @@ -1501,4 +1509,55 @@ mod tests { assert!(!rendered.contains("topsecret"), "{rendered}"); assert!(!rendered.contains("SK"), "{rendered}"); } + /// rustfs/backlog#2148: the accessor reports absence as `Ok(None)` and a + /// stored payload it cannot parse as a typed error, never as a default + /// and never as `ConfigNotFound`. + #[tokio::test] + async fn get_on_demand_migration_config_distinguishes_absent_from_corrupt() { + use super::super::storage_api::StorageError as Error; + use super::super::storage_api::test_support::{ + BUCKET_ON_DEMAND_MIGRATION_CONFIG, BucketMetadata, BucketMetadataSys, isolated_store_over_temp_disks, + }; + use std::sync::Arc; + const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#; + + let (_dirs, ecstore) = isolated_store_over_temp_disks().await; + let sys = BucketMetadataSys::new(ecstore); + let bucket = "odm-accessor"; + + sys.set(bucket.to_string(), Arc::new(BucketMetadata::new(bucket))).await; + assert_eq!( + decode_stored_config(sys.get_on_demand_migration_config(bucket).await.unwrap()).unwrap(), + None + ); + + let mut corrupt = BucketMetadata::new(bucket); + corrupt.on_demand_migration_config_json = br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec(); + sys.set(bucket.to_string(), Arc::new(corrupt)).await; + let err = decode_stored_config(sys.get_on_demand_migration_config(bucket).await.unwrap()) + .expect_err("corrupt config must not read as a default"); + assert_ne!(err, Error::ConfigNotFound, "corruption must not be reported as absence"); + let typed = match &err { + Error::Io(io) => io + .get_ref() + .and_then(|source| source.downcast_ref::()), + _ => None, + }; + assert!( + matches!(typed, Some(OnDemandMigrationConfigError::Malformed(_))), + "typed parse error must survive the Result boundary, got: {err:?}" + ); + + let mut valid = BucketMetadata::new(bucket); + valid + .update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec()) + .unwrap(); + let stamped = valid.on_demand_migration_config_updated_at; + sys.set(bucket.to_string(), Arc::new(valid)).await; + let (config, updated_at) = decode_stored_config(sys.get_on_demand_migration_config(bucket).await.unwrap()) + .unwrap() + .expect("stored config is returned"); + assert_eq!(config, OnDemandMigrationConfig::from_json(ODM_JSON).unwrap()); + assert_eq!(updated_at, stamped); + } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/gcs.rs b/rustfs/src/on_demand_migration/gcs.rs similarity index 98% rename from crates/ecstore/src/bucket/on_demand_migration/gcs.rs rename to rustfs/src/on_demand_migration/gcs.rs index 874ddb0e3..bce647b97 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/gcs.rs +++ b/rustfs/src/on_demand_migration/gcs.rs @@ -43,8 +43,8 @@ use super::source_client::{ GcsSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceTimeouts, range_header_value, }; -use crate::bucket::remote_s3_client::RemoteS3ClientError; -use crate::storage_api_contracts::range::HTTPRangeSpec; +use super::storage_api::HTTPRangeSpec; +use super::storage_api::remote_s3_client::RemoteS3ClientError; use google_cloud_auth::credentials::service_account::{AccessSpecifier, Builder as ServiceAccountBuilder}; use google_cloud_auth::credentials::{CacheableResource, Credentials}; use http::{HeaderMap, HeaderValue, Method}; @@ -318,8 +318,8 @@ fn parse_objects_list(body: &str) -> Result { #[cfg(test)] mod tests { use super::*; - use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract}; - use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; + use crate::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract}; + use crate::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder; const LIST_PAGE_ONE: &str = r#"{ diff --git a/crates/ecstore/src/bucket/on_demand_migration/list_through.rs b/rustfs/src/on_demand_migration/list_through.rs similarity index 100% rename from crates/ecstore/src/bucket/on_demand_migration/list_through.rs rename to rustfs/src/on_demand_migration/list_through.rs diff --git a/rustfs/src/on_demand_migration/metrics.rs b/rustfs/src/on_demand_migration/metrics.rs new file mode 100644 index 000000000..cf6e035cc --- /dev/null +++ b/rustfs/src/on_demand_migration/metrics.rs @@ -0,0 +1,147 @@ +// 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. + +//! Projection of application runtime state onto observability-owned DTOs. + +use crate::on_demand_migration::backfill::{ + BackfillCheckpoint as SourceBackfillCheckpoint, global_backfill_runner as source_global_backfill_runner, +}; +use crate::on_demand_migration::{ + BreakerState as SourceOdmBreakerState, OdmBucketSnapshot as SourceOdmBucketSnapshot, + OnDemandMigrationSys as SourceOnDemandMigrationSys, +}; +use rustfs_obs::metrics::{ + OdmBackfillBucketStats, OnDemandMigrationBreakerState, OnDemandMigrationBucketStats, + register_on_demand_migration_metrics_source, +}; + +pub(super) fn register() { + register_on_demand_migration_metrics_source(snapshot, backfill_snapshot); +} + +fn on_demand_migration_stats_from_snapshot(snapshot: SourceOdmBucketSnapshot) -> 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. +fn snapshot() -> Vec { + SourceOnDemandMigrationSys::get() + .snapshot() + .into_iter() + .map(on_demand_migration_stats_from_snapshot) + .collect() +} + +fn on_demand_migration_backfill_stats_from_checkpoint( + bucket: String, + checkpoint: SourceBackfillCheckpoint, +) -> OdmBackfillBucketStats { + OdmBackfillBucketStats { + bucket, + state: checkpoint.state.as_str().to_string(), + listed: checkpoint.listed, + enqueued: checkpoint.enqueued, + pulled: checkpoint.pulled, + skipped_existing: checkpoint.skipped_existing, + failed: checkpoint.failed, + bytes: checkpoint.bytes, + } +} + +/// Backfill jobs running on this node, sorted by bucket. Empty until the +/// runner is installed, and empty again once a job finishes: the series are +/// per-node job progress, not a cluster-wide history. +fn backfill_snapshot() -> Vec { + source_global_backfill_runner() + .map(|runner| { + runner + .local_job_snapshots() + .into_iter() + .map(|(bucket, checkpoint)| on_demand_migration_backfill_stats_from_checkpoint(bucket, checkpoint)) + .collect() + }) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn on_demand_migration_snapshot_projects_counters_and_breaker_state() { + // Pin the runtime wire snapshot and its observability projection together. + 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); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/mod.rs b/rustfs/src/on_demand_migration/mod.rs similarity index 84% rename from crates/ecstore/src/bucket/on_demand_migration/mod.rs rename to rustfs/src/on_demand_migration/mod.rs index 554dadd60..823f37e99 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/rustfs/src/on_demand_migration/mod.rs @@ -33,11 +33,13 @@ pub mod config; #[cfg(feature = "gcs")] pub mod gcs; pub mod list_through; +mod metrics; mod native_http; pub mod negative_cache; pub mod pull; pub mod source_client; pub mod stats; +mod storage_api; pub mod sys; #[cfg(test)] mod test_http_fixture; @@ -47,9 +49,9 @@ pub use breaker::{ BreakerState, BreakerTransition, BreakerVerdict, }; pub use config::{ - AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, - ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, - RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, + AzureSourceConfig, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, + OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig, SourceCredentials, + SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, }; pub use list_through::{ FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, @@ -63,6 +65,9 @@ pub use pull::{ PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, SourceIdleGuard, WriteBackBody, WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, idle_guarded_body, }; +pub use source_client::{ + SourceClient, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceSse, is_multipart_etag, +}; pub use stats::{ GaugeGuard, LastSourceError, LatencyBucketSnapshot, OdmOp, OdmOutcome, OdmStats, OdmStatsSnapshot, PullFailureReason, PullPath, SOURCE_LATENCY_BUCKET_BOUNDS_MS, SourceLatencySnapshot, @@ -72,3 +77,7 @@ pub use sys::{ OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec, source_client_spec, }; + +pub(crate) fn register_metrics() { + metrics::register(); +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs b/rustfs/src/on_demand_migration/native_http.rs similarity index 99% rename from crates/ecstore/src/bucket/on_demand_migration/native_http.rs rename to rustfs/src/on_demand_migration/native_http.rs index 8e5c12dd4..ae084d084 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs +++ b/rustfs/src/on_demand_migration/native_http.rs @@ -26,7 +26,7 @@ //! the log line and the admin response. use super::source_client::{SourceError, SourceHead, SourceTimeouts, USER_AGENT_SUFFIX, classify_status, is_multipart_etag}; -use crate::bucket::remote_s3_client::{RemoteS3ClientError, validate_remote_endpoint, validate_target_ca_pem}; +use super::storage_api::remote_s3_client::{RemoteS3ClientError, validate_remote_endpoint, validate_target_ca_pem}; use aws_sdk_s3::primitives::ByteStream; use aws_smithy_types::body::SdkBody; use futures::StreamExt; diff --git a/crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs b/rustfs/src/on_demand_migration/negative_cache.rs similarity index 100% rename from crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs rename to rustfs/src/on_demand_migration/negative_cache.rs diff --git a/crates/ecstore/src/bucket/on_demand_migration/pull.rs b/rustfs/src/on_demand_migration/pull.rs similarity index 99% rename from crates/ecstore/src/bucket/on_demand_migration/pull.rs rename to rustfs/src/on_demand_migration/pull.rs index 8d18bdbdd..9d52af009 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/pull.rs +++ b/rustfs/src/on_demand_migration/pull.rs @@ -1093,7 +1093,7 @@ impl OnDemandMigrationSys { #[cfg(test)] mod tests { use super::*; - use crate::bucket::on_demand_migration::config::{ + use crate::on_demand_migration::config::{ FilterConfig, OnDemandMigrationConfig, PathStyle as ConfigPathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, TlsConfig, }; @@ -1499,7 +1499,7 @@ mod tests { assert_eq!(failures(&state).get("queue_full"), Some(&1)); assert!(!queue.is_stopped()); - assert_eq!(sys.remove(BUCKET), crate::bucket::on_demand_migration::ApplyOutcome::Removed); + assert_eq!(sys.remove(BUCKET), crate::on_demand_migration::ApplyOutcome::Removed); tokio::time::timeout(Duration::from_secs(5), queue.wait_until_stopped()) .await .expect("dispatcher and in-flight job must exit after cancel"); diff --git a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs b/rustfs/src/on_demand_migration/source_client.rs similarity index 99% rename from crates/ecstore/src/bucket/on_demand_migration/source_client.rs rename to rustfs/src/on_demand_migration/source_client.rs index d2fc21c3e..9f3e050c9 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/rustfs/src/on_demand_migration/source_client.rs @@ -29,10 +29,10 @@ use super::azure::AzureSourceBackend; #[cfg(feature = "gcs")] use super::gcs::GcsNativeSourceBackend; use super::list_through::{ListPageError, validate_list_page}; -use crate::bucket::remote_s3_client::{ +use super::storage_api::HTTPRangeSpec; +use super::storage_api::remote_s3_client::{ PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config, }; -use crate::storage_api_contracts::range::HTTPRangeSpec; use aws_sdk_s3::Client as S3Client; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::get_object::GetObjectOutput; @@ -995,7 +995,7 @@ fn s3_source_object(object: SdkObject) -> Result { #[cfg(test)] mod tests { use super::*; - use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract}; + use crate::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract}; use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn}; use aws_smithy_runtime_api::client::orchestrator::HttpRequest; use aws_smithy_runtime_api::client::result::ConnectorError; diff --git a/crates/ecstore/src/bucket/on_demand_migration/stats.rs b/rustfs/src/on_demand_migration/stats.rs similarity index 100% rename from crates/ecstore/src/bucket/on_demand_migration/stats.rs rename to rustfs/src/on_demand_migration/stats.rs diff --git a/rustfs/src/on_demand_migration/storage_api.rs b/rustfs/src/on_demand_migration/storage_api.rs new file mode 100644 index 000000000..d7917ad29 --- /dev/null +++ b/rustfs/src/on_demand_migration/storage_api.rs @@ -0,0 +1,28 @@ +// 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. + +//! Storage capabilities used by on-demand migration orchestration. + +#[cfg(test)] +pub(super) use crate::storage_api::on_demand_migration::test_support; +pub(super) use crate::storage_api::on_demand_migration::{ + BUCKET_CONFIG_PUBLISH_HOOK, BUCKET_META_PREFIX, BUCKET_ON_DEMAND_MIGRATION_CONFIG, ECStore, HTTPPreconditions, HTTPRangeSpec, + NamespaceLocking, ObjectOperations, ObjectOptions, RUSTFS_META_BUCKET, StorageError, WriteCompletion, + get_lock_acquire_timeout, get_on_demand_migration_config, get_on_demand_migration_config_in, read_config_with_metadata, + remote_s3_client, save_config_with_opts, +}; + +pub(super) async fn local_node_name() -> String { + rustfs_common::get_global_local_node_name().await +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/sys.rs b/rustfs/src/on_demand_migration/sys.rs similarity index 96% rename from crates/ecstore/src/bucket/on_demand_migration/sys.rs rename to rustfs/src/on_demand_migration/sys.rs index ab42bd602..d529594d4 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/sys.rs +++ b/rustfs/src/on_demand_migration/sys.rs @@ -19,7 +19,7 @@ //! [`SourceClient`], a circuit breaker, a negative cache, a per-key //! singleflight table, a pull concurrency limit and counters. Its lifecycle //! follows the bucket metadata cache through the publish hook registered in -//! [`ON_DEMAND_MIGRATION_CONFIG_HOOK`]; the hook fires on every cache install +//! [`BUCKET_CONFIG_PUBLISH_HOOK`]; the hook fires on every cache install //! path (initial load, admin update, peer reload, refresh loop, lazy load). //! //! Change detection compares the config by value (`PartialEq`) rather than @@ -41,9 +41,7 @@ use super::backfill::{PriorityPullPermits, PullPermit, PullPriority}; use super::breaker::{Breaker, BreakerState, BreakerTransition, BreakerVerdict}; -use super::config::{ - ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig, PathStyle as ConfigPathStyle, Provider, SourceConfig, -}; +use super::config::{OnDemandMigrationConfig, PathStyle as ConfigPathStyle, Provider, SourceConfig}; use super::list_through::{SOURCE_LIST_RATE_PER_SEC, SourceListRateLimiter}; use super::negative_cache::NegativeCache; use super::pull::{OdmWriteBack, PullQueue}; @@ -52,9 +50,10 @@ use super::source_client::{ SourceTimeouts, }; use super::stats::{GaugeGuard, OdmStats, OdmStatsSnapshot, PullFailureReason}; -use crate::bucket::remote_s3_client::{ +use super::storage_api::remote_s3_client::{ PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy, }; +use super::storage_api::{BUCKET_CONFIG_PUBLISH_HOOK, BUCKET_ON_DEMAND_MIGRATION_CONFIG}; use parking_lot::{Mutex, RwLock}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -740,11 +739,34 @@ impl OnDemandMigrationSys { /// Registers `publish` as the bucket-metadata publish hook. Returns /// `false` when a hook was already registered. pub fn register_config_hook(&'static self) -> bool { - ON_DEMAND_MIGRATION_CONFIG_HOOK - .set(Box::new(move |bucket, config| self.publish(bucket, config))) + BUCKET_CONFIG_PUBLISH_HOOK + .set(Box::new(move |bucket, config_file, stored| { + if config_file == BUCKET_ON_DEMAND_MIGRATION_CONFIG { + self.publish_stored(bucket, stored.map(|(bytes, _)| bytes)); + } + })) .is_ok() } + /// Corrupt persisted bytes withdraw state synchronously, just like deletion. + fn publish_stored(&'static self, bucket: &str, stored: Option<&[u8]>) { + match stored.map(OnDemandMigrationConfig::from_json).transpose() { + Ok(config) => self.publish(bucket, config.as_ref()), + Err(err) => { + warn!( + event = EVENT_ODM_BUCKET_STATE_APPLIED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + result = "invalid", + bucket = %bucket, + error = %err, + "Failed to parse on-demand migration config" + ); + self.publish(bucket, None); + } + } + } + /// Hook entry point: removals apply immediately, installs are spawned /// (client construction is async). Requires a Tokio runtime for the /// install path; without one the config is logged and skipped. @@ -951,8 +973,8 @@ impl OnDemandMigrationSys { #[cfg(test)] mod tests { use super::*; - use crate::bucket::on_demand_migration::breaker::BREAKER_FAILURE_THRESHOLD; - use crate::bucket::on_demand_migration::config::{FilterConfig, PolicyConfig, SourceCredentials, SourceTimeout, TlsConfig}; + use crate::on_demand_migration::breaker::BREAKER_FAILURE_THRESHOLD; + use crate::on_demand_migration::config::{FilterConfig, PolicyConfig, SourceCredentials, SourceTimeout, TlsConfig}; use std::sync::atomic::AtomicUsize; use tokio::sync::Barrier; @@ -1344,6 +1366,17 @@ mod tests { assert!(state.is_cancelled()); } + #[tokio::test] + async fn corrupt_stored_config_withdraws_runtime_state() { + let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys())); + let cfg = config(None); + assert_eq!(sys.apply("corrupt", Some(&cfg)).await, ApplyOutcome::Installed); + let state = sys.state("corrupt").expect("state installed"); + sys.publish_stored("corrupt", Some(b"not-json")); + assert!(sys.state("corrupt").is_none(), "corruption cannot keep an older source active"); + assert!(state.is_cancelled(), "corruption cancels in-flight work"); + } + #[tokio::test] async fn absent_config_updates_do_not_allocate_bucket_slots() { let sys = enabled_sys(); diff --git a/crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs b/rustfs/src/on_demand_migration/test_http_fixture.rs similarity index 100% rename from crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs rename to rustfs/src/on_demand_migration/test_http_fixture.rs diff --git a/rustfs/src/startup_background.rs b/rustfs/src/startup_background.rs index 01b0886de..393b7ad3d 100644 --- a/rustfs/src/startup_background.rs +++ b/rustfs/src/startup_background.rs @@ -17,10 +17,11 @@ use crate::module_switches::{ bitrot_selftest_enabled_from_env, bitrot_selftest_strict_from_env, heal_enabled_from_env, is_on_demand_migration_module_enabled, scanner_enabled_from_env, }; -use crate::storage_api::startup::background::{ - BackfillRunner, ECStore, OnDemandMigrationSys, SysBackfillContexts, install_global_backfill_runner, - set_workload_admission_snapshot_provider, spawn_backfill_recovery_loop, +use crate::on_demand_migration::OnDemandMigrationSys; +use crate::on_demand_migration::backfill::{ + BackfillRunner, SysBackfillContexts, install_global_backfill_runner, spawn_backfill_recovery_loop, }; +use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider}; use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider; use rustfs_concurrency::WorkloadAdmissionSnapshotProvider; use rustfs_heal::{ diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index 1c3cdb270..a85d71ab3 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -14,10 +14,11 @@ use crate::app::object::OnDemandMigrationWriteBack; use crate::module_switches::{on_demand_migration_enabled_from_env, set_on_demand_migration_module_enabled}; +use crate::on_demand_migration::OnDemandMigrationSys; use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::bucket_metadata::{ - ECStore, Error as StorageError, OnDemandMigrationSys, Result as StorageResult, get_global_replication_pool, - init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, + ECStore, Error as StorageError, Result as StorageResult, get_global_replication_pool, init_bucket_metadata_sys, + reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, }; use std::{ io::{Error as IoError, Result as IoResult}, @@ -98,6 +99,7 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance /// `OnDemandMigrationSys` with a usable write-back (rustfs/backlog#2152). /// Idempotent across embedded and server startups. fn init_on_demand_migration_runtime() { + crate::on_demand_migration::register_metrics(); let enabled = on_demand_migration_enabled_from_env(); set_on_demand_migration_module_enabled(enabled); let sys = OnDemandMigrationSys::get(); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 2216db7f4..fe9a55641 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -407,8 +407,8 @@ pub(crate) mod ecstore_bucket { #[cfg(test)] pub(crate) use rustfs_ecstore::api::bucket::lifecycle::tier_delete_journal::test_util::install_all_v6_fleet_capability_proof; pub(crate) use rustfs_ecstore::api::bucket::{ - bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, on_demand_migration, - policy_sys, remote_s3_client, replication, tagging, target, utils, + bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, policy_sys, + remote_s3_client, replication, tagging, target, utils, }; pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys}; } @@ -466,10 +466,10 @@ pub(crate) mod ecstore_data_usage { #[allow(unused_imports)] pub(crate) mod ecstore_disk { pub(crate) use rustfs_ecstore::api::disk::{ - BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore, - FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, - ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, - get_object_disk_read_timeout, validate_batch_read_version_item_count, + BUCKET_META_PREFIX, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, + DiskInfoOptions, DiskStore, FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, + RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, + UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, validate_batch_read_version_item_count, }; pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce}; } @@ -560,7 +560,7 @@ pub(crate) mod ecstore_object { pub(crate) use rustfs_ecstore::api::object::{ EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, ObjectEncryptionResolver, ObjectMutationHook, PrepareSelectObjectSnapshotError, ReadEncryptionMaterial, - ReadEncryptionMode, ReadEncryptionRequest, SelectObjectSnapshot, get_object_body_cache_plaintext_len, + ReadEncryptionMode, ReadEncryptionRequest, SelectObjectSnapshot, WriteCompletion, get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook, }; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 4f824f8da..60de2c850 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -278,10 +278,6 @@ pub(crate) mod startup { } pub(crate) mod background { - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::OnDemandMigrationSys; - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::backfill::{ - BackfillRunner, SysBackfillContexts, install_global_backfill_runner, spawn_backfill_recovery_loop, - }; pub(crate) use crate::storage::storage_api::{ BitrotSelfTestError, ECStore, bitrot_self_test, set_workload_admission_snapshot_provider, }; @@ -294,7 +290,6 @@ pub(crate) mod startup { } } - pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::OnDemandMigrationSys; pub(crate) use crate::storage::storage_api::{ ECStore, Error, Result, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config, @@ -415,3 +410,30 @@ pub(crate) mod table { get_lock_acquire_timeout, table_catalog_path_hash, }; } + +pub(crate) mod on_demand_migration { + pub(crate) use crate::storage::storage_api::ECStore; + pub(crate) use crate::storage::storage_api::StorageObjectOptions as ObjectOptions; + pub(crate) use crate::storage::storage_api::contract::{ + namespace::NamespaceLocking, object::HTTPPreconditions, object::ObjectOperations, range::HTTPRangeSpec, + }; + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG; + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::{ + BUCKET_CONFIG_PUBLISH_HOOK, get_on_demand_migration_config, get_on_demand_migration_config_in, + }; + pub(crate) use crate::storage::storage_api::ecstore_bucket::remote_s3_client; + pub(crate) use crate::storage::storage_api::ecstore_config::com::{read_config_with_metadata, save_config_with_opts}; + pub(crate) use crate::storage::storage_api::ecstore_disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; + pub(crate) use crate::storage::storage_api::ecstore_error::Error as StorageError; + pub(crate) use crate::storage::storage_api::ecstore_object::WriteCompletion; + pub(crate) use crate::storage::storage_api::ecstore_set_disk::get_lock_acquire_timeout; + + #[cfg(test)] + pub(crate) mod test_support { + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::{ + BUCKET_ON_DEMAND_MIGRATION_CONFIG, BucketMetadata, + }; + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::BucketMetadataSys; + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::test_support::isolated_store_over_temp_disks; + } +}