gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 03:59:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(odm): move migration orchestration into application (#7226)

Browse Source 
* refactor(odm): move migration orchestration into application

* style(odm): format relocated listing test imports
This commit is contained in:
Zhengchao An
2026-09-06 01:31:58 +08:00
committed by GitHub
parent 14cef91423
commit 8fc1c9281e
45 changed files with 619 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -1
View File
@@ -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",
crates/ecstore/Cargo.toml
-1
View File
@@ -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"] }
crates/ecstore/src/api/mod.rs
+13 -59
View File
@@ -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,
crates/ecstore/src/bucket/metadata.rs
+25 -64
View File
@@ -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::OnDemandMigrationConfig>,
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<super::durability::BucketDurabilityConfig> {
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");
crates/ecstore/src/bucket/metadata_sys.rs
+54 -98
View File
@@ -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<dyn Fn(&str, &str, Option<(&[u8], OffsetDateTime)>) + Send + Sync>;
pub static BUCKET_CONFIG_PUBLISH_HOOK: std::sync::OnceLock<BucketConfigPublishHook> = 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<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
/// `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<Option<(Vec<u8>, 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<Option<(Vec<u8>, 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<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
pub async fn get_on_demand_migration_config(&self, bucket: &str) -> Result<Option<(Vec<u8>, 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<tempfile::TempDir>, Arc<ECStore>) {
pub async fn isolated_store_over_temp_disks() -> (Vec<tempfile::TempDir>, Arc<ECStore>) {
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<Vec<(String, Option<OnDemandMigrationConfig>)>> = std::sync::Mutex::new(Vec::new());
static ODM_HOOK_CALLS: std::sync::Mutex<Vec<(String, Option<(Vec<u8>, 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<Option<OnDemandMigrationConfig>> {
fn odm_hook_calls(bucket: &str) -> Vec<Option<(Vec<u8>, 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::<OnDemandMigrationConfigError>()),
_ => 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]
crates/ecstore/src/bucket/mod.rs
-1
View File
@@ -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;
crates/ecstore/src/bucket/remote_s3_client.rs
+2 -4
View File
@@ -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<aws_sdk_s3::config::Builder, RemoteS3ClientError> {
pub async fn build_remote_s3_config(spec: &RemoteS3EndpointSpec) -> Result<aws_sdk_s3::config::Builder, RemoteS3ClientError> {
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)
}
crates/obs/src/metrics/mod.rs
+1
View File
@@ -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;
crates/obs/src/metrics/storage_api.rs
+53 -115
View File
@@ -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<ObsBucketRepl
buckets
}
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,
},
}
struct OnDemandMigrationMetricsSource {
snapshot: fn() -> Vec<OnDemandMigrationBucketStats>,
backfill_snapshot: fn() -> Vec<OdmBackfillBucketStats>,
}
/// 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<OnDemandMigrationBucketStats> {
SourceOnDemandMigrationSys::get()
.snapshot()
.into_iter()
.map(on_demand_migration_stats_from_snapshot)
.collect()
}
static ON_DEMAND_MIGRATION_METRICS_SOURCE: std::sync::OnceLock<OnDemandMigrationMetricsSource> = 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<OnDemandMigrationBucketStats>,
backfill_snapshot: fn() -> Vec<OdmBackfillBucketStats>,
) -> bool {
ON_DEMAND_MIGRATION_METRICS_SOURCE
.set(OnDemandMigrationMetricsSource {
snapshot,
backfill_snapshot,
})
.is_ok()
}
pub(crate) fn obs_on_demand_migration_snapshot() -> Vec<OnDemandMigrationBucketStats> {
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(
docs/architecture/background-services-inventory.md
+3 -3
View File
@@ -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.
docs/architecture/crate-boundaries.md
+25
View File
@@ -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.
docs/architecture/ecstore-api-facade-inventory.md
+7
View File
@@ -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::<owner>::{...}` 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.
docs/architecture/remote-credential-sealing-adr.md
+4 -4
View File
@@ -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
docs/operations/on-demand-migration.md
+3 -3
View File
@@ -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://<account>.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.
rustfs/Cargo.toml
+4 -1
View File
@@ -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"] }
rustfs/src/admin/handlers/on_demand_migration.rs
+26 -13
View File
@@ -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(&params)?;
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(&params)?;
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)
rustfs/src/admin/storage_api.rs
+2 -38
View File
@@ -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<Option<(super::on_demand_migration::OnDemandMigrationConfig, OffsetDateTime)>> {
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;
rustfs/src/app/bucket_list_through.rs
+9 -9
View File
@@ -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};
rustfs/src/app/object/get.rs
+4 -4
View File
@@ -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<S: OdmGetSource>(
#[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;
rustfs/src/app/object/head.rs
+5 -7
View File
@@ -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,
}
rustfs/src/app/object/on_demand_migration_put.rs
+3 -3
View File
@@ -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};
rustfs/src/app/object/shared.rs
+1 -3
View File
@@ -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";
rustfs/src/app/storage_api.rs
-33
View File
@@ -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;
rustfs/src/lib.rs
+1
View File
@@ -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;
crates/ecstore/src/bucket/on_demand_migration/azure.rs → rustfs/src/on_demand_migration/azure.rs
+5 -5
View File
@@ -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#"<?xml version="1.0" encoding="utf-8"?>
<EnumerationResults ServiceEndpoint="https://acct.blob.core.windows.net/" ContainerName="legacy">
crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs → rustfs/src/on_demand_migration/backend_contract.rs
+1 -1
View File
@@ -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.
crates/ecstore/src/bucket/on_demand_migration/backfill.rs → rustfs/src/on_demand_migration/backfill.rs
+13 -19
View File
@@ -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<Option<OffsetDateTime>, 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<BackfillRunner>, 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;
crates/ecstore/src/bucket/on_demand_migration/breaker.rs → rustfs/src/on_demand_migration/breaker.rs
View File
crates/ecstore/src/bucket/on_demand_migration/config.rs → rustfs/src/on_demand_migration/config.rs
+71 -12
View File
@@ -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<u8>, time::OffsetDateTime)>,
) -> Result<Option<(OnDemandMigrationConfig, time::OffsetDateTime)>, 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<Option<(OnDemandMigrationConfig, time::OffsetDateTime)>, 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<dyn Fn(&str, Option<&OnDemandMigrationConfig>) + 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<ConfigPublishHook> = 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::<OnDemandMigrationConfigError>()),
_ => 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);
}
}
crates/ecstore/src/bucket/on_demand_migration/gcs.rs → rustfs/src/on_demand_migration/gcs.rs
+4 -4
View File
@@ -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<SourcePage, SourceError> {
#[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#"{
crates/ecstore/src/bucket/on_demand_migration/list_through.rs → rustfs/src/on_demand_migration/list_through.rs
View File
rustfs/src/on_demand_migration/metrics.rs
+147
View File
@@ -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<OnDemandMigrationBucketStats> {
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<OdmBackfillBucketStats> {
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);
}
}
crates/ecstore/src/bucket/on_demand_migration/mod.rs → rustfs/src/on_demand_migration/mod.rs
+12 -3
View File
@@ -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();
}
crates/ecstore/src/bucket/on_demand_migration/native_http.rs → rustfs/src/on_demand_migration/native_http.rs
+1 -1
View File
@@ -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;
crates/ecstore/src/bucket/on_demand_migration/negative_cache.rs → rustfs/src/on_demand_migration/negative_cache.rs
View File
crates/ecstore/src/bucket/on_demand_migration/pull.rs → rustfs/src/on_demand_migration/pull.rs
+2 -2
View File
@@ -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");
crates/ecstore/src/bucket/on_demand_migration/source_client.rs → rustfs/src/on_demand_migration/source_client.rs
+3 -3
View File
@@ -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<SourceObject, SourceError> {
#[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;
crates/ecstore/src/bucket/on_demand_migration/stats.rs → rustfs/src/on_demand_migration/stats.rs
View File
rustfs/src/on_demand_migration/storage_api.rs
+28
View File
@@ -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
}
crates/ecstore/src/bucket/on_demand_migration/sys.rs → rustfs/src/on_demand_migration/sys.rs
+42 -9
View File
@@ -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();
crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs → rustfs/src/on_demand_migration/test_http_fixture.rs
View File
rustfs/src/startup_background.rs
+4 -3
View File
@@ -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::{
rustfs/src/startup_bucket_metadata.rs
+4 -2
View File
@@ -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<ECStore>, 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();
rustfs/src/storage/storage_api.rs
+7 -7
View File
@@ -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,
};
rustfs/src/storage_api.rs
+27 -5
View File
@@ -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;
}
}