mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 03:59:14 +00:00
feat(ecstore): add on-demand migration bucket config model (#7061)
* feat(ecstore): add on-demand migration bucket config model Introduce OnDemandMigrationConfig (deny_unknown_fields, version 1) with typed validation, credential redaction, a secret-free Debug impl, and the OnceLock publish hook the runtime registers into. Exported through the api facade. * feat(ecstore): persist on-demand migration config in bucket metadata Store the config as a RustFS extension entry (on-demand-migration.json) with its update time in .metadata.bin, add the typed BucketMetadataSys accessor, and publish the config through the hook on every cache-install path alongside the durability sync. * test(e2e): rename stall timing variable flagged by typos * test(storage): heap-pin the RestoreObject usecase future in the generation guard test
This commit is contained in:
Zhengchao An
@@ -128,7 +128,6 @@ pub mod bucket {
|
||||
}
|
||||
|
||||
pub mod metadata {
|
||||
pub use crate::bucket::metadata::BUCKET_DURABILITY_CONFIG;
|
||||
pub use crate::bucket::metadata::{
|
||||
BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG,
|
||||
BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_QUOTA_CONFIG_FILE,
|
||||
@@ -137,6 +136,7 @@ pub mod bucket {
|
||||
BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BUCKET_WEBSITE_CONFIG, BucketMetadata, OBJECT_LOCK_CONFIG,
|
||||
load_bucket_metadata, table_catalog_path_hash,
|
||||
};
|
||||
pub use crate::bucket::metadata::{BUCKET_DURABILITY_CONFIG, BUCKET_ON_DEMAND_MIGRATION_CONFIG};
|
||||
}
|
||||
|
||||
pub mod durability {
|
||||
@@ -145,6 +145,21 @@ pub mod bucket {
|
||||
};
|
||||
}
|
||||
|
||||
pub mod on_demand_migration {
|
||||
pub use crate::bucket::on_demand_migration::{
|
||||
ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
|
||||
OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
|
||||
SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
|
||||
};
|
||||
pub mod source_client {
|
||||
pub use crate::bucket::on_demand_migration::source_client::{
|
||||
SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, 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;
|
||||
@@ -154,11 +169,11 @@ pub mod bucket {
|
||||
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_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_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,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -166,16 +181,6 @@ pub mod bucket {
|
||||
pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
|
||||
}
|
||||
|
||||
pub mod on_demand_migration {
|
||||
pub mod source_client {
|
||||
pub use crate::bucket::on_demand_migration::source_client::{
|
||||
SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
|
||||
SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
|
||||
resolve_path_style,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
pub mod object_lock {
|
||||
pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
|
||||
|
||||
|
||||
@@ -270,6 +270,7 @@ pub const BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG: &str = "public-access-block.xml";
|
||||
pub const BUCKET_ACL_CONFIG: &str = "bucket-acl.json";
|
||||
pub const BUCKET_TABLE_CONFIG: &str = "table-bucket.json";
|
||||
pub const BUCKET_DURABILITY_CONFIG: &str = "durability.json";
|
||||
pub const BUCKET_ON_DEMAND_MIGRATION_CONFIG: &str = "on-demand-migration.json";
|
||||
pub const BUCKET_TABLE_RESERVED_PREFIX: &str = ".rustfs-table";
|
||||
pub const BUCKET_TABLE_CATALOG_META_PREFIX: &str = "s3tables/catalog";
|
||||
pub const BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX: &str = "table-buckets";
|
||||
@@ -321,6 +322,7 @@ pub struct BucketMetadata {
|
||||
pub bucket_acl_config_json: Vec<u8>,
|
||||
pub table_bucket_config_json: Vec<u8>,
|
||||
pub durability_config_json: Vec<u8>,
|
||||
pub on_demand_migration_config_json: Vec<u8>,
|
||||
|
||||
pub policy_config_updated_at: OffsetDateTime,
|
||||
pub object_lock_config_updated_at: OffsetDateTime,
|
||||
@@ -342,6 +344,7 @@ pub struct BucketMetadata {
|
||||
pub bucket_acl_config_updated_at: OffsetDateTime,
|
||||
pub table_bucket_config_updated_at: OffsetDateTime,
|
||||
pub durability_config_updated_at: OffsetDateTime,
|
||||
pub on_demand_migration_config_updated_at: OffsetDateTime,
|
||||
|
||||
pub new_field_updated_at: OffsetDateTime,
|
||||
|
||||
@@ -393,6 +396,7 @@ impl Default for BucketMetadata {
|
||||
bucket_acl_config_json: Default::default(),
|
||||
table_bucket_config_json: Default::default(),
|
||||
durability_config_json: Default::default(),
|
||||
on_demand_migration_config_json: Default::default(),
|
||||
policy_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
object_lock_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
encryption_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
@@ -413,6 +417,7 @@ impl Default for BucketMetadata {
|
||||
bucket_acl_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
table_bucket_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
durability_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
on_demand_migration_config_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
new_field_updated_at: OffsetDateTime::UNIX_EPOCH,
|
||||
policy_config: Default::default(),
|
||||
notification_config: Default::default(),
|
||||
@@ -477,6 +482,23 @@ impl BucketMetadata {
|
||||
/// 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)
|
||||
}
|
||||
|
||||
pub fn durability_config(&self) -> Option<super::durability::BucketDurabilityConfig> {
|
||||
if self.durability_config_json.is_empty() {
|
||||
return None;
|
||||
@@ -555,6 +577,9 @@ impl BucketMetadata {
|
||||
"BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?,
|
||||
"TableBucketConfigJSON" | "TableBucketConfigJson" => self.table_bucket_config_json = read_msgp_bin(rd)?,
|
||||
"DurabilityConfigJSON" | "DurabilityConfigJson" => self.durability_config_json = read_msgp_bin(rd)?,
|
||||
"OnDemandMigrationConfigJSON" | "OnDemandMigrationConfigJson" => {
|
||||
self.on_demand_migration_config_json = read_msgp_bin(rd)?
|
||||
}
|
||||
"CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?,
|
||||
"LoggingConfigUpdatedAt" => self.logging_config_updated_at = read_msgp_time_value(rd)?,
|
||||
"WebsiteConfigUpdatedAt" => self.website_config_updated_at = read_msgp_time_value(rd)?,
|
||||
@@ -564,6 +589,7 @@ impl BucketMetadata {
|
||||
"BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?,
|
||||
"TableBucketConfigUpdatedAt" => self.table_bucket_config_updated_at = read_msgp_time_value(rd)?,
|
||||
"DurabilityConfigUpdatedAt" => self.durability_config_updated_at = read_msgp_time_value(rd)?,
|
||||
"OnDemandMigrationConfigUpdatedAt" => self.on_demand_migration_config_updated_at = read_msgp_time_value(rd)?,
|
||||
other => {
|
||||
tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field");
|
||||
skip_msgp_value(rd)?;
|
||||
@@ -576,8 +602,8 @@ impl BucketMetadata {
|
||||
|
||||
/// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility.
|
||||
pub fn encode_to<W: Write>(&self, wr: &mut W) -> Result<()> {
|
||||
// Map size: MinIO fields (25) + RustFS extensions (19)
|
||||
let map_len: u32 = 44;
|
||||
// Map size: MinIO fields (25) + RustFS extensions (21)
|
||||
let map_len: u32 = 46;
|
||||
rmp::encode::write_map_len(wr, map_len)?;
|
||||
|
||||
// MinIO field order (same as Go struct)
|
||||
@@ -637,6 +663,7 @@ impl BucketMetadata {
|
||||
write_bin_field(wr, "BucketAclConfigJSON", &self.bucket_acl_config_json)?;
|
||||
write_bin_field(wr, "TableBucketConfigJSON", &self.table_bucket_config_json)?;
|
||||
write_bin_field(wr, "DurabilityConfigJSON", &self.durability_config_json)?;
|
||||
write_bin_field(wr, "OnDemandMigrationConfigJSON", &self.on_demand_migration_config_json)?;
|
||||
rmp::encode::write_str(wr, "CorsConfigUpdatedAt")?;
|
||||
write_msgp_time(wr, self.cors_config_updated_at)?;
|
||||
rmp::encode::write_str(wr, "LoggingConfigUpdatedAt")?;
|
||||
@@ -655,6 +682,8 @@ impl BucketMetadata {
|
||||
write_msgp_time(wr, self.table_bucket_config_updated_at)?;
|
||||
rmp::encode::write_str(wr, "DurabilityConfigUpdatedAt")?;
|
||||
write_msgp_time(wr, self.durability_config_updated_at)?;
|
||||
rmp::encode::write_str(wr, "OnDemandMigrationConfigUpdatedAt")?;
|
||||
write_msgp_time(wr, self.on_demand_migration_config_updated_at)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -756,6 +785,9 @@ impl BucketMetadata {
|
||||
if self.durability_config_updated_at == OffsetDateTime::UNIX_EPOCH {
|
||||
self.durability_config_updated_at = self.created
|
||||
}
|
||||
if self.on_demand_migration_config_updated_at == OffsetDateTime::UNIX_EPOCH {
|
||||
self.on_demand_migration_config_updated_at = self.created
|
||||
}
|
||||
}
|
||||
|
||||
pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
|
||||
@@ -871,6 +903,17 @@ impl BucketMetadata {
|
||||
self.durability_config_json = data;
|
||||
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;
|
||||
}
|
||||
_ => return Err(Error::other(format!("config file not found : {config_file}"))),
|
||||
}
|
||||
|
||||
@@ -1779,6 +1822,117 @@ mod test {
|
||||
assert!(!bm.table_bucket_enabled());
|
||||
}
|
||||
|
||||
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.
|
||||
#[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();
|
||||
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())));
|
||||
|
||||
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;
|
||||
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!(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(_))));
|
||||
}
|
||||
|
||||
/// rustfs/backlog#2148: a `.metadata.bin` written before the on-demand
|
||||
/// migration keys existed decodes with an empty blob and an epoch
|
||||
/// timestamp that `default_timestamps` back-fills from `created`.
|
||||
#[test]
|
||||
fn on_demand_migration_config_absent_in_legacy_blob_defaults_to_created() {
|
||||
let blob = decode_hex(include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex"));
|
||||
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));
|
||||
|
||||
bm.default_timestamps();
|
||||
assert_ne!(bm.created, OffsetDateTime::UNIX_EPOCH, "fixture must carry a real creation time");
|
||||
assert_eq!(bm.on_demand_migration_config_updated_at, bm.created);
|
||||
|
||||
// A metadata blob from this build with no config set stays
|
||||
// indistinguishable from the legacy one for these fields.
|
||||
let fresh = BucketMetadata::unmarshal(&BucketMetadata::new("fresh").marshal_msg().unwrap()).unwrap();
|
||||
assert!(fresh.on_demand_migration_config_json.is_empty());
|
||||
assert_eq!(fresh.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#2148: a reader that predates the two on-demand
|
||||
/// migration keys takes `decode_from`'s unknown-field branch, which is
|
||||
/// `skip_msgp_value`. Walk the new-format blob with exactly that
|
||||
/// primitive and prove both keys are skipped without desynchronising the
|
||||
/// stream, so the fields that follow them still decode.
|
||||
#[test]
|
||||
fn old_decoder_skips_on_demand_migration_fields_without_desync() {
|
||||
let mut bm = BucketMetadata::new("odm-skip");
|
||||
bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
|
||||
.unwrap();
|
||||
bm.update_config(BUCKET_DURABILITY_CONFIG, br#"{"mode":"relaxed"}"#.to_vec())
|
||||
.unwrap();
|
||||
let buf = bm.marshal_msg().unwrap();
|
||||
|
||||
let mut rd = std::io::Cursor::new(buf.as_slice());
|
||||
let fields = rmp::decode::read_map_len(&mut rd).unwrap();
|
||||
let mut skipped = Vec::new();
|
||||
let mut durability_json = Vec::new();
|
||||
for _ in 0..fields {
|
||||
let key_len = rmp::decode::read_str_len(&mut rd).unwrap();
|
||||
let mut key = vec![0u8; key_len as usize];
|
||||
rd.read_exact(&mut key).unwrap();
|
||||
let key = String::from_utf8(key).unwrap();
|
||||
match key.as_str() {
|
||||
// The field an old reader knows that is encoded *after* the
|
||||
// unknown JSON key and *before* the unknown timestamp key.
|
||||
"DurabilityConfigJSON" => durability_json = read_msgp_bin(&mut rd).unwrap(),
|
||||
other => {
|
||||
if other.starts_with("OnDemandMigration") {
|
||||
skipped.push(other.to_string());
|
||||
}
|
||||
skip_msgp_value(&mut rd).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
assert_eq!(skipped, ["OnDemandMigrationConfigJSON", "OnDemandMigrationConfigUpdatedAt"]);
|
||||
assert_eq!(durability_json, br#"{"mode":"relaxed"}"#);
|
||||
assert_eq!(rd.position() as usize, buf.len(), "old-style walk must consume the blob exactly");
|
||||
}
|
||||
|
||||
/// HP-5b (rustfs/backlog#938): the durability override is a RustFS
|
||||
/// extension entry and must survive an encode/decode round trip.
|
||||
#[test]
|
||||
|
||||
@@ -19,6 +19,7 @@ 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};
|
||||
@@ -384,6 +385,42 @@ 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.
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get(bucket: &str) -> Result<Arc<BucketMetadata>> {
|
||||
let sys = get_bucket_metadata_sys()?;
|
||||
let lock = sys.read().await;
|
||||
@@ -970,6 +1007,16 @@ pub async fn get_durability_config(
|
||||
Ok((bm.durability_config(), bm.durability_config_updated_at))
|
||||
}
|
||||
|
||||
/// 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)>> {
|
||||
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
|
||||
}
|
||||
|
||||
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;
|
||||
@@ -1492,6 +1539,7 @@ impl BucketMetadataSys {
|
||||
if removed {
|
||||
BucketTargetSys::get().delete(bucket).await;
|
||||
clear_bucket_durability(bucket);
|
||||
clear_on_demand_migration(bucket);
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
@@ -1529,6 +1577,7 @@ impl BucketMetadataSys {
|
||||
self.missing_buckets.invalidate(bucket).await;
|
||||
sync_bucket_target_sys(bucket, &bm).await;
|
||||
sync_bucket_durability(bucket, &bm);
|
||||
sync_on_demand_migration(bucket, &bm);
|
||||
}
|
||||
MetadataLoadMode::Initial => {
|
||||
let _publish_guard = self
|
||||
@@ -1575,6 +1624,7 @@ impl BucketMetadataSys {
|
||||
if removed {
|
||||
BucketTargetSys::get().delete(bucket).await;
|
||||
clear_bucket_durability(bucket);
|
||||
clear_on_demand_migration(bucket);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
@@ -1597,6 +1647,7 @@ impl BucketMetadataSys {
|
||||
self.missing_buckets.invalidate(bucket).await;
|
||||
sync_bucket_target_sys(bucket, &metadata).await;
|
||||
sync_bucket_durability(bucket, &metadata);
|
||||
sync_on_demand_migration(bucket, &metadata);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -1624,6 +1675,7 @@ impl BucketMetadataSys {
|
||||
self.missing_buckets.invalidate(&bucket).await;
|
||||
sync_bucket_target_sys(&bucket, &bm).await;
|
||||
sync_bucket_durability(&bucket, &bm);
|
||||
sync_on_demand_migration(&bucket, &bm);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1644,6 +1696,7 @@ impl BucketMetadataSys {
|
||||
if removed {
|
||||
BucketTargetSys::get().delete(bucket).await;
|
||||
clear_bucket_durability(bucket);
|
||||
clear_on_demand_migration(bucket);
|
||||
}
|
||||
removed || removed_fabricated
|
||||
}
|
||||
@@ -1933,6 +1986,7 @@ impl BucketMetadataSys {
|
||||
self.missing_buckets.invalidate(bucket).await;
|
||||
sync_bucket_target_sys(bucket, &bm).await;
|
||||
sync_bucket_durability(bucket, &bm);
|
||||
sync_on_demand_migration(bucket, &bm);
|
||||
} else {
|
||||
let exists = self
|
||||
.bucket_exists(bucket, &guard, "lazy bucket metadata existence check")
|
||||
@@ -2271,6 +2325,7 @@ impl BucketMetadataSys {
|
||||
self.missing_buckets.invalidate(bucket).await;
|
||||
sync_bucket_target_sys(bucket, &metadata).await;
|
||||
sync_bucket_durability(bucket, &metadata);
|
||||
sync_on_demand_migration(bucket, &metadata);
|
||||
Ok(BucketMetadataAuthority::Authoritative(metadata))
|
||||
}
|
||||
|
||||
@@ -2463,6 +2518,17 @@ impl BucketMetadataSys {
|
||||
Err(Error::ConfigNotFound)
|
||||
}
|
||||
}
|
||||
|
||||
/// See [`get_on_demand_migration_config`].
|
||||
pub async fn get_on_demand_migration_config(
|
||||
&self,
|
||||
bucket: &str,
|
||||
) -> Result<Option<(OnDemandMigrationConfig, 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)))
|
||||
}
|
||||
}
|
||||
|
||||
/// Test-only fixture shared with sibling modules (e.g. the quota checker
|
||||
@@ -4043,6 +4109,151 @@ mod tests {
|
||||
assert_eq!(bucket_durability::lookup(bucket), None);
|
||||
}
|
||||
|
||||
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"}}}"#;
|
||||
|
||||
/// 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());
|
||||
|
||||
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()));
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
fn odm_hook_calls(bucket: &str) -> Vec<Option<OnDemandMigrationConfig>> {
|
||||
ODM_HOOK_CALLS
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|(name, _)| name == bucket)
|
||||
.map(|(_, config)| config.clone())
|
||||
.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
|
||||
/// `sync_bucket_durability`.
|
||||
#[tokio::test]
|
||||
async fn on_demand_migration_hook_fires_on_every_cache_install_path() {
|
||||
install_recording_odm_hook();
|
||||
|
||||
let (dirs, ecstore) = isolated_store_over_temp_disks().await;
|
||||
let bucket = "odm-hook-paths";
|
||||
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");
|
||||
};
|
||||
|
||||
// set (via persist_new_and_set, which installs through `set`).
|
||||
let mut bm = BucketMetadata::new(bucket);
|
||||
bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
|
||||
.unwrap();
|
||||
let writer = BucketMetadataSys::new(ecstore.clone());
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
writer.persist_new_and_set(bm).await.expect("metadata should persist");
|
||||
expect_publish(before, "set");
|
||||
|
||||
// init (initial load on a cold system).
|
||||
let mut cold = BucketMetadataSys::new(ecstore.clone());
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
cold.init(vec![bucket.to_string()]).await;
|
||||
assert!(cold.get(bucket).await.is_ok(), "initial load must cache the bucket");
|
||||
expect_publish(before, "init");
|
||||
|
||||
// peer reload.
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
cold.reload_from_store(bucket).await.expect("peer reload should publish");
|
||||
expect_publish(before, "peer reload");
|
||||
|
||||
// refresh loop.
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
let mut failed = HashSet::new();
|
||||
cold.concurrent_load(&[bucket.to_string()], &mut failed, MetadataLoadMode::Refresh)
|
||||
.await;
|
||||
assert!(failed.is_empty(), "refresh must succeed");
|
||||
expect_publish(before, "refresh loop");
|
||||
|
||||
// lazy load on another cold system.
|
||||
let lazy = BucketMetadataSys::new(ecstore);
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
let (_, loaded) = lazy.get_config(bucket).await.expect("lazy load should publish");
|
||||
assert!(loaded, "the lazy path must have gone to disk");
|
||||
expect_publish(before, "lazy load");
|
||||
|
||||
// Removal withdraws the config.
|
||||
let before = odm_hook_calls(bucket).len();
|
||||
assert!(lazy.remove(bucket).await);
|
||||
let calls = odm_hook_calls(bucket);
|
||||
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.
|
||||
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");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_wait_exits_when_cancelled() {
|
||||
let cancel_token = CancellationToken::new();
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -12,7 +12,17 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! On-demand migration (ODM): serve and back-fill objects from an external
|
||||
//! S3-compatible source bucket.
|
||||
//! On-Demand Migration (ODM): a bucket can name an external S3-compatible
|
||||
//! source bucket; GET misses are served from that source and backfilled
|
||||
//! locally. This module owns the bucket-level configuration model
|
||||
//! (`on-demand-migration.json` in the bucket metadata file); the runtime is
|
||||
//! layered on top of it by later tasks (rustfs/backlog#2147).
|
||||
|
||||
pub mod config;
|
||||
pub mod source_client;
|
||||
|
||||
pub use config::{
|
||||
ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
|
||||
OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
|
||||
SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
|
||||
};
|
||||
|
||||
@@ -4361,10 +4361,15 @@ mod tests {
|
||||
apply_bucket_generation_guard(&req, &bucket, &mut opts).expect("apply the RestoreObject authorization guard");
|
||||
assert_eq!(opts.expected_bucket_incarnation_id, Some(authorized_incarnation_id));
|
||||
|
||||
let err = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(app_context))
|
||||
.execute_restore_object(req)
|
||||
.await
|
||||
.expect_err("the old RestoreObject authorization must not reach the recreated bucket");
|
||||
// The RestoreObject usecase future is large enough that, inlined into
|
||||
// this test body, the test thread's 2 MiB stack sits within a few KiB
|
||||
// of overflowing on Linux; heap-pin it so unrelated growth in bucket
|
||||
// metadata futures cannot tip the test over.
|
||||
let err = Box::pin(
|
||||
crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(app_context)).execute_restore_object(req),
|
||||
)
|
||||
.await
|
||||
.expect_err("the old RestoreObject authorization must not reach the recreated bucket");
|
||||
assert_eq!(err.code(), &S3ErrorCode::NoSuchBucket);
|
||||
store
|
||||
.delete_bucket(&bucket, &DeleteBucketOptions::default())
|
||||
|
||||
Reference in New Issue
Block a user