diff --git a/README.md b/README.md index c58af61d..a87c0c28 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,14 @@ rc admin service-account update local/ SAKEY123 --policy ./service-account-polic rc admin access-key info local/ AKIAIOSFODNN7EXAMPLE rc admin access-key info local/ AKIAIOSFODNN7EXAMPLE --json +# On-demand migration: serve misses from an external source bucket and backfill the rest +RC_ODM_SECRET_KEY=... rc admin bucket migration set local/photos --provider minio \ + --endpoint https://source.example.com:9000 --region us-east-1 \ + --source-bucket legacy-photos --access-key AKIASOURCE --dry-run +rc admin bucket migration status local/photos +rc admin bucket migration backfill start local/photos +rc admin bucket migration backfill status local/photos --watch + # Manage bucket event notifications rc event add local/my-bucket arn:aws:sns:us-east-1:123456789012:topic --event 's3:ObjectCreated:*' rc event list local/my-bucket diff --git a/crates/cli/src/commands/admin/bucket.rs b/crates/cli/src/commands/admin/bucket.rs new file mode 100644 index 00000000..c5a363aa --- /dev/null +++ b/crates/cli/src/commands/admin/bucket.rs @@ -0,0 +1,1704 @@ +//! Per-bucket administrative features. +//! +//! `rc admin bucket migration` manages On-Demand Migration: a bucket names an +//! external S3-compatible source bucket, a GET that misses locally is served +//! from that source and stored locally, and a background backfill job pulls +//! the rest. The wire contract is pinned by the fixtures under +//! `crates/core/tests/fixtures/on_demand_migration/`. + +use clap::{Args, Subcommand, ValueEnum}; +use rc_core::admin::{ + BackfillJob, BackfillStartRequest, HeadPolicy, MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES, + ON_DEMAND_MIGRATION_CAPABILITY, OnDemandMigrationApi, OnDemandMigrationConfigRequest, + OnDemandMigrationConfigView, OnDemandMigrationSetResult, OnDemandMigrationStatus, PathStyle, + REDACTED_SECRET, RangeGetPolicy, SkipExisting, SourceCredentialsRequest, SourceErrorPolicy, + SourceProvider, SourceRequest, TlsRequest, validate_ca_cert_pem, validate_local_bucket, +}; +use rc_core::{Error, Result}; +use serde::Serialize; +use serde_json::{Value, json}; +use std::io::Read; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use zeroize::Zeroizing; + +use crate::exit_code::ExitCode; +use crate::output::Formatter; +use crate::secret_input::{SecretSource, can_prompt}; + +/// Environment variable read for the source secret key when `--secret-key` is absent. +pub const SECRET_KEY_ENV: &str = "RC_ODM_SECRET_KEY"; + +const OUTPUT_TYPE: &str = "on_demand_migration"; + +/// Rendered for a ratio the server did not compute. Never zero: a missing +/// ratio and a zero ratio mean different things. +const NO_VALUE: &str = "\u{2014}"; + +#[derive(Subcommand, Debug)] +pub enum BucketCommands { + /// Serve misses from an external S3-compatible source bucket and migrate it in place + #[command(subcommand)] + Migration(MigrationCommands), +} + +#[derive(Subcommand, Debug)] +pub enum MigrationCommands { + /// Validate, probe and save the source configuration, replacing any existing one + Set(SetArgs), + /// Show the saved configuration with credentials redacted + Get(TargetArgs), + /// Remove the configuration; already-pulled objects stay in place + Rm(TargetArgs), + /// Show the answering node's runtime status: hit ratio, pulls, breaker, last error + Status(StatusArgs), + /// Control the background backfill job + #[command(subcommand)] + Backfill(BackfillCommands), +} + +#[derive(Subcommand, Debug)] +pub enum BackfillCommands { + /// Walk the source listing and pull every object that is missing locally + Start(BackfillStartArgs), + /// Ask the running job to stop at its next checkpoint + Cancel(TargetArgs), + /// Show the job checkpoint + Status(StatusArgs), +} + +#[derive(Args, Debug)] +pub struct TargetArgs { + /// Local bucket as alias/bucket + pub target: String, +} + +#[derive(Args, Debug)] +pub struct StatusArgs { + #[command(flatten)] + pub target: TargetArgs, + /// Refresh until interrupted, or until a backfill job reaches a terminal state + #[arg(long)] + pub watch: bool, + /// Seconds between refreshes with --watch + #[arg(long, default_value_t = 2, value_parser = clap::value_parser!(u64).range(1..=3600))] + pub interval: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ProviderArg { + S3, + Aws, + Minio, + Rustfs, + R2, + Gcs, +} + +impl From for SourceProvider { + fn from(value: ProviderArg) -> Self { + match value { + ProviderArg::S3 => Self::S3, + ProviderArg::Aws => Self::Aws, + ProviderArg::Minio => Self::Minio, + ProviderArg::Rustfs => Self::Rustfs, + ProviderArg::R2 => Self::R2, + ProviderArg::Gcs => Self::Gcs, + } + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +pub enum PathStyleArg { + #[default] + Auto, + Path, + Virtual, +} + +impl From for PathStyle { + fn from(value: PathStyleArg) -> Self { + match value { + PathStyleArg::Auto => Self::Auto, + PathStyleArg::Path => Self::Path, + PathStyleArg::Virtual => Self::Virtual, + } + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +#[value(rename_all = "snake_case")] +pub enum HeadArg { + #[default] + Proxy, + LocalOnly, +} + +impl From for HeadPolicy { + fn from(value: HeadArg) -> Self { + match value { + HeadArg::Proxy => Self::Proxy, + HeadArg::LocalOnly => Self::LocalOnly, + } + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +#[value(rename_all = "snake_case")] +pub enum RangeGetArg { + #[default] + ServeAndBackfill, + ServeOnly, +} + +impl From for RangeGetPolicy { + fn from(value: RangeGetArg) -> Self { + match value { + RangeGetArg::ServeAndBackfill => Self::ServeAndBackfill, + RangeGetArg::ServeOnly => Self::ServeOnly, + } + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +#[value(rename_all = "snake_case")] +pub enum SourceErrorArg { + #[default] + Propagate, + NotFound, +} + +impl From for SourceErrorPolicy { + fn from(value: SourceErrorArg) -> Self { + match value { + SourceErrorArg::Propagate => Self::Propagate, + SourceErrorArg::NotFound => Self::NotFound, + } + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)] +#[value(rename_all = "snake_case")] +pub enum SkipExistingArg { + #[default] + Always, + EtagOrSize, +} + +impl From for SkipExisting { + fn from(value: SkipExistingArg) -> Self { + match value { + SkipExistingArg::Always => Self::Always, + SkipExistingArg::EtagOrSize => Self::EtagOrSize, + } + } +} + +/// A secret taken from the command line, kept in zeroizing storage and never +/// shown by `Debug`. +#[derive(Clone)] +pub struct SecretArg(Zeroizing); + +impl std::fmt::Debug for SecretArg { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("SecretArg([REDACTED])") + } +} + +fn parse_secret_arg(value: &str) -> std::result::Result { + Ok(SecretArg(Zeroizing::new(value.to_string()))) +} + +#[derive(Args, Debug)] +pub struct SetArgs { + #[command(flatten)] + pub target: TargetArgs, + /// Source vendor family; drives addressing defaults + #[arg(long, value_enum)] + pub provider: ProviderArg, + /// Source endpoint as scheme://host[:port]; derived from --region for aws + #[arg(long, value_name = "URL")] + pub endpoint: Option, + /// Source signing region; 'auto' is accepted for r2, minio and rustfs + #[arg(long, value_name = "R")] + pub region: String, + /// Bucket on the source + #[arg(long, value_name = "B")] + pub source_bucket: String, + /// Only local keys with this prefix consult the source + #[arg(long, value_name = "P")] + pub prefix: Option, + /// Prepended to the local key to form the source key + #[arg(long, value_name = "SP")] + pub source_prefix: Option, + /// Source access key + #[arg(long, value_name = "AK", conflicts_with = "public")] + pub access_key: Option, + /// Source secret key; prefer RC_ODM_SECRET_KEY or the hidden prompt, which stay out of shell history + #[arg(long, value_name = "SK", requires = "access_key", conflicts_with = "public", value_parser = parse_secret_arg)] + pub secret_key: Option, + /// Read the source anonymously + #[arg(long)] + pub public: bool, + /// Bucket addressing style + #[arg(long, value_enum, default_value_t = PathStyleArg::Auto)] + pub path_style: PathStyleArg, + /// Do not verify the source TLS certificate + #[arg(long)] + pub skip_tls_verify: bool, + /// PEM CA bundle used to verify the source TLS certificate + #[arg(long, value_name = "FILE")] + pub ca_cert: Option, + /// What a HEAD that misses locally does + #[arg(long, value_enum, default_value_t = HeadArg::Proxy)] + pub head: HeadArg, + /// Whether a Range GET also queues a whole-object background pull + #[arg(long, value_enum, default_value_t = RangeGetArg::ServeAndBackfill)] + pub range_get: RangeGetArg, + /// How a source failure is answered to the client + #[arg(long, value_enum, default_value_t = SourceErrorArg::Propagate)] + pub source_error: SourceErrorArg, + /// Do not keep the source ETag on stored objects + #[arg(long)] + pub no_preserve_etag: bool, + /// Copy source object tags; costs one extra source call per inline pull + #[arg(long)] + pub copy_tags: bool, + /// Do not emit ObjectCreated notifications for write-backs + #[arg(long)] + pub no_events: bool, + /// Largest object teed inline on a GET miss; larger objects stream through and pull in the background + #[arg(long, value_name = "N")] + pub inline_max_bytes: Option, + /// Pull concurrency shared by inline and background paths (1-256) + #[arg(long, value_name = "N")] + pub max_concurrent_pulls: Option, + /// Validate the configuration and probe the source without saving + #[arg(long)] + pub dry_run: bool, +} + +#[derive(Args, Debug)] +pub struct BackfillStartArgs { + #[command(flatten)] + pub target: TargetArgs, + /// Only walk source keys with this prefix + #[arg(long, value_name = "P")] + pub prefix: Option, + /// When a local object counts as already migrated + #[arg(long, value_enum, default_value_t = SkipExistingArg::Always)] + pub skip_existing: SkipExistingArg, + /// List and count only; nothing is queued + #[arg(long)] + pub dry_run: bool, +} + +// --------------------------------------------------------------------------- +// Preparation: everything that can fail before the network +// --------------------------------------------------------------------------- + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +enum Operation { + Set, + Get, + Remove, + Status, + BackfillStart, + BackfillCancel, + BackfillStatus, +} + +impl Operation { + /// Reads can be retried by a caller; a write is a second decision. + const fn is_read(self) -> bool { + matches!(self, Self::Get | Self::Status | Self::BackfillStatus) + } +} + +#[derive(Debug)] +enum Action { + Set { + request: Box, + dry_run: bool, + }, + Get, + Remove, + Status { + watch: Option, + }, + BackfillStart(BackfillStartRequest), + BackfillCancel, + BackfillStatus { + watch: Option, + }, +} + +impl Action { + const fn operation(&self) -> Operation { + match self { + Self::Set { .. } => Operation::Set, + Self::Get => Operation::Get, + Self::Remove => Operation::Remove, + Self::Status { .. } => Operation::Status, + Self::BackfillStart(_) => Operation::BackfillStart, + Self::BackfillCancel => Operation::BackfillCancel, + Self::BackfillStatus { .. } => Operation::BackfillStatus, + } + } +} + +#[derive(Debug)] +struct Prepared { + alias: String, + bucket: String, + action: Action, +} + +/// `alias/bucket`, nothing more: a key or a trailing slash is a usage error. +fn parse_target(target: &str) -> Result<(String, String)> { + let (alias, bucket) = target + .split_once('/') + .ok_or_else(|| Error::InvalidPath("Expected alias/bucket".into()))?; + if alias.is_empty() { + return Err(Error::InvalidPath("Expected alias/bucket".into())); + } + validate_local_bucket(bucket)?; + Ok((alias.to_string(), bucket.to_string())) +} + +fn watch_interval(args: &StatusArgs) -> Option { + args.watch.then(|| Duration::from_secs(args.interval)) +} + +/// Where the secret comes from, in order: `--secret-key`, `RC_ODM_SECRET_KEY`, +/// then a hidden prompt when there is a terminal and output is human-readable. +/// +/// The server replaces a saved secret with `REDACTED` in every response and a +/// `set` replaces the configuration wholesale, so the placeholder is refused: +/// saving it would overwrite a working credential with the literal word. +fn resolve_secret( + args: &SetArgs, + environment_secret: Option, + formatter: &Formatter, +) -> Result>> { + if args.public { + return Ok(None); + } + if args.access_key.is_none() { + return Err(Error::Config( + "Provide --access-key (with the secret from RC_ODM_SECRET_KEY, --secret-key or the prompt) or --public".into(), + )); + } + let secret = if let Some(SecretArg(secret)) = &args.secret_key { + secret.clone() + } else if let Some(value) = environment_secret { + let value = Zeroizing::new(value.to_string_lossy().into_owned()); + let trimmed = Zeroizing::new(value.trim_end_matches(['\r', '\n']).to_string()); + if trimmed.is_empty() { + return Err(Error::Config(format!("{SECRET_KEY_ENV} is set but empty"))); + } + trimmed + } else if can_prompt(formatter.is_json()) { + SecretSource::Prompt.load("Source secret key: ")? + } else { + return Err(Error::Config(format!( + "Provide the source secret key with {SECRET_KEY_ENV} (or --secret-key) when running non-interactively or with --json" + ))); + }; + if secret.as_str() == REDACTED_SECRET { + return Err(Error::Config( + "The secret key is the redaction placeholder; set replaces the configuration, so pass the real secret again".into(), + )); + } + Ok(Some(secret)) +} + +fn read_ca_cert(path: &Path) -> Result { + let mut file = std::fs::File::open(path).map_err(|error| { + Error::Config(format!( + "Failed to read --ca-cert '{}': {error}", + path.display() + )) + })?; + let mut bytes = Vec::new(); + file.by_ref() + .take(MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES as u64 + 1) + .read_to_end(&mut bytes)?; + if bytes.len() > MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES { + return Err(Error::Config(format!( + "--ca-cert exceeds {MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES} bytes" + ))); + } + let pem = String::from_utf8(bytes) + .map_err(|_| Error::Config("--ca-cert must be a UTF-8 PEM file".into()))?; + validate_ca_cert_pem(&pem)?; + Ok(pem) +} + +fn build_request( + args: &SetArgs, + secret: Option>, +) -> Result { + let credentials = match (&args.access_key, secret) { + (Some(access_key), Some(secret_key)) => Some(SourceCredentialsRequest { + access_key: access_key.clone(), + secret_key, + session_token: None, + }), + _ => None, + }; + let ca_cert_pem = args.ca_cert.as_deref().map(read_ca_cert).transpose()?; + let mut request = OnDemandMigrationConfigRequest::new(SourceRequest { + provider: args.provider.into(), + endpoint: args.endpoint.clone(), + region: args.region.clone(), + bucket: args.source_bucket.clone(), + path_style: args.path_style.into(), + credentials, + tls: TlsRequest { + skip_verify: args.skip_tls_verify, + ca_cert_pem, + }, + }); + request.filter.prefix = args.prefix.clone(); + request.filter.source_prefix = args.source_prefix.clone(); + let policy = &mut request.policy; + policy.head = args.head.into(); + policy.range_get = args.range_get.into(); + policy.source_error = args.source_error.into(); + policy.preserve_etag = !args.no_preserve_etag; + policy.copy_tags = args.copy_tags; + policy.emit_events = !args.no_events; + if let Some(value) = args.inline_max_bytes { + policy.inline_max_bytes = value; + } + if let Some(value) = args.max_concurrent_pulls { + policy.max_concurrent_pulls = value; + } + request.validate()?; + Ok(request) +} + +fn prepare(command: MigrationCommands, formatter: &Formatter) -> Result { + let (target, action) = match command { + MigrationCommands::Set(args) => { + let (alias, bucket) = parse_target(&args.target.target)?; + // Validate everything that does not need the secret first, so a + // typo never costs the operator a prompt. + build_request(&args, None)?; + let secret = resolve_secret(&args, std::env::var_os(SECRET_KEY_ENV), formatter)?; + let request = build_request(&args, secret)?; + return Ok(Prepared { + alias, + bucket, + action: Action::Set { + request: Box::new(request), + dry_run: args.dry_run, + }, + }); + } + MigrationCommands::Get(args) => (args.target, Action::Get), + MigrationCommands::Rm(args) => (args.target, Action::Remove), + MigrationCommands::Status(args) => { + let watch = watch_interval(&args); + (args.target.target, Action::Status { watch }) + } + MigrationCommands::Backfill(BackfillCommands::Start(args)) => { + let request = BackfillStartRequest { + prefix: args.prefix, + skip_existing: Some(args.skip_existing.into()), + dry_run: args.dry_run, + }; + request.validate()?; + (args.target.target, Action::BackfillStart(request)) + } + MigrationCommands::Backfill(BackfillCommands::Cancel(args)) => { + (args.target, Action::BackfillCancel) + } + MigrationCommands::Backfill(BackfillCommands::Status(args)) => { + let watch = watch_interval(&args); + (args.target.target, Action::BackfillStatus { watch }) + } + }; + let (alias, bucket) = parse_target(&target)?; + Ok(Prepared { + alias, + bucket, + action, + }) +} + +// --------------------------------------------------------------------------- +// Execution +// --------------------------------------------------------------------------- + +pub async fn execute(command: BucketCommands, formatter: &Formatter) -> ExitCode { + match command { + BucketCommands::Migration(command) => execute_migration(command, formatter).await, + } +} + +const fn command_operation(command: &MigrationCommands) -> Operation { + match command { + MigrationCommands::Set(_) => Operation::Set, + MigrationCommands::Get(_) => Operation::Get, + MigrationCommands::Rm(_) => Operation::Remove, + MigrationCommands::Status(_) => Operation::Status, + MigrationCommands::Backfill(BackfillCommands::Start(_)) => Operation::BackfillStart, + MigrationCommands::Backfill(BackfillCommands::Cancel(_)) => Operation::BackfillCancel, + MigrationCommands::Backfill(BackfillCommands::Status(_)) => Operation::BackfillStatus, + } +} + +async fn execute_migration(command: MigrationCommands, formatter: &Formatter) -> ExitCode { + let operation = command_operation(&command); + let prepared = match prepare(command, formatter) { + Ok(prepared) => prepared, + Err(error) => return emit_error(&error, operation, formatter), + }; + let client = rc_core::AliasManager::new() + .and_then(|aliases| aliases.get(&prepared.alias)) + .and_then(|alias| rc_s3::AdminClient::new(&alias)); + let client = match client { + Ok(client) => client, + Err(error) => return emit_error(&error, prepared.action.operation(), formatter), + }; + execute_with_api(prepared, &client, formatter).await +} + +async fn execute_with_api( + prepared: Prepared, + api: &dyn OnDemandMigrationApi, + formatter: &Formatter, +) -> ExitCode { + let operation = prepared.action.operation(); + let bucket = prepared.bucket; + let result = match prepared.action { + Action::Set { request, dry_run } => api + .set_on_demand_migration(&bucket, &request, dry_run) + .await + .map(|result| { + render_set(&bucket, &result, dry_run, formatter); + serde_json::to_value(result) + }), + Action::Get => api.get_on_demand_migration(&bucket).await.map(|result| { + if !formatter.is_json() { + match &result.config { + Some(config) => { + render_config(&bucket, config, result.updated_at.as_deref(), formatter); + } + None => formatter.println(&format!( + "No on-demand migration configuration returned for '{}'", + formatter.sanitize_text(&bucket) + )), + } + } + serde_json::to_value(result) + }), + Action::Remove => api.delete_on_demand_migration(&bucket).await.map(|()| { + formatter.success(&format!( + "On-demand migration removed from '{}'; already-pulled objects stay in place", + formatter.sanitize_text(&bucket) + )); + Ok(json!({"bucket": bucket.as_str(), "removed": true})) + }), + Action::Status { watch: None } => { + api.on_demand_migration_status(&bucket).await.map(|status| { + render_status(&bucket, &status, formatter); + serde_json::to_value(status) + }) + } + Action::Status { + watch: Some(interval), + } => watch_status(api, &bucket, interval, formatter).await, + Action::BackfillStart(request) => api + .start_on_demand_migration_backfill(&bucket, &request) + .await + .map(|result| { + if !formatter.is_json() { + let verb = if request.dry_run { + "Backfill dry run started" + } else { + "Backfill started" + }; + formatter.success(&format!( + "{verb} for '{}'", + formatter.sanitize_text(&bucket) + )); + render_backfill(&bucket, result.job.as_ref(), formatter); + } + serde_json::to_value(result) + }), + Action::BackfillCancel => { + api.cancel_on_demand_migration_backfill(&bucket) + .await + .map(|result| { + if !formatter.is_json() { + formatter.success(&format!( + "Backfill cancellation requested for '{}'", + formatter.sanitize_text(&bucket) + )); + render_backfill(&bucket, result.job.as_ref(), formatter); + } + serde_json::to_value(result) + }) + } + Action::BackfillStatus { watch: None } => api + .on_demand_migration_backfill_status(&bucket) + .await + .map(|result| { + if !formatter.is_json() { + render_backfill(&bucket, result.job.as_ref(), formatter); + } + serde_json::to_value(result) + }), + Action::BackfillStatus { + watch: Some(interval), + } => watch_backfill(api, &bucket, interval, formatter).await, + }; + match result { + Ok(Ok(value)) => { + // Watch mode already streamed one record per refresh. + if formatter.is_json() && !value.is_null() { + formatter.json(&success_output(operation, &bucket, value)); + } + ExitCode::Success + } + Ok(Err(error)) => emit_error(&Error::Json(error), operation, formatter), + Err(error) => emit_error(&error, operation, formatter), + } +} + +/// Refresh the runtime status until interrupted. Human output redraws the +/// full block each tick; JSON output is one record per tick. +async fn watch_status( + api: &dyn OnDemandMigrationApi, + bucket: &str, + interval: Duration, + formatter: &Formatter, +) -> Result> { + loop { + let status = api.on_demand_migration_status(bucket).await?; + if formatter.is_json() { + formatter.json_line(&success_output( + Operation::Status, + bucket, + serde_json::to_value(&status)?, + )); + } else { + render_status(bucket, &status, formatter); + formatter.println(""); + } + tokio::time::sleep(interval).await; + } +} + +/// Refresh one progress line until the job reaches a terminal state, then +/// print the final checkpoint. JSON output is one record per tick. +async fn watch_backfill( + api: &dyn OnDemandMigrationApi, + bucket: &str, + interval: Duration, + formatter: &Formatter, +) -> Result> { + let term = console::Term::stderr(); + loop { + let result = api.on_demand_migration_backfill_status(bucket).await?; + let terminal = result.job.as_ref().is_none_or(BackfillJob::is_terminal); + if formatter.is_json() { + formatter.json_line(&success_output( + Operation::BackfillStatus, + bucket, + serde_json::to_value(&result)?, + )); + } else if let Some(job) = &result.job { + // The line is stderr so stdout stays clean for the final document. + let _ = term.clear_line(); + let _ = term.write_str(&formatter.sanitize_text(&backfill_progress_line(job))); + } + if terminal { + if !formatter.is_json() { + let _ = term.write_line(""); + render_backfill(bucket, result.job.as_ref(), formatter); + } + // The stream already carried the final record. + return Ok(Ok(Value::Null)); + } + tokio::time::sleep(interval).await; + } +} + +// --------------------------------------------------------------------------- +// Output +// --------------------------------------------------------------------------- + +fn success_output(operation: Operation, bucket: &str, result: Value) -> Value { + json!({ + "schema_version": 3, + "type": OUTPUT_TYPE, + "status": "success", + "data": {"operation": operation, "bucket": bucket, "result": result}, + }) +} + +fn error_output(error: &Error, operation: Operation) -> Value { + let kind = match error.exit_code() { + 2 => "usage_error", + 3 => "network_error", + 4 => "auth_error", + 5 => "not_found", + 6 => "conflict", + 7 => "unsupported_feature", + 130 => "interrupted", + _ => "general_error", + }; + let mut detail = json!({ + "type": kind, + "message": error.to_string(), + "retryable": matches!(error, Error::Network(_)) && operation.is_read(), + }); + if let Some(suggestion) = suggestion(error, operation) { + detail["suggestion"] = json!(suggestion); + } + if error.exit_code() == 7 { + detail["capability"] = json!(ON_DEMAND_MIGRATION_CAPABILITY); + detail["server"] = Value::Null; + } + json!({"schema_version": 3, "type": OUTPUT_TYPE, "status": "error", "error": detail}) +} + +fn suggestion(error: &Error, operation: Operation) -> Option<&'static str> { + match error { + Error::Network(_) if operation == Operation::Set => Some( + "Check the source endpoint, region, bucket and credentials, then retry with --dry-run.", + ), + Error::Network(_) => Some("Verify the endpoint and network connectivity, then retry."), + Error::Auth(_) => Some( + "Verify admin permissions (admin:SetBucketOnDemandMigration / admin:GetBucketOnDemandMigration) and the server licence.", + ), + Error::Conflict(_) => { + Some("A backfill job already holds the lease; cancel it or wait for it to finish.") + } + Error::UnsupportedFeature(_) => { + Some("Upgrade RustFS to a release that ships on-demand migration.") + } + Error::InvalidPath(_) | Error::Config(_) => Some("Review the command arguments and retry."), + _ => None, + } +} + +fn emit_error(error: &Error, operation: Operation, formatter: &Formatter) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + if formatter.is_json() { + formatter.json_error(&error_output(error, operation)); + } else if let Some(suggestion) = suggestion(error, operation) { + formatter.error_with_suggestion(code, &error.to_string(), suggestion); + } else { + formatter.error_with_code(code, &error.to_string()); + } + code +} + +fn yes_no(value: bool) -> &'static str { + if value { "yes" } else { "no" } +} + +fn or_dash(value: Option<&str>) -> &str { + value.filter(|value| !value.is_empty()).unwrap_or(NO_VALUE) +} + +fn bytes_text(bytes: u64) -> String { + format!( + "{} ({bytes})", + humansize::format_size(bytes, humansize::BINARY) + ) +} + +fn row(formatter: &Formatter, label: &str, value: &str) { + formatter.println(&format!( + "{:<22}{}", + format!("{label}:"), + formatter.sanitize_text(value) + )); +} + +fn render_set( + bucket: &str, + result: &OnDemandMigrationSetResult, + dry_run: bool, + formatter: &Formatter, +) { + if formatter.is_json() { + return; + } + let bucket_text = formatter.sanitize_text(bucket); + if dry_run || result.dry_run { + formatter.success(&format!( + "Dry run: on-demand migration configuration for '{bucket_text}' is valid; nothing was saved" + )); + } else { + formatter.success(&format!( + "On-demand migration configured for '{bucket_text}'" + )); + } + if let Some(probe) = &result.probe { + row(formatter, "Source reachable", yes_no(probe.reachable)); + row(formatter, "Source listable", yes_no(probe.listable)); + row( + formatter, + "Sample key", + or_dash(probe.sample_key.as_deref()), + ); + } + if let Some(config) = &result.config { + formatter.println(""); + render_config(bucket, config, result.updated_at.as_deref(), formatter); + } +} + +fn render_config( + bucket: &str, + config: &OnDemandMigrationConfigView, + updated_at: Option<&str>, + formatter: &Formatter, +) { + let source = &config.source; + let policy = &config.policy; + formatter.println(&formatter.style_name(&format!( + "On-Demand Migration: {}", + formatter.sanitize_text(bucket) + ))); + row(formatter, "Enabled", yes_no(config.enabled)); + row(formatter, "Provider", or_dash(Some(&source.provider))); + row(formatter, "Endpoint", or_dash(source.endpoint.as_deref())); + row(formatter, "Region", or_dash(Some(&source.region))); + row(formatter, "Source bucket", or_dash(Some(&source.bucket))); + row(formatter, "Path style", source.path_style.as_str()); + match &source.credentials { + Some(credentials) => { + row( + formatter, + "Access key", + or_dash(Some(&credentials.access_key)), + ); + row( + formatter, + "Secret key", + if credentials.has_secret_key { + REDACTED_SECRET + } else { + NO_VALUE + }, + ); + if credentials.has_session_token { + row(formatter, "Session token", REDACTED_SECRET); + } + } + None => row(formatter, "Credentials", "anonymous"), + } + row(formatter, "TLS verify", yes_no(!source.tls.skip_verify)); + row( + formatter, + "CA certificate", + if source.tls.has_ca_cert { + "custom bundle" + } else { + "system roots" + }, + ); + row( + formatter, + "Prefix", + or_dash(config.filter.prefix.as_deref()), + ); + row( + formatter, + "Source prefix", + or_dash(config.filter.source_prefix.as_deref()), + ); + row(formatter, "HEAD policy", policy.head.as_str()); + row(formatter, "Range GET policy", policy.range_get.as_str()); + row( + formatter, + "Source error policy", + policy.source_error.as_str(), + ); + row(formatter, "List through", yes_no(policy.list_through)); + row(formatter, "Preserve ETag", yes_no(policy.preserve_etag)); + row(formatter, "Copy tags", yes_no(policy.copy_tags)); + row(formatter, "Emit events", yes_no(policy.emit_events)); + row( + formatter, + "Negative cache TTL", + &format!("{} s", policy.negative_cache_ttl_secs), + ); + row( + formatter, + "Inline max bytes", + &bytes_text(policy.inline_max_bytes), + ); + row( + formatter, + "Multipart part size", + &bytes_text(policy.multipart_part_size_bytes), + ); + row( + formatter, + "Max concurrent pulls", + &policy.max_concurrent_pulls.to_string(), + ); + row( + formatter, + "Pull queue capacity", + &policy.pull_queue_capacity.to_string(), + ); + row( + formatter, + "Source timeout", + &format!( + "connect {} ms, first byte {} ms, idle {} ms", + policy.source_timeout.connect_ms, + policy.source_timeout.first_byte_ms, + policy.source_timeout.idle_ms + ), + ); + row( + formatter, + "Bandwidth limit", + &policy + .bandwidth_limit_bytes_per_sec + .map(|limit| format!("{}/s", humansize::format_size(limit, humansize::BINARY))) + .unwrap_or_else(|| "unlimited".to_string()), + ); + row(formatter, "Updated at", or_dash(updated_at)); +} + +fn counter_line(counters: &std::collections::BTreeMap, nonzero_only: bool) -> String { + let parts = counters + .iter() + .filter(|(_, count)| !nonzero_only || **count > 0) + .map(|(label, count)| format!("{label} {count}")) + .collect::>(); + if parts.is_empty() { + "none".to_string() + } else { + parts.join(", ") + } +} + +fn ratio_text(ratio: Option) -> String { + match ratio { + Some(ratio) if ratio.is_finite() => format!("{:.1}%", ratio * 100.0), + _ => NO_VALUE.to_string(), + } +} + +fn render_status(bucket: &str, status: &OnDemandMigrationStatus, formatter: &Formatter) { + if formatter.is_json() { + return; + } + formatter.println(&formatter.style_name(&format!( + "On-Demand Migration Status: {}", + formatter.sanitize_text(bucket) + ))); + row(formatter, "Configured", yes_no(status.configured)); + row(formatter, "Enabled", yes_no(status.enabled)); + row(formatter, "Module enabled", yes_no(status.module_enabled)); + row(formatter, "Provider", or_dash(status.provider.as_deref())); + row( + formatter, + "Source host", + or_dash(status.endpoint_host.as_deref()), + ); + match &status.breaker { + Some(breaker) => { + let mut text = or_dash(Some(&breaker.state)).to_string(); + if let Some(opened_at) = breaker.opened_at.as_deref() { + text.push_str(&format!(" (opened {opened_at})")); + } + row(formatter, "Breaker", &text); + } + None => row(formatter, "Breaker", NO_VALUE), + } + row( + formatter, + "Source-hit ratio", + &ratio_text(status.served_by_source_ratio), + ); + match &status.counters { + Some(counters) => { + row( + formatter, + "Migrated bytes", + &bytes_text(counters.pulled_bytes_total), + ); + row( + formatter, + "Pulled objects", + &counter_line(&counters.pulled_objects_total, false), + ); + for (operation, outcomes) in &counters.requests_total { + row( + formatter, + &format!("Requests ({operation})"), + &counter_line(outcomes, true), + ); + } + row( + formatter, + "Pull failures", + &counter_line(&counters.pull_failures_total, true), + ); + if let Some(latency) = &counters.source_latency { + let text = if latency.count == 0 { + "no samples".to_string() + } else { + format!( + "{} samples, mean {} ms", + latency.count, + latency.sum_ms / latency.count + ) + }; + row(formatter, "Source latency", &text); + } + } + None => row(formatter, "Counters", NO_VALUE), + } + row( + formatter, + "In-flight pulls", + &status.inflight_pulls.to_string(), + ); + row(formatter, "Queued pulls", &status.queue_depth.to_string()); + match &status.last_source_error { + Some(error) => { + let mut text = or_dash(Some(&error.class)).to_string(); + if let Some(at) = error.at.as_deref() { + text.push_str(&format!(" at {at}")); + } + row(formatter, "Last source error", &text); + } + None => row(formatter, "Last source error", "none"), + } + if let Some(backfill) = &status.backfill { + row( + formatter, + "Backfill", + &format!( + "{} (job {}): listed {}, enqueued {}, pulled {}, skipped {}, failed {}, {}", + or_dash(Some(&backfill.state)), + or_dash(Some(&backfill.job_id)), + backfill.listed, + backfill.enqueued, + backfill.pulled, + backfill.skipped_existing, + backfill.failed, + humansize::format_size(backfill.bytes, humansize::BINARY) + ), + ); + } + row( + formatter, + "Updated at", + or_dash(status.updated_at.as_deref()), + ); +} + +fn backfill_progress_line(job: &BackfillJob) -> String { + format!( + "[{}] listed {} \u{b7} enqueued {} \u{b7} pulled {} \u{b7} skipped {} \u{b7} failed {} \u{b7} {} \u{b7} updated {}", + or_dash(Some(&job.state)), + job.listed, + job.enqueued, + job.pulled, + job.skipped_existing, + job.failed, + humansize::format_size(job.bytes, humansize::BINARY), + or_dash(job.updated_at.as_deref()) + ) +} + +fn render_backfill(bucket: &str, job: Option<&BackfillJob>, formatter: &Formatter) { + if formatter.is_json() { + return; + } + formatter + .println(&formatter.style_name(&format!("Backfill: {}", formatter.sanitize_text(bucket)))); + let Some(job) = job else { + row(formatter, "Job", "none recorded"); + return; + }; + row(formatter, "Job", or_dash(Some(&job.job_id))); + row(formatter, "State", or_dash(Some(&job.state))); + row(formatter, "Dry run", yes_no(job.dry_run)); + row(formatter, "Prefix", or_dash(job.prefix.as_deref())); + row(formatter, "Skip existing", job.skip_existing.as_str()); + row(formatter, "Listed", &job.listed.to_string()); + row(formatter, "Enqueued", &job.enqueued.to_string()); + row(formatter, "Pulled", &job.pulled.to_string()); + row( + formatter, + "Skipped existing", + &job.skipped_existing.to_string(), + ); + row(formatter, "Failed", &job.failed.to_string()); + row(formatter, "Bytes", &bytes_text(job.bytes)); + row(formatter, "Last key", or_dash(job.last_key.as_deref())); + match &job.last_error { + Some(error) => { + let mut text = or_dash(Some(&error.class)).to_string(); + if let Some(hash) = error.key_hash.as_deref() { + text.push_str(&format!(" (key hash {hash})")); + } + if let Some(at) = error.at.as_deref() { + text.push_str(&format!(" at {at}")); + } + row(formatter, "Last error", &text); + } + None => row(formatter, "Last error", "none"), + } + if !job.failed_keys.is_empty() { + row( + formatter, + "Failed key hashes", + &job.failed_keys.len().to_string(), + ); + } + match &job.owner { + Some(owner) => { + let mut text = or_dash(Some(&owner.node)).to_string(); + if let Some(lease) = owner.lease_until.as_deref() { + text.push_str(&format!(" (lease until {lease})")); + } + row(formatter, "Owner", &text); + } + None => row(formatter, "Owner", "none"), + } + row(formatter, "Started at", or_dash(job.started_at.as_deref())); + row(formatter, "Updated at", or_dash(job.updated_at.as_deref())); + row( + formatter, + "Config updated at", + or_dash(job.config_updated_at.as_deref()), + ); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::output::OutputConfig; + use async_trait::async_trait; + use clap::Parser; + use rc_core::admin::{BackfillJobResult, OnDemandMigrationConfigResult}; + use std::sync::Mutex; + + const GET_RESPONSE: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/get_response.json"); + const STATUS: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/status.json"); + const BACKFILL_JOB: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/backfill_job.json"); + + #[derive(Parser)] + struct TestCli { + #[command(subcommand)] + command: BucketCommands, + } + + fn formatter() -> Formatter { + Formatter::new(OutputConfig { + json: true, + no_color: true, + no_progress: true, + quiet: false, + }) + } + + fn human_formatter() -> Formatter { + Formatter::new(OutputConfig { + json: false, + no_color: true, + no_progress: true, + quiet: false, + }) + } + + fn set_args(extra: &[&str]) -> SetArgs { + let mut args = vec![ + "rc", + "migration", + "set", + "local/photos", + "--provider", + "minio", + "--endpoint", + "https://source.example.com:9000", + "--region", + "us-east-1", + "--source-bucket", + "legacy-photos", + ]; + args.extend_from_slice(extra); + match TestCli::parse_from(args).command { + BucketCommands::Migration(MigrationCommands::Set(args)) => args, + _ => panic!("expected set"), + } + } + + #[test] + fn parses_the_full_set_surface_with_snake_case_policy_values() { + let args = set_args(&[ + "--source-prefix", + "photos/", + "--prefix", + "img/", + "--access-key", + "AKIASOURCE", + "--secret-key", + "s3cr3t", + "--path-style", + "virtual", + "--skip-tls-verify", + "--head", + "local_only", + "--range-get", + "serve_only", + "--source-error", + "not_found", + "--no-preserve-etag", + "--copy-tags", + "--no-events", + "--inline-max-bytes", + "1024", + "--max-concurrent-pulls", + "4", + "--dry-run", + ]); + assert_eq!(args.provider, ProviderArg::Minio); + assert_eq!(args.head, HeadArg::LocalOnly); + assert_eq!(args.range_get, RangeGetArg::ServeOnly); + assert_eq!(args.source_error, SourceErrorArg::NotFound); + assert_eq!(args.path_style, PathStyleArg::Virtual); + assert!(args.dry_run && args.skip_tls_verify && args.copy_tags); + assert!(args.no_preserve_etag && args.no_events); + assert_eq!(args.inline_max_bytes, Some(1024)); + assert_eq!(args.max_concurrent_pulls, Some(4)); + // The secret never appears in the derived Debug output. + let debug = format!("{args:?}"); + assert!(!debug.contains("s3cr3t")); + assert!(debug.contains("AKIASOURCE")); + + let request = build_request(&args, Some(Zeroizing::new("s3cr3t".into()))).unwrap(); + assert_eq!(request.policy.head, HeadPolicy::LocalOnly); + assert_eq!(request.policy.range_get, RangeGetPolicy::ServeOnly); + assert_eq!(request.policy.source_error, SourceErrorPolicy::NotFound); + assert!(!request.policy.preserve_etag && request.policy.copy_tags); + assert!(!request.policy.emit_events); + assert_eq!(request.policy.inline_max_bytes, 1024); + assert_eq!(request.policy.max_concurrent_pulls, 4); + assert!(request.source.tls.skip_verify); + assert_eq!(request.filter.prefix.as_deref(), Some("img/")); + } + + #[test] + fn credential_flags_are_mutually_exclusive_with_public() { + for args in [ + vec!["--public", "--access-key", "AK"], + vec!["--public", "--secret-key", "SK"], + vec!["--secret-key", "SK"], + vec!["--head", "local-only"], + vec!["--skip-existing", "always"], + ] { + let mut full = vec![ + "rc", + "migration", + "set", + "local/photos", + "--provider", + "s3", + "--endpoint", + "https://s.example.com", + "--region", + "r", + "--source-bucket", + "b", + ]; + full.extend(args.iter()); + assert!(TestCli::try_parse_from(&full).is_err(), "{args:?}"); + } + } + + #[test] + fn secret_comes_from_flag_then_environment_and_refuses_the_placeholder() { + let formatter = formatter(); + let env = |value: &str| Some(std::ffi::OsString::from(value)); + + let args = set_args(&["--access-key", "AK", "--secret-key", "from-flag"]); + assert_eq!( + resolve_secret(&args, env("ignored"), &formatter) + .unwrap() + .unwrap() + .as_str(), + "from-flag" + ); + + let args = set_args(&["--access-key", "AK"]); + assert_eq!( + resolve_secret(&args, env("from-env\n"), &formatter) + .unwrap() + .unwrap() + .as_str(), + "from-env" + ); + let error = resolve_secret(&args, env(REDACTED_SECRET), &formatter).unwrap_err(); + assert_eq!(error.exit_code(), 2); + assert!(error.to_string().contains("placeholder")); + assert_eq!( + resolve_secret(&args, env(""), &formatter) + .unwrap_err() + .exit_code(), + 2 + ); + // JSON output never prompts. + let error = resolve_secret(&args, None, &formatter).unwrap_err(); + assert!(error.to_string().contains(SECRET_KEY_ENV)); + + let args = set_args(&["--public"]); + assert!( + resolve_secret(&args, env("ignored"), &formatter) + .unwrap() + .is_none() + ); + let args = set_args(&[]); + assert_eq!( + resolve_secret(&args, None, &formatter) + .unwrap_err() + .exit_code(), + 2 + ); + } + + #[test] + fn targets_are_alias_slash_bucket_only() { + assert_eq!( + parse_target("local/photos").unwrap(), + ("local".to_string(), "photos".to_string()) + ); + for target in [ + "local", + "/photos", + "local/", + "local/photos/key", + "local/a b", + ] { + assert_eq!(parse_target(target).unwrap_err().exit_code(), 2, "{target}"); + } + } + + #[test] + fn ca_cert_must_be_a_bounded_pem_file() { + let dir = tempfile::tempdir().unwrap(); + let good = dir.path().join("ca.pem"); + std::fs::write( + &good, + "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n", + ) + .unwrap(); + assert!(read_ca_cert(&good).is_ok()); + let bad = dir.path().join("bad.pem"); + std::fs::write(&bad, "not a certificate").unwrap(); + assert_eq!(read_ca_cert(&bad).unwrap_err().exit_code(), 2); + let huge = dir.path().join("huge.pem"); + std::fs::write(&huge, "-".repeat(MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES + 1)).unwrap(); + assert_eq!(read_ca_cert(&huge).unwrap_err().exit_code(), 2); + assert_eq!( + read_ca_cert(&dir.path().join("missing.pem")) + .unwrap_err() + .exit_code(), + 2 + ); + } + + #[test] + fn set_validates_arguments_before_asking_for_the_secret() { + // Provider s3 without an endpoint fails before the secret is resolved, + // so no prompt and no environment lookup happen. + let command = MigrationCommands::Set(set_args_without_endpoint()); + let error = prepare(command, &formatter()).unwrap_err(); + assert_eq!(error.exit_code(), 2); + assert!(error.to_string().contains("--endpoint")); + } + + fn set_args_without_endpoint() -> SetArgs { + match TestCli::parse_from([ + "rc", + "migration", + "set", + "local/photos", + "--provider", + "s3", + "--region", + "us-east-1", + "--source-bucket", + "b", + "--access-key", + "AK", + ]) + .command + { + BucketCommands::Migration(MigrationCommands::Set(args)) => args, + _ => panic!("expected set"), + } + } + + #[test] + fn human_rendering_uses_an_em_dash_for_a_null_ratio_and_never_prints_secrets() { + let status: OnDemandMigrationStatus = serde_json::from_str(STATUS).unwrap(); + assert_eq!(ratio_text(status.served_by_source_ratio), NO_VALUE); + assert_eq!(ratio_text(Some(0.0)), "0.0%"); + assert_eq!(ratio_text(Some(0.4567)), "45.7%"); + assert_eq!(ratio_text(Some(f64::NAN)), NO_VALUE); + let config: OnDemandMigrationConfigResult = serde_json::from_str(GET_RESPONSE).unwrap(); + let text = serde_json::to_string(&config).unwrap(); + assert!(text.contains(REDACTED_SECRET)); + let job: BackfillJobResult = serde_json::from_str(BACKFILL_JOB).unwrap(); + let line = backfill_progress_line(job.job.as_ref().unwrap()); + assert!(line.starts_with("[running] listed 2000")); + assert!(line.contains("pulled 1400")); + assert!(line.contains("70 MiB")); + } + + // ---- exit code tests against a fake API ---- + + #[derive(Default)] + struct FakeApi { + status: Option, + get: Option, + backfill: Mutex>, + error: Option Error>, + calls: Mutex>, + } + + impl FakeApi { + fn fail(&self) -> Result<()> { + match self.error { + Some(error) => Err(error()), + None => Ok(()), + } + } + fn record(&self, call: &'static str) { + self.calls.lock().unwrap().push(call); + } + } + + #[async_trait] + impl OnDemandMigrationApi for FakeApi { + async fn set_on_demand_migration( + &self, + bucket: &str, + config: &OnDemandMigrationConfigRequest, + dry_run: bool, + ) -> Result { + self.record("set"); + self.fail()?; + Ok(OnDemandMigrationSetResult { + bucket: bucket.to_string(), + dry_run, + config: Some(OnDemandMigrationConfigView { + source: rc_core::admin::SourceView { + provider: config.source.provider.as_str().to_string(), + ..Default::default() + }, + ..Default::default() + }), + updated_at: (!dry_run).then(|| "2026-09-02T10:00:00Z".to_string()), + probe: None, + }) + } + async fn get_on_demand_migration( + &self, + _bucket: &str, + ) -> Result { + self.record("get"); + self.fail()?; + Ok(self.get.clone().unwrap_or_default()) + } + async fn delete_on_demand_migration(&self, _bucket: &str) -> Result<()> { + self.record("delete"); + self.fail() + } + async fn on_demand_migration_status( + &self, + _bucket: &str, + ) -> Result { + self.record("status"); + self.fail()?; + Ok(self.status.clone().unwrap_or_default()) + } + async fn start_on_demand_migration_backfill( + &self, + _bucket: &str, + _request: &BackfillStartRequest, + ) -> Result { + self.record("backfill_start"); + self.fail()?; + Ok(self.backfill.lock().unwrap().remove(0)) + } + async fn cancel_on_demand_migration_backfill( + &self, + _bucket: &str, + ) -> Result { + self.record("backfill_cancel"); + self.fail()?; + Ok(self.backfill.lock().unwrap().remove(0)) + } + async fn on_demand_migration_backfill_status( + &self, + _bucket: &str, + ) -> Result { + self.record("backfill_status"); + self.fail()?; + Ok(self.backfill.lock().unwrap().remove(0)) + } + } + + fn prepared(action: Action) -> Prepared { + Prepared { + alias: "local".into(), + bucket: "photos".into(), + action, + } + } + + #[tokio::test] + async fn status_succeeds_and_json_carries_the_null_ratio() { + let api = FakeApi { + status: Some(serde_json::from_str(STATUS).unwrap()), + ..Default::default() + }; + let code = + execute_with_api(prepared(Action::Status { watch: None }), &api, &formatter()).await; + assert_eq!(code, ExitCode::Success); + let human = human_formatter(); + let code = execute_with_api(prepared(Action::Status { watch: None }), &api, &human).await; + assert_eq!(code, ExitCode::Success); + } + + #[tokio::test] + async fn unsupported_server_maps_to_exit_7() { + let api = FakeApi { + error: Some(|| { + Error::UnsupportedFeature("server does not support on-demand migration".into()) + }), + ..Default::default() + }; + assert_eq!( + execute_with_api(prepared(Action::Get), &api, &formatter()).await, + ExitCode::UnsupportedFeature + ); + } + + #[tokio::test] + async fn not_found_conflict_network_and_auth_keep_their_exit_codes() { + for (error, expected) in [ + ( + (|| Error::NotFound("NoSuchConfiguration".into())) as fn() -> Error, + ExitCode::NotFound, + ), + ( + || Error::Conflict("backfill running".into()), + ExitCode::Conflict, + ), + ( + || Error::Network("source unreachable".into()), + ExitCode::NetworkError, + ), + (|| Error::Auth("licence".into()), ExitCode::AuthError), + (|| Error::Config("bad".into()), ExitCode::UsageError), + ] { + let api = FakeApi { + error: Some(error), + ..Default::default() + }; + assert_eq!( + execute_with_api(prepared(Action::BackfillCancel), &api, &formatter()).await, + expected + ); + } + } + + #[tokio::test] + async fn remove_reports_success_without_a_body() { + let api = FakeApi::default(); + assert_eq!( + execute_with_api(prepared(Action::Remove), &api, &formatter()).await, + ExitCode::Success + ); + assert_eq!(*api.calls.lock().unwrap(), vec!["delete"]); + } + + #[tokio::test] + async fn backfill_watch_stops_at_a_terminal_state() { + let running: BackfillJobResult = serde_json::from_str(BACKFILL_JOB).unwrap(); + let mut done = running.clone(); + done.job.as_mut().unwrap().state = "completed".into(); + let api = FakeApi { + backfill: Mutex::new(vec![running, done]), + ..Default::default() + }; + let code = execute_with_api( + prepared(Action::BackfillStatus { + watch: Some(Duration::from_millis(10)), + }), + &api, + &formatter(), + ) + .await; + assert_eq!(code, ExitCode::Success); + assert_eq!( + *api.calls.lock().unwrap(), + vec!["backfill_status", "backfill_status"] + ); + } + + #[tokio::test] + async fn backfill_watch_treats_a_missing_job_as_final() { + let api = FakeApi { + backfill: Mutex::new(vec![BackfillJobResult::default()]), + ..Default::default() + }; + let code = execute_with_api( + prepared(Action::BackfillStatus { + watch: Some(Duration::from_millis(10)), + }), + &api, + &human_formatter(), + ) + .await; + assert_eq!(code, ExitCode::Success); + assert_eq!(api.calls.lock().unwrap().len(), 1); + } + + #[test] + fn json_error_envelope_marks_reads_retryable_and_unsupported_capability() { + let value = error_output(&Error::Network("down".into()), Operation::Get); + assert_eq!(value["type"], OUTPUT_TYPE); + assert_eq!(value["error"]["type"], "network_error"); + assert_eq!(value["error"]["retryable"], true); + let value = error_output(&Error::Network("probe".into()), Operation::Set); + assert_eq!(value["error"]["retryable"], false); + assert!( + value["error"]["suggestion"] + .as_str() + .unwrap() + .contains("--dry-run") + ); + let value = error_output(&Error::UnsupportedFeature("nope".into()), Operation::Status); + assert_eq!(value["error"]["capability"], ON_DEMAND_MIGRATION_CAPABILITY); + assert_eq!( + success_output(Operation::BackfillStart, "photos", json!({"a": 1}))["data"]["operation"], + "backfill_start" + ); + } +} diff --git a/crates/cli/src/commands/admin/mod.rs b/crates/cli/src/commands/admin/mod.rs index 84a19f1e..cd4e6f8a 100644 --- a/crates/cli/src/commands/admin/mod.rs +++ b/crates/cli/src/commands/admin/mod.rs @@ -5,6 +5,7 @@ mod access_key; mod account; +mod bucket; mod bucket_metadata; mod capabilities; mod config; @@ -47,6 +48,10 @@ pub enum AdminCommands { #[command(subcommand)] Account(account::AccountCommands), + /// Manage per-bucket features such as on-demand migration from an external source + #[command(subcommand)] + Bucket(bucket::BucketCommands), + /// Discover effective RustFS runtime capabilities Capabilities(capabilities::CapabilitiesArgs), @@ -145,6 +150,7 @@ pub async fn execute(cmd: AdminCommands, output_config: OutputConfig) -> ExitCod match cmd { AdminCommands::Table(cmd) => super::table::execute_admin(cmd, &formatter).await, AdminCommands::Account(account_cmd) => account::execute(account_cmd, &formatter).await, + AdminCommands::Bucket(command) => bucket::execute(command, &formatter).await, AdminCommands::Capabilities(args) => capabilities::execute(args, &formatter).await, AdminCommands::Diagnostics(command) => diagnostics::execute(command, &formatter).await, AdminCommands::Config(config_cmd) => config::execute(config_cmd, &formatter).await, diff --git a/crates/cli/tests/admin_bucket_migration.rs b/crates/cli/tests/admin_bucket_migration.rs new file mode 100644 index 00000000..22a88383 --- /dev/null +++ b/crates/cli/tests/admin_bucket_migration.rs @@ -0,0 +1,430 @@ +//! Binary-level contract tests for `rc admin bucket migration`. +//! +//! Every server answer comes from the wire fixtures vendored under +//! `crates/core/tests/fixtures/on_demand_migration/`. + +#![cfg(not(windows))] + +mod admin_support; + +use admin_support::{rc_binary, rc_host_alias, start_admin_sequence_test_server}; +use serde_json::Value; +use std::process::{Command, Output}; +use std::time::Duration; + +const SET_REQUEST: &str = + include_str!("../../core/tests/fixtures/on_demand_migration/set_request.json"); +const SET_RESPONSE: &str = + include_str!("../../core/tests/fixtures/on_demand_migration/set_response.json"); +const GET_RESPONSE: &str = + include_str!("../../core/tests/fixtures/on_demand_migration/get_response.json"); +const STATUS: &str = include_str!("../../core/tests/fixtures/on_demand_migration/status.json"); +const STATUS_WITH_BACKFILL: &str = + include_str!("../../core/tests/fixtures/on_demand_migration/status_with_backfill.json"); +const BACKFILL_JOB: &str = + include_str!("../../core/tests/fixtures/on_demand_migration/backfill_job.json"); +const BACKFILL_JOB_COMPLETED: &str = r#"{"bucket":"photos","job":{"format_version":1,"job_id":"11111111-1111-4111-8111-111111111111","state":"completed","config_updated_at":"2026-09-02T10:00:00Z","prefix":"photos/","skip_existing":"always","dry_run":false,"listed":2000,"enqueued":2000,"pulled":1500,"skipped_existing":500,"failed":0,"bytes":73400320,"started_at":"2026-09-02T10:00:30Z","updated_at":"2026-09-02T10:09:10Z"}}"#; + +fn run(endpoint: &str, args: &[&str], env: &[(&str, &str)]) -> Output { + let config = tempfile::tempdir().unwrap(); + let mut command = Command::new(rc_binary()); + command + .args(["admin", "bucket", "migration"]) + .args(args) + .env("RC_CONFIG_DIR", config.path()) + .env("RC_HOST_myalias", rc_host_alias(endpoint)) + .env_remove("RC_ODM_SECRET_KEY") + .env("NO_PROXY", "*") + .env("no_proxy", "*"); + for (key, value) in env { + command.env(key, value); + } + command.output().unwrap() +} + +fn stdout(output: &Output) -> String { + String::from_utf8_lossy(&output.stdout).into_owned() +} + +fn stderr(output: &Output) -> String { + String::from_utf8_lossy(&output.stderr).into_owned() +} + +#[test] +fn get_prints_a_redacted_table_and_json_envelope() { + let (endpoint, requests, server) = + start_admin_sequence_test_server(vec![("200 OK", GET_RESPONSE), ("200 OK", GET_RESPONSE)]); + let human = run(&endpoint, &["get", "myalias/photos"], &[]); + assert!(human.status.success(), "{}", stderr(&human)); + let text = stdout(&human); + assert!(text.contains("On-Demand Migration: photos")); + assert!(text.contains("legacy-photos")); + assert!(text.contains("AKIASOURCE")); + assert!(text.contains("REDACTED")); + assert!(text.contains("Source prefix: photos/")); + assert!(text.contains("Max concurrent pulls: 8")); + + let json = run(&endpoint, &["get", "myalias/photos", "--json"], &[]); + assert!(json.status.success(), "{}", stderr(&json)); + let data: Value = serde_json::from_slice(&json.stdout).unwrap(); + assert_eq!(data["schema_version"], 3); + assert_eq!(data["type"], "on_demand_migration"); + assert_eq!(data["data"]["operation"], "get"); + assert_eq!(data["data"]["bucket"], "photos"); + assert_eq!( + data["data"]["result"]["config"]["source"]["credentials"]["secret_key"], + "REDACTED" + ); + assert_eq!(data["data"]["result"]["updated_at"], "2026-09-02T10:00:00Z"); + + let first = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(first.method, "GET"); + assert_eq!(first.target, "/rustfs/admin/v3/on-demand-migration/photos"); + assert!( + first + .headers + .to_ascii_lowercase() + .contains("authorization: aws4-hmac-sha256") + ); + server.join().unwrap(); +} + +#[test] +fn status_renders_the_null_ratio_as_an_em_dash_and_keeps_it_null_in_json() { + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![ + ("200 OK", STATUS), + ("200 OK", STATUS_WITH_BACKFILL), + ]); + let human = run(&endpoint, &["status", "myalias/photos"], &[]); + assert!(human.status.success(), "{}", stderr(&human)); + let text = stdout(&human); + assert!(text.contains("Source-hit ratio: \u{2014}"), "{text}"); + assert!(!text.contains("Source-hit ratio: 0"), "{text}"); + assert!(text.contains("Migrated bytes: 4 KiB (4096)")); + assert!(text.contains("In-flight pulls: 1")); + assert!(text.contains("Queued pulls: 1")); + assert!(text.contains("Breaker: half_open")); + assert!(text.contains("Last source error: server_error at 2026-09-02T10:00:00Z")); + assert!(text.contains("Requests (get): source_hit 2")); + + let json = run(&endpoint, &["status", "myalias/photos", "--json"], &[]); + assert!(json.status.success(), "{}", stderr(&json)); + let data: Value = serde_json::from_slice(&json.stdout).unwrap(); + assert_eq!(data["data"]["operation"], "status"); + assert!(data["data"]["result"]["served_by_source_ratio"].is_null()); + assert_eq!(data["data"]["result"]["backfill"]["state"], "running"); + + let first = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + first.target, + "/rustfs/admin/v3/on-demand-migration/photos/status" + ); + server.join().unwrap(); +} + +#[test] +fn set_reads_the_secret_from_the_environment_and_sends_the_fixture_body() { + let (endpoint, requests, server) = + start_admin_sequence_test_server(vec![("200 OK", SET_RESPONSE)]); + let output = run( + &endpoint, + &[ + "set", + "myalias/photos", + "--provider", + "minio", + "--endpoint", + "https://source.example.com:9000", + "--region", + "us-east-1", + "--source-bucket", + "legacy-photos", + "--source-prefix", + "photos/", + "--access-key", + "AKIASOURCE", + "--dry-run", + "--json", + ], + &[("RC_ODM_SECRET_KEY", "sourceSecretKey123")], + ); + assert!(output.status.success(), "{}", stderr(&output)); + assert!(!stdout(&output).contains("sourceSecretKey123")); + assert!(!stderr(&output).contains("sourceSecretKey123")); + let data: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(data["data"]["operation"], "set"); + assert_eq!(data["data"]["result"]["probe"]["reachable"], true); + assert_eq!( + data["data"]["result"]["config"]["source"]["credentials"]["secret_key"], + "REDACTED" + ); + + let request = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(request.method, "PUT"); + assert_eq!( + request.target, + "/rustfs/admin/v3/on-demand-migration/photos?dry-run=true" + ); + let body: Value = serde_json::from_slice(&request.body).unwrap(); + let expected: Value = serde_json::from_str(SET_REQUEST).unwrap(); + assert_eq!(body, expected); + server.join().unwrap(); +} + +#[test] +fn set_without_a_secret_source_is_a_usage_error_before_any_request() { + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![]); + let output = run( + &endpoint, + &[ + "set", + "myalias/photos", + "--provider", + "minio", + "--endpoint", + "https://source.example.com:9000", + "--region", + "us-east-1", + "--source-bucket", + "legacy-photos", + "--access-key", + "AKIASOURCE", + "--json", + ], + &[], + ); + assert_eq!(output.status.code(), Some(2)); + let error: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!(error["type"], "on_demand_migration"); + assert_eq!(error["error"]["type"], "usage_error"); + assert!( + error["error"]["message"] + .as_str() + .unwrap() + .contains("RC_ODM_SECRET_KEY") + ); + server.join().unwrap(); + assert!(requests.try_recv().is_err()); +} + +#[test] +fn a_server_without_the_route_family_is_reported_as_unsupported() { + let (endpoint, _requests, server) = + start_admin_sequence_test_server(vec![("404 Not Found", ""), ("404 Not Found", "")]); + let human = run(&endpoint, &["get", "myalias/photos"], &[]); + assert_eq!(human.status.code(), Some(7)); + assert!(stderr(&human).contains("server does not support on-demand migration")); + assert!(human.stdout.is_empty()); + + let json = run(&endpoint, &["status", "myalias/photos", "--json"], &[]); + assert_eq!(json.status.code(), Some(7)); + let error: Value = serde_json::from_slice(&json.stderr).unwrap(); + assert_eq!(error["error"]["type"], "unsupported_feature"); + assert_eq!(error["error"]["capability"], "admin.on-demand-migration"); + server.join().unwrap(); +} + +#[test] +fn exit_codes_follow_the_server_answer() { + let (endpoint, _requests, server) = start_admin_sequence_test_server(vec![ + ( + "404 Not Found", + r#"{"Code":"NoSuchConfiguration","Message":"on-demand migration is not configured for bucket photos"}"#, + ), + ( + "409 Conflict", + r#"{"Code":"OnDemandMigrationBackfillRunning","Message":"a backfill job is already running for bucket photos"}"#, + ), + ( + "400 Bad Request", + r#"{"Code":"OnDemandMigrationSourceUnreachable","Message":"connect"}"#, + ), + ( + "403 Forbidden", + r#"{"Code":"AccessDenied","Message":"licence does not include on-demand migration"}"#, + ), + ( + "400 Bad Request", + r#"{"Code":"InvalidArgument","Message":"source_timeout.connect_ms out of range"}"#, + ), + ]); + let not_found = run(&endpoint, &["get", "myalias/photos", "--json"], &[]); + assert_eq!(not_found.status.code(), Some(5)); + + let conflict = run( + &endpoint, + &["backfill", "start", "myalias/photos", "--json"], + &[], + ); + assert_eq!(conflict.status.code(), Some(6)); + let error: Value = serde_json::from_slice(&conflict.stderr).unwrap(); + assert_eq!(error["error"]["type"], "conflict"); + assert_eq!(error["error"]["retryable"], false); + + let set_args = [ + "set", + "myalias/photos", + "--provider", + "s3", + "--endpoint", + "https://source.example.com", + "--region", + "us-east-1", + "--source-bucket", + "legacy", + "--access-key", + "AK", + "--json", + ]; + let secret = [("RC_ODM_SECRET_KEY", "sk")]; + let unreachable = run(&endpoint, &set_args, &secret); + assert_eq!(unreachable.status.code(), Some(3)); + let error: Value = serde_json::from_slice(&unreachable.stderr).unwrap(); + assert_eq!(error["error"]["type"], "network_error"); + assert!( + error["error"]["suggestion"] + .as_str() + .unwrap() + .contains("--dry-run") + ); + + let licence = run(&endpoint, &set_args, &secret); + assert_eq!(licence.status.code(), Some(4)); + + let invalid = run(&endpoint, &set_args, &secret); + assert_eq!(invalid.status.code(), Some(2)); + server.join().unwrap(); +} + +#[test] +fn remove_and_backfill_control_use_their_routes() { + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![ + ("204 No Content", ""), + ("200 OK", BACKFILL_JOB), + ("200 OK", BACKFILL_JOB), + ]); + let removed = run(&endpoint, &["rm", "myalias/photos", "--json"], &[]); + assert!(removed.status.success(), "{}", stderr(&removed)); + let data: Value = serde_json::from_slice(&removed.stdout).unwrap(); + assert_eq!(data["data"]["operation"], "remove"); + assert_eq!(data["data"]["result"]["removed"], true); + + let started = run( + &endpoint, + &[ + "backfill", + "start", + "myalias/photos", + "--prefix", + "photos/", + "--skip-existing", + "etag_or_size", + "--dry-run", + ], + &[], + ); + assert!(started.status.success(), "{}", stderr(&started)); + let text = stdout(&started); + assert!(text.contains("Backfill dry run started for 'photos'")); + assert!(text.contains("State: running")); + assert!(text.contains("Owner: node-a:9000 (lease until 2026-09-02T10:06:10Z)")); + + let cancelled = run( + &endpoint, + &["backfill", "cancel", "myalias/photos", "--json"], + &[], + ); + assert!(cancelled.status.success(), "{}", stderr(&cancelled)); + + let delete = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(delete.method, "DELETE"); + assert_eq!(delete.target, "/rustfs/admin/v3/on-demand-migration/photos"); + let start = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(start.method, "POST"); + assert_eq!( + start.target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill?op=start" + ); + let body: Value = serde_json::from_slice(&start.body).unwrap(); + assert_eq!( + body, + serde_json::json!({"prefix": "photos/", "skip_existing": "etag_or_size", "dry_run": true}) + ); + let cancel = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + cancel.target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill?op=cancel" + ); + assert!(cancel.body.is_empty()); + server.join().unwrap(); +} + +#[test] +fn backfill_status_watch_streams_until_the_job_finishes() { + let (endpoint, requests, server) = start_admin_sequence_test_server(vec![ + ("200 OK", BACKFILL_JOB), + ("200 OK", BACKFILL_JOB_COMPLETED), + ]); + let output = run( + &endpoint, + &[ + "backfill", + "status", + "myalias/photos", + "--watch", + "--interval", + "1", + "--json", + ], + &[], + ); + assert!(output.status.success(), "{}", stderr(&output)); + let lines = stdout(&output) + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0]["data"]["operation"], "backfill_status"); + assert_eq!(lines[0]["data"]["result"]["job"]["state"], "running"); + assert_eq!(lines[1]["data"]["result"]["job"]["state"], "completed"); + for _ in 0..2 { + let request = requests.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!( + request.target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill" + ); + } + server.join().unwrap(); +} + +#[test] +fn malformed_targets_and_missing_aliases_never_reach_the_server() { + let config = tempfile::tempdir().unwrap(); + let output = Command::new(rc_binary()) + .args(["admin", "bucket", "migration", "get", "myalias", "--json"]) + .env("RC_CONFIG_DIR", config.path()) + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + let error: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!(error["error"]["type"], "usage_error"); + + let output = Command::new(rc_binary()) + .args([ + "admin", + "bucket", + "migration", + "status", + "missing-odm-alias/photos", + "--json", + ]) + .env("RC_CONFIG_DIR", config.path()) + .env_remove("RC_HOST_missing-odm-alias") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(5)); + let error: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!(error["type"], "on_demand_migration"); + assert_eq!(error["error"]["type"], "not_found"); +} diff --git a/crates/cli/tests/fixtures/output_v3/on_demand_migration/empty.json b/crates/cli/tests/fixtures/output_v3/on_demand_migration/empty.json new file mode 100644 index 00000000..a7ff5469 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/on_demand_migration/empty.json @@ -0,0 +1,13 @@ +{ + "schema_version": 3, + "type": "on_demand_migration", + "status": "success", + "data": { + "operation": "backfill_status", + "bucket": "photos", + "result": { + "bucket": "photos", + "job": null + } + } +} diff --git a/crates/cli/tests/fixtures/output_v3/on_demand_migration/error.json b/crates/cli/tests/fixtures/output_v3/on_demand_migration/error.json new file mode 100644 index 00000000..4f14b7f3 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/on_demand_migration/error.json @@ -0,0 +1,13 @@ +{ + "schema_version": 3, + "type": "on_demand_migration", + "status": "error", + "error": { + "type": "unsupported_feature", + "message": "Unsupported feature: server does not support on-demand migration", + "retryable": false, + "capability": "admin.on-demand-migration", + "server": null, + "suggestion": "Upgrade RustFS to a release that ships on-demand migration." + } +} diff --git a/crates/cli/tests/fixtures/output_v3/on_demand_migration/success.json b/crates/cli/tests/fixtures/output_v3/on_demand_migration/success.json new file mode 100644 index 00000000..13c63fa4 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/on_demand_migration/success.json @@ -0,0 +1,58 @@ +{ + "schema_version": 3, + "type": "on_demand_migration", + "status": "success", + "data": { + "operation": "get", + "bucket": "photos", + "result": { + "bucket": "photos", + "config": { + "version": 1, + "enabled": true, + "source": { + "provider": "minio", + "endpoint": "https://source.example.com:9000", + "region": "us-east-1", + "bucket": "legacy-photos", + "path_style": "auto", + "credentials": { + "access_key": "AKIASOURCE", + "secret_key": "REDACTED", + "session_token": null + }, + "tls": { + "skip_verify": false, + "ca_cert_pem": null + } + }, + "filter": { + "prefix": null, + "source_prefix": "photos/" + }, + "policy": { + "head": "proxy", + "range_get": "serve_and_backfill", + "source_error": "propagate", + "list_through": false, + "respect_local_delete_marker": true, + "preserve_etag": true, + "copy_tags": false, + "emit_events": true, + "negative_cache_ttl_secs": 30, + "inline_max_bytes": 16777216, + "multipart_part_size_bytes": 67108864, + "max_concurrent_pulls": 8, + "pull_queue_capacity": 1024, + "source_timeout": { + "connect_ms": 5000, + "first_byte_ms": 15000, + "idle_ms": 30000 + }, + "bandwidth_limit_bytes_per_sec": null + } + }, + "updated_at": "2026-09-02T10:00:00Z" + } + } +} diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index c7208a35..9e51b2d2 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -170,6 +170,7 @@ fn top_level_command_help_contract() { args: &["admin"], usage: "Usage: rc admin [OPTIONS] ", expected_tokens: &[ + "bucket", "diagnostics", "info", "scanner", @@ -186,6 +187,51 @@ fn top_level_command_help_contract() { "access-key", ], }, + HelpCase { + args: &["admin", "bucket"], + usage: "Usage: rc admin bucket [OPTIONS] ", + expected_tokens: &["migration"], + }, + HelpCase { + args: &["admin", "bucket", "migration"], + usage: "Usage: rc admin bucket migration [OPTIONS] ", + expected_tokens: &["set", "get", "rm", "status", "backfill"], + }, + HelpCase { + args: &["admin", "bucket", "migration", "set"], + usage: "Usage: rc admin bucket migration set [OPTIONS] --provider --region --source-bucket ", + expected_tokens: &[ + "--endpoint", + "--prefix", + "--source-prefix", + "--access-key", + "--secret-key", + "--public", + "--path-style", + "--skip-tls-verify", + "--ca-cert", + "--head", + "--range-get", + "--source-error", + "--no-preserve-etag", + "--copy-tags", + "--no-events", + "--inline-max-bytes", + "--max-concurrent-pulls", + "--dry-run", + "RC_ODM_SECRET_KEY", + ], + }, + HelpCase { + args: &["admin", "bucket", "migration", "backfill"], + usage: "Usage: rc admin bucket migration backfill [OPTIONS] ", + expected_tokens: &["start", "cancel", "status"], + }, + HelpCase { + args: &["admin", "bucket", "migration", "backfill", "start"], + usage: "Usage: rc admin bucket migration backfill start [OPTIONS] ", + expected_tokens: &["--prefix", "--skip-existing", "--dry-run"], + }, HelpCase { args: &["bucket"], usage: "Usage: rc bucket [OPTIONS] ", diff --git a/crates/cli/tests/output_schema_v3.rs b/crates/cli/tests/output_schema_v3.rs index 7a8adc50..799c6b8a 100644 --- a/crates/cli/tests/output_schema_v3.rs +++ b/crates/cli/tests/output_schema_v3.rs @@ -27,6 +27,7 @@ const V3_FAMILIES: &[&str] = &[ "bucket_operations", "iam_policy_entities", "iam_policy_detach", + "on_demand_migration", ]; fn repository_root() -> PathBuf { diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index 208822f6..92b0086c 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -16,6 +16,7 @@ mod kms; mod kms_diagnostic; mod observability; mod oidc; +mod on_demand_migration; mod replication; mod site; pub mod tier; @@ -113,6 +114,18 @@ pub use oidc::{ OidcProvider, OidcProviderList, OidcProviderSource, OidcReadApi, OidcValidationRequest, OidcValidationResult, }; +pub use on_demand_migration::{ + BackfillJob, BackfillJobResult, BackfillLastError, BackfillOwner, BackfillStartRequest, + BackfillSummary, BreakerStatus, FilterRequest, FilterView, HeadPolicy, LastSourceError, + LatencyBucket, MAX_CONCURRENT_PULLS, MAX_INLINE_MAX_BYTES, + MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES, MAX_ON_DEMAND_MIGRATION_RESPONSE_BYTES, + ON_DEMAND_MIGRATION_CAPABILITY, OnDemandMigrationApi, OnDemandMigrationConfigRequest, + OnDemandMigrationConfigResult, OnDemandMigrationConfigView, OnDemandMigrationSetResult, + OnDemandMigrationStatus, PathStyle, PolicyConfig, ProbeSummary, REDACTED_SECRET, + RangeGetPolicy, RuntimeCounters, SkipExisting, SourceCredentialsRequest, SourceCredentialsView, + SourceErrorPolicy, SourceLatency, SourceProvider, SourceRequest, SourceTimeout, SourceView, + TlsRequest, TlsView, is_terminal_backfill_state, validate_ca_cert_pem, validate_local_bucket, +}; pub use replication::{ MAX_REPLICATION_DIFF_RESPONSE_BYTES, MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, ReplicationCountSize, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry, diff --git a/crates/core/src/admin/on_demand_migration.rs b/crates/core/src/admin/on_demand_migration.rs new file mode 100644 index 00000000..cb557019 --- /dev/null +++ b/crates/core/src/admin/on_demand_migration.rs @@ -0,0 +1,1210 @@ +//! On-Demand Migration administration, independent of HTTP transport. +//! +//! A bucket names an external S3-compatible source bucket. A GET that misses +//! locally is served from the source and stored locally, and a background +//! backfill job pulls the remainder. The wire contract is pinned by the +//! vendored fixtures in `tests/fixtures/on_demand_migration/`. +//! +//! Two shapes live here on purpose. [`OnDemandMigrationConfigRequest`] is what +//! `rc` sends: it carries the plaintext secret in zeroizing storage and is +//! serialized once, into a zeroizing buffer. [`OnDemandMigrationConfigView`] is +//! what the server returns: every field is optional with the server's own +//! default, so an older server that omits a field still parses, and the +//! credential values are dropped during deserialization so nothing downstream +//! can print them. + +use crate::{Error, Result}; +use async_trait::async_trait; +use serde::de::Deserializer; +use serde::ser::{SerializeStruct, Serializer}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use zeroize::{Zeroize, Zeroizing}; + +/// Capability label used in unsupported-feature diagnostics. +pub const ON_DEMAND_MIGRATION_CAPABILITY: &str = "admin.on-demand-migration"; + +/// Upper bound for one admin response body. A status document with latency +/// histograms is a few kilobytes; a megabyte is generous without being unbounded. +pub const MAX_ON_DEMAND_MIGRATION_RESPONSE_BYTES: usize = 1024 * 1024; + +/// Upper bound for a CA bundle passed with `--ca-cert`. +pub const MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES: usize = 64 * 1024; + +/// The placeholder the server substitutes for a credential in every response. +pub const REDACTED_SECRET: &str = "REDACTED"; + +/// Largest value the server accepts for `policy.inline_max_bytes` (256 MiB). +pub const MAX_INLINE_MAX_BYTES: u64 = 256 * 1024 * 1024; + +/// Largest value the server accepts for `policy.max_concurrent_pulls`. +pub const MAX_CONCURRENT_PULLS: u32 = 256; + +// --------------------------------------------------------------------------- +// Enumerations shared by the request and the view +// --------------------------------------------------------------------------- + +/// Source vendor family for the S3-speaking providers `rc` can configure. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SourceProvider { + /// Generic S3-compatible endpoint. + S3, + Aws, + Minio, + Rustfs, + R2, + /// GCS XML interoperability API with HMAC keys. + Gcs, +} + +impl SourceProvider { + pub const fn as_str(self) -> &'static str { + match self { + Self::S3 => "s3", + Self::Aws => "aws", + Self::Minio => "minio", + Self::Rustfs => "rustfs", + Self::R2 => "r2", + Self::Gcs => "gcs", + } + } + + /// Only AWS derives its endpoint from the region. + pub const fn requires_endpoint(self) -> bool { + !matches!(self, Self::Aws) + } +} + +/// Bucket addressing style; `auto` is resolved by the server. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum PathStyle { + #[default] + Auto, + Path, + Virtual, +} + +impl PathStyle { + pub const fn as_str(self) -> &'static str { + match self { + Self::Auto => "auto", + Self::Path => "path", + Self::Virtual => "virtual", + } + } +} + +/// What a HEAD that misses locally does. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HeadPolicy { + #[default] + Proxy, + LocalOnly, +} + +impl HeadPolicy { + pub const fn as_str(self) -> &'static str { + match self { + Self::Proxy => "proxy", + Self::LocalOnly => "local_only", + } + } +} + +/// Whether a Range GET also queues a whole-object background pull. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RangeGetPolicy { + #[default] + ServeAndBackfill, + ServeOnly, +} + +impl RangeGetPolicy { + pub const fn as_str(self) -> &'static str { + match self { + Self::ServeAndBackfill => "serve_and_backfill", + Self::ServeOnly => "serve_only", + } + } +} + +/// How a source failure is answered to the client. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SourceErrorPolicy { + #[default] + Propagate, + NotFound, +} + +impl SourceErrorPolicy { + pub const fn as_str(self) -> &'static str { + match self { + Self::Propagate => "propagate", + Self::NotFound => "not_found", + } + } +} + +/// When the backfill job treats a local object as already migrated. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SkipExisting { + #[default] + Always, + EtagOrSize, +} + +impl SkipExisting { + pub const fn as_str(self) -> &'static str { + match self { + Self::Always => "always", + Self::EtagOrSize => "etag_or_size", + } + } +} + +// --------------------------------------------------------------------------- +// Request shape +// --------------------------------------------------------------------------- + +/// Static credentials for the source, with the secret in zeroizing storage. +/// +/// `Debug` never prints the secret or the session token. +#[derive(Clone)] +pub struct SourceCredentialsRequest { + pub access_key: String, + pub secret_key: Zeroizing, + pub session_token: Option>, +} + +impl std::fmt::Debug for SourceCredentialsRequest { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("SourceCredentialsRequest") + .field("access_key", &self.access_key) + .field("secret_key", &REDACTED_SECRET) + .field( + "session_token", + &self.session_token.as_ref().map(|_| REDACTED_SECRET), + ) + .finish() + } +} + +impl Serialize for SourceCredentialsRequest { + fn serialize(&self, serializer: S) -> std::result::Result { + let mut state = serializer.serialize_struct("SourceCredentials", 3)?; + state.serialize_field("access_key", &self.access_key)?; + state.serialize_field("secret_key", self.secret_key.as_str())?; + state.serialize_field( + "session_token", + &self.session_token.as_deref().map(String::as_str), + )?; + state.end() + } +} + +/// TLS settings for the source connection. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct TlsRequest { + pub skip_verify: bool, + pub ca_cert_pem: Option, +} + +/// The external source bucket. +#[derive(Clone, Debug, Serialize)] +pub struct SourceRequest { + pub provider: SourceProvider, + pub endpoint: Option, + pub region: String, + pub bucket: String, + pub path_style: PathStyle, + /// `None` means anonymous access to a public source bucket. + pub credentials: Option, + pub tls: TlsRequest, +} + +/// Key filters. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct FilterRequest { + /// Only local keys with this prefix consult the source. + pub prefix: Option, + /// Prepended to the local key to form the source key. + pub source_prefix: Option, +} + +/// Per-request source timeouts, in milliseconds. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceTimeout { + #[serde(default = "default_connect_ms")] + pub connect_ms: u64, + #[serde(default = "default_first_byte_ms")] + pub first_byte_ms: u64, + #[serde(default = "default_idle_ms")] + pub idle_ms: u64, +} + +impl Default for SourceTimeout { + fn default() -> Self { + Self { + connect_ms: default_connect_ms(), + first_byte_ms: default_first_byte_ms(), + idle_ms: default_idle_ms(), + } + } +} + +/// Read-path policy. +/// +/// The request sends every field explicitly, filled with the server defaults +/// the fixtures pin, so the body `rc` produces is the documented wire shape +/// rather than a partial document whose meaning depends on server defaults. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PolicyConfig { + #[serde(default)] + pub head: HeadPolicy, + #[serde(default)] + pub range_get: RangeGetPolicy, + #[serde(default)] + pub source_error: SourceErrorPolicy, + #[serde(default)] + pub list_through: bool, + #[serde(default = "default_true")] + pub respect_local_delete_marker: bool, + #[serde(default = "default_true")] + pub preserve_etag: bool, + #[serde(default)] + pub copy_tags: bool, + #[serde(default = "default_true")] + pub emit_events: bool, + #[serde(default = "default_negative_cache_ttl_secs")] + pub negative_cache_ttl_secs: u64, + #[serde(default = "default_inline_max_bytes")] + pub inline_max_bytes: u64, + #[serde(default = "default_multipart_part_size_bytes")] + pub multipart_part_size_bytes: u64, + #[serde(default = "default_max_concurrent_pulls")] + pub max_concurrent_pulls: u32, + #[serde(default = "default_pull_queue_capacity")] + pub pull_queue_capacity: u32, + #[serde(default)] + pub source_timeout: SourceTimeout, + #[serde(default)] + pub bandwidth_limit_bytes_per_sec: Option, +} + +impl Default for PolicyConfig { + fn default() -> Self { + Self { + head: HeadPolicy::default(), + range_get: RangeGetPolicy::default(), + source_error: SourceErrorPolicy::default(), + list_through: false, + respect_local_delete_marker: true, + preserve_etag: true, + copy_tags: false, + emit_events: true, + negative_cache_ttl_secs: default_negative_cache_ttl_secs(), + inline_max_bytes: default_inline_max_bytes(), + multipart_part_size_bytes: default_multipart_part_size_bytes(), + max_concurrent_pulls: default_max_concurrent_pulls(), + pull_queue_capacity: default_pull_queue_capacity(), + source_timeout: SourceTimeout::default(), + bandwidth_limit_bytes_per_sec: None, + } + } +} + +const fn default_true() -> bool { + true +} +const fn default_version() -> u32 { + 1 +} +const fn default_negative_cache_ttl_secs() -> u64 { + 30 +} +const fn default_inline_max_bytes() -> u64 { + 16 * 1024 * 1024 +} +const fn default_multipart_part_size_bytes() -> u64 { + 64 * 1024 * 1024 +} +const fn default_max_concurrent_pulls() -> u32 { + 8 +} +const fn default_pull_queue_capacity() -> u32 { + 1024 +} +const fn default_connect_ms() -> u64 { + 5000 +} +const fn default_first_byte_ms() -> u64 { + 15_000 +} +const fn default_idle_ms() -> u64 { + 30_000 +} + +/// The document `PUT .../on-demand-migration/{bucket}` accepts. +#[derive(Clone, Debug, Serialize)] +pub struct OnDemandMigrationConfigRequest { + pub version: u32, + pub enabled: bool, + pub source: SourceRequest, + pub filter: FilterRequest, + pub policy: PolicyConfig, +} + +impl OnDemandMigrationConfigRequest { + /// A version-1, enabled configuration with default filter and policy. + pub fn new(source: SourceRequest) -> Self { + Self { + version: default_version(), + enabled: true, + source, + filter: FilterRequest::default(), + policy: PolicyConfig::default(), + } + } + + /// Reject locally what the server would reject, before any network access + /// and before a secret is read. Every failure is a usage error. + pub fn validate(&self) -> Result<()> { + let source = &self.source; + match source.endpoint.as_deref() { + Some(endpoint) => validate_endpoint(endpoint)?, + None if source.provider.requires_endpoint() => { + return Err(Error::Config(format!( + "--endpoint is required for provider {}", + source.provider.as_str() + ))); + } + None => {} + } + if source.region.trim().is_empty() { + return Err(Error::Config("--region must not be empty".into())); + } + validate_source_bucket(&source.bucket)?; + if let Some(credentials) = &source.credentials { + if credentials.access_key.is_empty() { + return Err(Error::Config("--access-key must not be empty".into())); + } + if credentials.secret_key.is_empty() { + return Err(Error::Config( + "The source secret key must not be empty".into(), + )); + } + if credentials + .session_token + .as_ref() + .is_some_and(|token| token.is_empty()) + { + return Err(Error::Config( + "The source session token must not be empty".into(), + )); + } + } + if let Some(pem) = source.tls.ca_cert_pem.as_deref() { + validate_ca_cert_pem(pem)?; + } + for (flag, value) in [ + ("--prefix", &self.filter.prefix), + ("--source-prefix", &self.filter.source_prefix), + ] { + if value.as_deref().is_some_and(str::is_empty) { + return Err(Error::Config(format!("{flag} must not be empty"))); + } + } + let policy = &self.policy; + if policy.inline_max_bytes > MAX_INLINE_MAX_BYTES { + return Err(Error::Config(format!( + "--inline-max-bytes must be at most {MAX_INLINE_MAX_BYTES}" + ))); + } + if !(1..=MAX_CONCURRENT_PULLS).contains(&policy.max_concurrent_pulls) { + return Err(Error::Config(format!( + "--max-concurrent-pulls must be between 1 and {MAX_CONCURRENT_PULLS}" + ))); + } + Ok(()) + } + + /// Serialize into a zeroizing buffer. The result is the only copy of the + /// plaintext body; callers hand it to the transport without cloning. + pub fn to_wire_json(&self) -> Result>> { + self.validate()?; + Ok(Zeroizing::new(serde_json::to_vec(self)?)) + } +} + +/// `http(s)://host[:port]` with nothing else: no path, query, fragment or +/// userinfo. Userinfo would smuggle a credential into a log line. +fn validate_endpoint(endpoint: &str) -> Result<()> { + let parsed = url::Url::parse(endpoint) + .map_err(|_| Error::Config("--endpoint must be an http(s) URL".into()))?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(Error::Config("--endpoint must use http or https".into())); + } + if parsed.host_str().is_none_or(str::is_empty) { + return Err(Error::Config("--endpoint must name a host".into())); + } + if !parsed.username().is_empty() || parsed.password().is_some() { + return Err(Error::Config( + "--endpoint must not embed credentials; pass them with --access-key".into(), + )); + } + if !matches!(parsed.path(), "" | "/") || parsed.query().is_some() || parsed.fragment().is_some() + { + return Err(Error::Config( + "--endpoint must be scheme://host[:port] with no path, query or fragment".into(), + )); + } + Ok(()) +} + +fn validate_source_bucket(bucket: &str) -> Result<()> { + if bucket.is_empty() || bucket.contains('/') || bucket.chars().any(char::is_whitespace) { + return Err(Error::Config( + "--source-bucket must be a non-empty bucket name without '/' or whitespace".into(), + )); + } + Ok(()) +} + +/// The server requires a PEM certificate block; checking here turns a wrong +/// file into a usage error before the secret is prompted for. +pub fn validate_ca_cert_pem(pem: &str) -> Result<()> { + if pem.len() > MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES { + return Err(Error::Config(format!( + "--ca-cert exceeds {MAX_ON_DEMAND_MIGRATION_CA_CERT_BYTES} bytes" + ))); + } + if !pem.contains("-----BEGIN CERTIFICATE-----") { + return Err(Error::Config( + "--ca-cert must contain a PEM certificate (-----BEGIN CERTIFICATE-----)".into(), + )); + } + Ok(()) +} + +/// A local bucket name as it appears in the admin route. Anything that could +/// change the route (a slash, a dot segment, whitespace) is refused here so the +/// transport never has to reason about it. +pub fn validate_local_bucket(bucket: &str) -> Result<()> { + if bucket.is_empty() + || bucket.len() > 255 + || matches!(bucket, "." | "..") + || bucket + .chars() + .any(|c| c == '/' || c == '\\' || c == '%' || c == '?' || c == '#' || c.is_whitespace()) + { + return Err(Error::InvalidPath( + "Expected alias/bucket with a plain bucket name".into(), + )); + } + Ok(()) +} + +/// Body of `POST .../backfill?op=start`; every field is optional. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct BackfillStartRequest { + #[serde(skip_serializing_if = "Option::is_none")] + pub prefix: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub skip_existing: Option, + /// List and count only; nothing is queued. + pub dry_run: bool, +} + +impl BackfillStartRequest { + pub fn validate(&self) -> Result<()> { + if self.prefix.as_deref().is_some_and(str::is_empty) { + return Err(Error::Config("--prefix must not be empty".into())); + } + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Response shapes +// --------------------------------------------------------------------------- + +/// Redacted credential summary. Only presence survives deserialization: the +/// server substitutes `REDACTED`, and a server that did not must still never +/// reach stdout, so the values are discarded at the parsing boundary. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SourceCredentialsView { + pub access_key: String, + pub has_secret_key: bool, + pub has_session_token: bool, +} + +impl<'de> Deserialize<'de> for SourceCredentialsView { + fn deserialize>(deserializer: D) -> std::result::Result { + #[derive(Deserialize)] + struct Wire { + #[serde(default)] + access_key: String, + #[serde(default)] + secret_key: Option, + #[serde(default)] + session_token: Option, + } + let wire = Wire::deserialize(deserializer)?; + // Wipe whatever the server sent before it goes out of scope. + let mut secret_key = Zeroizing::new(wire.secret_key.unwrap_or_default()); + let mut session_token = Zeroizing::new(wire.session_token.unwrap_or_default()); + let view = Self { + access_key: wire.access_key, + has_secret_key: !secret_key.is_empty(), + has_session_token: !session_token.is_empty(), + }; + secret_key.zeroize(); + session_token.zeroize(); + Ok(view) + } +} + +impl Serialize for SourceCredentialsView { + fn serialize(&self, serializer: S) -> std::result::Result { + let mut state = serializer.serialize_struct("SourceCredentials", 3)?; + state.serialize_field("access_key", &self.access_key)?; + state.serialize_field( + "secret_key", + &self.has_secret_key.then_some(REDACTED_SECRET), + )?; + state.serialize_field( + "session_token", + &self.has_session_token.then_some(REDACTED_SECRET), + )?; + state.end() + } +} + +/// TLS settings as returned by the server. The CA bundle is public material, +/// but it is long; only its presence is kept for display. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct TlsView { + pub skip_verify: bool, + pub has_ca_cert: bool, +} + +impl<'de> Deserialize<'de> for TlsView { + fn deserialize>(deserializer: D) -> std::result::Result { + #[derive(Deserialize)] + struct Wire { + #[serde(default)] + skip_verify: bool, + #[serde(default)] + ca_cert_pem: Option, + } + let wire = Wire::deserialize(deserializer)?; + Ok(Self { + skip_verify: wire.skip_verify, + has_ca_cert: wire.ca_cert_pem.is_some_and(|pem| !pem.is_empty()), + }) + } +} + +impl Serialize for TlsView { + fn serialize(&self, serializer: S) -> std::result::Result { + let mut state = serializer.serialize_struct("Tls", 2)?; + state.serialize_field("skip_verify", &self.skip_verify)?; + state.serialize_field("has_ca_cert", &self.has_ca_cert)?; + state.end() + } +} + +/// The source as returned by the server. `provider` stays a string so a +/// provider this build does not know (for example `azure`) still displays. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceView { + #[serde(default)] + pub provider: String, + #[serde(default)] + pub endpoint: Option, + #[serde(default)] + pub region: String, + #[serde(default)] + pub bucket: String, + #[serde(default)] + pub path_style: PathStyle, + #[serde(default)] + pub credentials: Option, + #[serde(default)] + pub tls: TlsView, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct FilterView { + #[serde(default)] + pub prefix: Option, + #[serde(default)] + pub source_prefix: Option, +} + +/// The redacted configuration document. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationConfigView { + #[serde(default = "default_version")] + pub version: u32, + #[serde(default = "default_true")] + pub enabled: bool, + #[serde(default)] + pub source: SourceView, + #[serde(default)] + pub filter: FilterView, + #[serde(default)] + pub policy: PolicyConfig, +} + +impl Default for OnDemandMigrationConfigView { + fn default() -> Self { + Self { + version: default_version(), + enabled: true, + source: SourceView::default(), + filter: FilterView::default(), + policy: PolicyConfig::default(), + } + } +} + +/// What the `PUT` probe learned about the source. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProbeSummary { + #[serde(default)] + pub reachable: bool, + #[serde(default)] + pub listable: bool, + #[serde(default)] + pub sample_key: Option, +} + +/// Response of `PUT .../on-demand-migration/{bucket}`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSetResult { + #[serde(default)] + pub bucket: String, + #[serde(default)] + pub dry_run: bool, + #[serde(default)] + pub config: Option, + /// `None` for a dry run, which saves nothing. + #[serde(default)] + pub updated_at: Option, + #[serde(default)] + pub probe: Option, +} + +/// Response of `GET .../on-demand-migration/{bucket}`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationConfigResult { + #[serde(default)] + pub bucket: String, + #[serde(default)] + pub config: Option, + #[serde(default)] + pub updated_at: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BreakerStatus { + #[serde(default)] + pub state: String, + #[serde(default)] + pub opened_at: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct LatencyBucket { + #[serde(default)] + pub le_ms: u64, + #[serde(default)] + pub count: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceLatency { + #[serde(default)] + pub buckets: Vec, + #[serde(default)] + pub count: u64, + #[serde(default)] + pub sum_ms: u64, +} + +/// Per-node runtime counters. The nested maps keep the outcome and path +/// labels open-ended so a new label on the server displays instead of failing. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeCounters { + /// Operation (`get`, `head`) to outcome (`source_hit`, `source_miss`, ...) to count. + #[serde(default)] + pub requests_total: BTreeMap>, + #[serde(default)] + pub pulled_bytes_total: u64, + /// Pull path (`inline`, `background`, `backfill`) to count. + #[serde(default)] + pub pulled_objects_total: BTreeMap, + /// Failure class to count. + #[serde(default)] + pub pull_failures_total: BTreeMap, + #[serde(default)] + pub source_latency: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct LastSourceError { + #[serde(default)] + pub class: String, + #[serde(default)] + pub at: Option, +} + +/// Counters of the bucket's backfill job as embedded in the status document. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillSummary { + #[serde(default)] + pub job_id: String, + #[serde(default)] + pub state: String, + #[serde(default)] + pub listed: u64, + #[serde(default)] + pub enqueued: u64, + #[serde(default)] + pub pulled: u64, + #[serde(default)] + pub skipped_existing: u64, + #[serde(default)] + pub failed: u64, + #[serde(default)] + pub bytes: u64, + #[serde(default)] + pub updated_at: Option, +} + +/// Response of `GET .../on-demand-migration/{bucket}/status`. +/// +/// This is the answering node's view: counters, queue depth and breaker state +/// are per node, while the configuration is cluster-wide. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct OnDemandMigrationStatus { + #[serde(default)] + pub configured: bool, + #[serde(default)] + pub enabled: bool, + #[serde(default)] + pub module_enabled: bool, + #[serde(default)] + pub provider: Option, + #[serde(default)] + pub endpoint_host: Option, + #[serde(default)] + pub breaker: Option, + #[serde(default)] + pub counters: Option, + #[serde(default)] + pub last_source_error: Option, + #[serde(default)] + pub inflight_pulls: u64, + #[serde(default)] + pub queue_depth: u64, + /// Deliberately `null` on the server today. Rendered as an em dash, never + /// as zero: a missing ratio and a zero ratio mean different things. + #[serde(default)] + pub served_by_source_ratio: Option, + #[serde(default)] + pub updated_at: Option, + #[serde(default)] + pub backfill: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillLastError { + #[serde(default)] + pub class: String, + /// Hash of the failing key; the key itself never leaves the server. + #[serde(default)] + pub key_hash: Option, + #[serde(default)] + pub at: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillOwner { + #[serde(default)] + pub node: String, + #[serde(default)] + pub lease_until: Option, +} + +/// The backfill checkpoint document as stored on the server. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillJob { + #[serde(default = "default_version")] + pub format_version: u32, + #[serde(default)] + pub job_id: String, + #[serde(default)] + pub state: String, + #[serde(default)] + pub config_updated_at: Option, + #[serde(default)] + pub prefix: Option, + #[serde(default)] + pub skip_existing: SkipExisting, + #[serde(default)] + pub dry_run: bool, + #[serde(default)] + pub listed: u64, + #[serde(default)] + pub enqueued: u64, + #[serde(default)] + pub pulled: u64, + #[serde(default)] + pub skipped_existing: u64, + #[serde(default)] + pub failed: u64, + #[serde(default)] + pub bytes: u64, + #[serde(default)] + pub last_key: Option, + #[serde(default)] + pub last_error: Option, + #[serde(default)] + pub failed_keys: Vec, + #[serde(default)] + pub started_at: Option, + #[serde(default)] + pub updated_at: Option, + #[serde(default)] + pub owner: Option, +} + +impl BackfillJob { + /// Whether the job can still change. `--watch` stops on a terminal state; + /// an unknown state from a newer server is treated as still running so the + /// watcher keeps refreshing rather than declaring victory early. + pub fn is_terminal(&self) -> bool { + is_terminal_backfill_state(&self.state) + } +} + +pub fn is_terminal_backfill_state(state: &str) -> bool { + matches!( + state, + "cancelled" | "completed" | "completed_with_failures" | "failed" + ) +} + +/// Response of `POST`/`GET .../on-demand-migration/{bucket}/backfill`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillJobResult { + #[serde(default)] + pub bucket: String, + #[serde(default)] + pub job: Option, +} + +// --------------------------------------------------------------------------- +// API +// --------------------------------------------------------------------------- + +/// Administrative operations for on-demand migration. +/// +/// Writes are never automatically retried: a `PUT` probes the source and a +/// backfill start takes a lease, so a repeated request is a second decision. +#[async_trait] +pub trait OnDemandMigrationApi: Send + Sync { + /// Validate, probe and (unless `dry_run`) save the configuration. + async fn set_on_demand_migration( + &self, + bucket: &str, + config: &OnDemandMigrationConfigRequest, + dry_run: bool, + ) -> Result; + + async fn get_on_demand_migration(&self, bucket: &str) -> Result; + + /// Idempotent; already-pulled objects stay in place. + async fn delete_on_demand_migration(&self, bucket: &str) -> Result<()>; + + async fn on_demand_migration_status(&self, bucket: &str) -> Result; + + async fn start_on_demand_migration_backfill( + &self, + bucket: &str, + request: &BackfillStartRequest, + ) -> Result; + + async fn cancel_on_demand_migration_backfill(&self, bucket: &str) -> Result; + + async fn on_demand_migration_backfill_status(&self, bucket: &str) -> Result; +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::{Value, json}; + + const SET_REQUEST: &str = + include_str!("../../tests/fixtures/on_demand_migration/set_request.json"); + const SET_RESPONSE: &str = + include_str!("../../tests/fixtures/on_demand_migration/set_response.json"); + const GET_RESPONSE: &str = + include_str!("../../tests/fixtures/on_demand_migration/get_response.json"); + const STATUS: &str = include_str!("../../tests/fixtures/on_demand_migration/status.json"); + const STATUS_WITH_BACKFILL: &str = + include_str!("../../tests/fixtures/on_demand_migration/status_with_backfill.json"); + const BACKFILL_JOB: &str = + include_str!("../../tests/fixtures/on_demand_migration/backfill_job.json"); + + fn fixture_request() -> OnDemandMigrationConfigRequest { + let mut request = OnDemandMigrationConfigRequest::new(SourceRequest { + provider: SourceProvider::Minio, + endpoint: Some("https://source.example.com:9000".into()), + region: "us-east-1".into(), + bucket: "legacy-photos".into(), + path_style: PathStyle::Auto, + credentials: Some(SourceCredentialsRequest { + access_key: "AKIASOURCE".into(), + secret_key: Zeroizing::new("sourceSecretKey123".into()), + session_token: None, + }), + tls: TlsRequest::default(), + }); + request.filter.source_prefix = Some("photos/".into()); + request + } + + #[test] + fn set_request_matches_the_plaintext_wire_fixture() { + let body = fixture_request().to_wire_json().unwrap(); + let actual: Value = serde_json::from_slice(&body).unwrap(); + let expected: Value = serde_json::from_str(SET_REQUEST).unwrap(); + assert_eq!(actual, expected); + } + + #[test] + fn request_debug_never_prints_the_secret() { + let request = fixture_request(); + let debug = format!("{request:?}"); + assert!(debug.contains("AKIASOURCE")); + assert!(!debug.contains("sourceSecretKey123")); + assert!(debug.contains(REDACTED_SECRET)); + } + + #[test] + fn set_response_parses_and_drops_the_redacted_secret() { + let result: OnDemandMigrationSetResult = serde_json::from_str(SET_RESPONSE).unwrap(); + assert_eq!(result.bucket, "photos"); + assert!(!result.dry_run); + assert_eq!(result.updated_at.as_deref(), Some("2026-09-02T10:00:00Z")); + let probe = result.probe.unwrap(); + assert!(probe.reachable && probe.listable); + assert_eq!(probe.sample_key.as_deref(), Some("photos/2024/01.jpg")); + let config = result.config.unwrap(); + assert_eq!(config.source.provider, "minio"); + let credentials = config.source.credentials.unwrap(); + assert_eq!(credentials.access_key, "AKIASOURCE"); + assert!(credentials.has_secret_key); + assert!(!credentials.has_session_token); + let serialized = serde_json::to_value(&credentials).unwrap(); + assert_eq!(serialized["secret_key"], REDACTED_SECRET); + assert_eq!(serialized["session_token"], Value::Null); + } + + #[test] + fn get_response_matches_the_fixture_and_redacts_a_leaked_secret() { + let result: OnDemandMigrationConfigResult = serde_json::from_str(GET_RESPONSE).unwrap(); + assert_eq!(result.bucket, "photos"); + let config = result.config.unwrap(); + assert!(config.enabled); + assert_eq!(config.version, 1); + assert_eq!(config.filter.source_prefix.as_deref(), Some("photos/")); + assert_eq!(config.policy, PolicyConfig::default()); + assert_eq!(config.source.path_style, PathStyle::Auto); + + // A server that failed to redact must not make it to output either. + let leaked = GET_RESPONSE.replace("\"REDACTED\"", "\"plaintext-secret\""); + let result: OnDemandMigrationConfigResult = serde_json::from_str(&leaked).unwrap(); + let text = serde_json::to_string(&result).unwrap(); + assert!(!text.contains("plaintext-secret")); + assert!(text.contains(REDACTED_SECRET)); + } + + #[test] + fn status_fixture_keeps_the_null_ratio_and_counters() { + let status: OnDemandMigrationStatus = serde_json::from_str(STATUS).unwrap(); + assert!(status.configured && status.enabled && status.module_enabled); + assert_eq!(status.provider.as_deref(), Some("minio")); + assert_eq!(status.endpoint_host.as_deref(), Some("source.example.com")); + assert_eq!(status.breaker.as_ref().unwrap().state, "half_open"); + assert_eq!(status.served_by_source_ratio, None); + assert_eq!(status.inflight_pulls, 1); + assert_eq!(status.queue_depth, 1); + assert!(status.backfill.is_none()); + let counters = status.counters.unwrap(); + assert_eq!(counters.pulled_bytes_total, 4096); + assert_eq!(counters.requests_total["get"]["source_hit"], 2); + assert_eq!(counters.pull_failures_total["source_timeout"], 1); + assert_eq!(counters.source_latency.unwrap().count, 3); + assert_eq!(status.last_source_error.unwrap().class, "server_error"); + } + + #[test] + fn status_with_backfill_fixture_carries_the_summary() { + let status: OnDemandMigrationStatus = serde_json::from_str(STATUS_WITH_BACKFILL).unwrap(); + let backfill = status.backfill.unwrap(); + assert_eq!(backfill.job_id, "11111111-1111-4111-8111-111111111111"); + assert_eq!(backfill.state, "running"); + assert_eq!(backfill.pulled, 1400); + assert_eq!(backfill.bytes, 73_400_320); + } + + #[test] + fn backfill_job_fixture_parses_and_is_not_terminal() { + let result: BackfillJobResult = serde_json::from_str(BACKFILL_JOB).unwrap(); + assert_eq!(result.bucket, "photos"); + let job = result.job.unwrap(); + assert_eq!(job.state, "running"); + assert!(!job.is_terminal()); + assert_eq!(job.skip_existing, SkipExisting::Always); + assert_eq!(job.prefix.as_deref(), Some("photos/")); + assert_eq!(job.last_error.unwrap().class, "source_timeout"); + assert_eq!(job.owner.unwrap().node, "node-a:9000"); + assert_eq!(job.failed_keys, vec!["9f2c3b0a1d4e5f60"]); + for state in [ + "cancelled", + "completed", + "completed_with_failures", + "failed", + ] { + assert!(is_terminal_backfill_state(state), "{state}"); + } + for state in ["pending", "running", "paused", "something_new"] { + assert!(!is_terminal_backfill_state(state), "{state}"); + } + } + + #[test] + fn older_servers_that_omit_fields_still_parse() { + let status: OnDemandMigrationStatus = serde_json::from_str("{}").unwrap(); + assert!(!status.configured); + assert_eq!(status.served_by_source_ratio, None); + let config: OnDemandMigrationConfigResult = + serde_json::from_str(r#"{"bucket":"b","config":{"source":{"provider":"s3"}}}"#) + .unwrap(); + let config = config.config.unwrap(); + assert!(config.enabled); + assert_eq!(config.policy.max_concurrent_pulls, 8); + assert!(config.source.credentials.is_none()); + let job: BackfillJobResult = serde_json::from_str(r#"{"bucket":"b"}"#).unwrap(); + assert!(job.job.is_none()); + // Unknown fields from a newer server are ignored rather than fatal. + let newer: OnDemandMigrationStatus = + serde_json::from_value(json!({"configured": true, "future_field": 1})).unwrap(); + assert!(newer.configured); + } + + #[test] + fn validation_rejects_what_the_server_would() { + let mut request = fixture_request(); + request.source.endpoint = None; + assert!(request.validate().is_err()); + request.source.provider = SourceProvider::Aws; + assert!(request.validate().is_ok()); + + for endpoint in [ + "source.example.com", + "ftp://source.example.com", + "https://user:pw@source.example.com", + "https://source.example.com/path", + "https://source.example.com/?x=1", + "https://source.example.com/#frag", + ] { + let mut request = fixture_request(); + request.source.endpoint = Some(endpoint.into()); + assert!(request.validate().is_err(), "{endpoint}"); + } + let mut request = fixture_request(); + request.source.endpoint = Some("http://127.0.0.1:9000/".into()); + assert!(request.validate().is_ok()); + + let mut request = fixture_request(); + request.source.region = " ".into(); + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.source.bucket = "a/b".into(); + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.filter.prefix = Some(String::new()); + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.policy.inline_max_bytes = MAX_INLINE_MAX_BYTES + 1; + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.policy.max_concurrent_pulls = 0; + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.source.tls.ca_cert_pem = Some("not a certificate".into()); + assert!(request.validate().is_err()); + let mut request = fixture_request(); + request.source.credentials = None; + assert!(request.validate().is_ok()); + assert_eq!( + fixture_request().validate().map_err(|e| e.exit_code()), + Ok(()) + ); + let mut request = fixture_request(); + request.source.endpoint = None; + assert_eq!(request.validate().unwrap_err().exit_code(), 2); + } + + #[test] + fn local_bucket_names_cannot_change_the_route() { + for bucket in ["photos", "my.bucket", "a-b_c"] { + assert!(validate_local_bucket(bucket).is_ok(), "{bucket}"); + } + for bucket in ["", ".", "..", "a/b", "a b", "a%2Fb", "a?x", "a#f", "a\\b"] { + assert_eq!( + validate_local_bucket(bucket).unwrap_err().exit_code(), + 2, + "{bucket}" + ); + } + } + + #[test] + fn backfill_start_request_omits_unset_fields() { + let request = BackfillStartRequest::default(); + assert_eq!( + serde_json::to_value(&request).unwrap(), + json!({"dry_run": false}) + ); + let request = BackfillStartRequest { + prefix: Some("photos/".into()), + skip_existing: Some(SkipExisting::EtagOrSize), + dry_run: true, + }; + assert_eq!( + serde_json::to_value(&request).unwrap(), + json!({"prefix": "photos/", "skip_existing": "etag_or_size", "dry_run": true}) + ); + assert!( + BackfillStartRequest { + prefix: Some(String::new()), + ..Default::default() + } + .validate() + .is_err() + ); + } +} diff --git a/crates/core/tests/fixtures/on_demand_migration/README.md b/crates/core/tests/fixtures/on_demand_migration/README.md new file mode 100644 index 00000000..ba848c18 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/README.md @@ -0,0 +1,24 @@ +# On-Demand Migration wire fixtures + +Vendored verbatim from `crates/madmin/fixtures/on_demand_migration/*.json` in +[rustfs/rustfs](https://github.com/rustfs/rustfs) at commit +`1a88870809896c989465540b519afc74731d7f33` (2026-09-06, "fix(odm): preserve +cursor compatibility and native source semantics (#7238)"). + +These files pin the admin API contract for +`/rustfs/admin/v3/on-demand-migration/{bucket}`. The parse tests in +`crates/core/src/admin/on_demand_migration.rs` and the transport tests in +`crates/s3/src/admin/on_demand_migration.rs` read them instead of hand-written +literals, so a server-side contract change shows up as a fixture diff rather +than a silently drifting test. + +| File | Route | +| --- | --- | +| `set_request.json` | Plaintext `PUT` body (the only place a secret appears) | +| `set_response.json` | `PUT` response with the redacted config and probe summary | +| `get_response.json` | `GET` response with the redacted config | +| `status.json` | `GET .../status` without a backfill job | +| `status_with_backfill.json` | `GET .../status` with a backfill summary | +| `backfill_job.json` | `POST`/`GET .../backfill` checkpoint document | + +Refresh by copying the upstream files again and updating the commit above. diff --git a/crates/core/tests/fixtures/on_demand_migration/backfill_job.json b/crates/core/tests/fixtures/on_demand_migration/backfill_job.json new file mode 100644 index 00000000..b087ce77 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/backfill_job.json @@ -0,0 +1 @@ +{"bucket":"photos","job":{"format_version":1,"job_id":"11111111-1111-4111-8111-111111111111","state":"running","config_updated_at":"2026-09-02T10:00:00Z","prefix":"photos/","skip_existing":"always","dry_run":false,"continuation_token":"cGhvdG9zLzEwMDA=","listed":2000,"enqueued":1500,"pulled":1400,"skipped_existing":500,"failed":3,"bytes":73400320,"last_key":"photos/2024/02.jpg","last_error":{"class":"source_timeout","key_hash":"9f2c3b0a1d4e5f60","at":"2026-09-02T10:05:00Z"},"failed_keys":["9f2c3b0a1d4e5f60"],"started_at":"2026-09-02T10:00:30Z","updated_at":"2026-09-02T10:05:10Z","owner":{"node":"node-a:9000","lease_until":"2026-09-02T10:06:10Z"}}} diff --git a/crates/core/tests/fixtures/on_demand_migration/get_response.json b/crates/core/tests/fixtures/on_demand_migration/get_response.json new file mode 100644 index 00000000..aff3a808 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/get_response.json @@ -0,0 +1 @@ +{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"} diff --git a/crates/core/tests/fixtures/on_demand_migration/set_request.json b/crates/core/tests/fixtures/on_demand_migration/set_request.json new file mode 100644 index 00000000..e5b7fb03 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/set_request.json @@ -0,0 +1 @@ +{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}} diff --git a/crates/core/tests/fixtures/on_demand_migration/set_response.json b/crates/core/tests/fixtures/on_demand_migration/set_response.json new file mode 100644 index 00000000..81bf6966 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/set_response.json @@ -0,0 +1 @@ +{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}} diff --git a/crates/core/tests/fixtures/on_demand_migration/status.json b/crates/core/tests/fixtures/on_demand_migration/status.json new file mode 100644 index 00000000..86acc06d --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/status.json @@ -0,0 +1 @@ +{"configured":true,"enabled":true,"module_enabled":true,"provider":"minio","endpoint_host":"source.example.com","breaker":{"state":"half_open","opened_at":null},"counters":{"requests_total":{"get":{"breaker_open":0,"filtered":0,"negative_cached":0,"source_error":0,"source_hit":2,"source_miss":0,"unsupported":0},"head":{"breaker_open":0,"filtered":0,"negative_cached":1,"source_error":0,"source_hit":0,"source_miss":0,"unsupported":0}},"pulled_bytes_total":4096,"pulled_objects_total":{"backfill":0,"background":0,"inline":1},"pull_failures_total":{"canceled":0,"etag_mismatch":0,"local_write":0,"queue_full":0,"quota":0,"source_access_denied":0,"source_connect":0,"source_not_found":0,"source_other":0,"source_server_error":0,"source_throttled":0,"source_timeout":1,"source_unsupported":0},"source_latency":{"buckets":[{"le_ms":5,"count":1},{"le_ms":10,"count":1},{"le_ms":20,"count":1},{"le_ms":50,"count":1},{"le_ms":100,"count":1},{"le_ms":200,"count":1},{"le_ms":500,"count":1},{"le_ms":1000,"count":2},{"le_ms":2000,"count":2},{"le_ms":5000,"count":2},{"le_ms":10000,"count":2},{"le_ms":20000,"count":2},{"le_ms":30000,"count":2},{"le_ms":60000,"count":2}],"count":3,"sum_ms":90753}},"last_source_error":{"class":"server_error","at":"2026-09-02T10:00:00Z"},"inflight_pulls":1,"queue_depth":1,"served_by_source_ratio":null,"updated_at":"2026-09-02T10:00:00Z"} diff --git a/crates/core/tests/fixtures/on_demand_migration/status_with_backfill.json b/crates/core/tests/fixtures/on_demand_migration/status_with_backfill.json new file mode 100644 index 00000000..6dc1df25 --- /dev/null +++ b/crates/core/tests/fixtures/on_demand_migration/status_with_backfill.json @@ -0,0 +1 @@ +{"configured":true,"enabled":true,"module_enabled":true,"provider":"minio","endpoint_host":"source.example.com","breaker":{"state":"half_open","opened_at":null},"counters":{"requests_total":{"get":{"breaker_open":0,"filtered":0,"negative_cached":0,"source_error":0,"source_hit":2,"source_miss":0,"unsupported":0},"head":{"breaker_open":0,"filtered":0,"negative_cached":1,"source_error":0,"source_hit":0,"source_miss":0,"unsupported":0}},"pulled_bytes_total":4096,"pulled_objects_total":{"backfill":0,"background":0,"inline":1},"pull_failures_total":{"canceled":0,"etag_mismatch":0,"local_write":0,"queue_full":0,"quota":0,"source_access_denied":0,"source_connect":0,"source_not_found":0,"source_other":0,"source_server_error":0,"source_throttled":0,"source_timeout":1,"source_unsupported":0},"source_latency":{"buckets":[{"le_ms":5,"count":1},{"le_ms":10,"count":1},{"le_ms":20,"count":1},{"le_ms":50,"count":1},{"le_ms":100,"count":1},{"le_ms":200,"count":1},{"le_ms":500,"count":1},{"le_ms":1000,"count":2},{"le_ms":2000,"count":2},{"le_ms":5000,"count":2},{"le_ms":10000,"count":2},{"le_ms":20000,"count":2},{"le_ms":30000,"count":2},{"le_ms":60000,"count":2}],"count":3,"sum_ms":90753}},"last_source_error":{"class":"server_error","at":"2026-09-02T10:00:00Z"},"inflight_pulls":1,"queue_depth":1,"served_by_source_ratio":null,"updated_at":"2026-09-02T10:00:00Z","backfill":{"job_id":"11111111-1111-4111-8111-111111111111","state":"running","listed":2000,"enqueued":1500,"pulled":1400,"skipped_existing":500,"failed":3,"bytes":73400320,"updated_at":"2026-09-02T10:05:10Z"}} diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index e7beecf4..cc48da4b 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -11076,3 +11076,4 @@ mod tests { } mod catalog; +mod on_demand_migration; diff --git a/crates/s3/src/admin/on_demand_migration.rs b/crates/s3/src/admin/on_demand_migration.rs new file mode 100644 index 00000000..b18b075c --- /dev/null +++ b/crates/s3/src/admin/on_demand_migration.rs @@ -0,0 +1,346 @@ +//! On-Demand Migration admin transport over the existing SigV4 client. +//! +//! Every route lives under `/rustfs/admin/v3/on-demand-migration/{bucket}`. +//! The `PUT` body carries a plaintext source secret, so it is built once in +//! zeroizing storage, handed to the HTTP client by ownership, and never +//! echoed: a transport failure reports a fixed message rather than the +//! request, and server error bodies are bounded and credential-scrubbed. + +use super::{AdminClient, SensitiveRequestBody, parse_admin_error, read_bounded_response_body}; +use async_trait::async_trait; +use bytes::Bytes; +use rc_core::admin::{ + BackfillJobResult, BackfillStartRequest, MAX_ON_DEMAND_MIGRATION_RESPONSE_BYTES, + OnDemandMigrationApi, OnDemandMigrationConfigRequest, OnDemandMigrationConfigResult, + OnDemandMigrationSetResult, OnDemandMigrationStatus, validate_local_bucket, +}; +use rc_core::{Error, Result}; +use reqwest::{Method, StatusCode}; +use serde::de::DeserializeOwned; +use zeroize::Zeroizing; + +/// Message for a 404 that is not one of the route's own not-found answers: +/// the whole route family is absent, so the server predates the feature. +pub const UNSUPPORTED_MESSAGE: &str = "server does not support on-demand migration"; + +/// Error codes the route family answers 404 with when the route itself exists. +const KNOWN_NOT_FOUND_CODES: &[&str] = + &["NoSuchConfiguration", "NoSuchBucket", "NoSuchBackfillJob"]; + +/// `400` code meaning the probe could not reach the source: a network class, +/// not a usage class, because the configuration itself was accepted. +const SOURCE_UNREACHABLE_CODE: &str = "OnDemandMigrationSourceUnreachable"; + +/// Longest server message echoed back to the operator. +const MAX_ERROR_MESSAGE_CHARS: usize = 512; + +impl AdminClient { + fn on_demand_migration_url( + &self, + bucket: &str, + suffix: &str, + query: &[(&str, &str)], + ) -> String { + let mut url = self.admin_url(&format!( + "/on-demand-migration/{}{suffix}", + urlencoding::encode(bucket) + )); + let query_string = query + .iter() + .map(|(key, value)| { + format!( + "{}={}", + urlencoding::encode(key), + urlencoding::encode(value) + ) + }) + .collect::>() + .join("&"); + if !query_string.is_empty() { + url.push('?'); + url.push_str(&query_string); + } + url + } + + /// One signed request to the route family with a bounded response. + /// + /// The body, when present, is owned by the request so the only plaintext + /// copy of the secret is wiped when the HTTP client drops it. + async fn on_demand_migration_request( + &self, + method: Method, + bucket: &str, + suffix: &str, + query: &[(&str, &str)], + body: Option>>, + ) -> Result<(StatusCode, Vec)> { + validate_local_bucket(bucket)?; + let url = self.on_demand_migration_url(bucket, suffix, query); + let body_bytes = body + .as_ref() + .map(|body| body.as_slice()) + .unwrap_or_default(); + let headers = self.request_headers(body_bytes)?; + let signed_headers = self + .sign_request(&method, &url, &headers, body_bytes) + .await?; + let write = !matches!(method, Method::GET | Method::HEAD); + let mut request = self.http_client.request(method, &url); + for (name, value) in signed_headers.iter() { + request = request.header(name, value); + } + if let Some(body) = body { + request = request.body(Bytes::from_owner(SensitiveRequestBody(body))); + } + // A fixed message: the reqwest error can embed the URL, and a write + // whose outcome is unknown must not be retried by a caller. + let response = request.send().await.map_err(|_| { + Error::Network(if write { + "On-demand migration request failed; outcome unknown and not retried".to_string() + } else { + "On-demand migration request failed".to_string() + }) + })?; + let status = response.status(); + let bytes = read_bounded_response_body( + response, + MAX_ON_DEMAND_MIGRATION_RESPONSE_BYTES, + "On-demand migration response", + ) + .await?; + if !status.is_success() { + let mut body_text = String::from_utf8_lossy(&bytes).into_owned(); + self.redact_admin_credentials(&mut body_text); + return Err(map_on_demand_migration_error(status, &body_text)); + } + Ok((status, bytes)) + } + + async fn on_demand_migration_json( + &self, + method: Method, + bucket: &str, + suffix: &str, + query: &[(&str, &str)], + body: Option>>, + ) -> Result { + let (_, bytes) = self + .on_demand_migration_request(method, bucket, suffix, query, body) + .await?; + if bytes.is_empty() { + return Err(Error::General( + "On-demand migration response was empty".to_string(), + )); + } + serde_json::from_slice(&bytes) + .map_err(|_| Error::General("Invalid on-demand migration JSON response".to_string())) + } +} + +/// Map the route family's answers onto exit classes. +/// +/// | Answer | Class | Exit | +/// |---|---|---| +/// | 400 `OnDemandMigrationSourceUnreachable` | network | 3 | +/// | 400 anything else (validation, module switch off) | usage | 2 | +/// | 401 / 403 (unauthorized, or the licence denies the entitlement) | auth | 4 | +/// | 404 with a known code (no config, no bucket, no job) | not found | 5 | +/// | 404 otherwise (route family absent) | unsupported | 7 | +/// | 409 (a backfill job holds the lease) | conflict | 6 | +/// | 501 (provider excluded at build time) | unsupported | 7 | +/// | 408 / 429 / 5xx | network | 3 | +pub(crate) fn map_on_demand_migration_error(status: StatusCode, body: &str) -> Error { + let structured = parse_admin_error(body); + let code = structured.as_ref().and_then(|error| error.code.clone()); + let message = structured + .as_ref() + .and_then(|error| error.message.clone()) + .map(|message| sanitize_message(&message)) + .filter(|message| !message.is_empty()); + let describe = |fallback: &str| { + let mut text = format!("HTTP {}", status.as_u16()); + if let Some(code) = &code { + text.push(' '); + text.push_str(&sanitize_message(code)); + } + text.push_str(": "); + text.push_str(message.as_deref().unwrap_or(fallback)); + text + }; + match status { + StatusCode::BAD_REQUEST if code.as_deref() == Some(SOURCE_UNREACHABLE_CODE) => { + Error::Network(describe("the source bucket did not answer the probe")) + } + StatusCode::BAD_REQUEST => Error::Config(describe("the request was rejected")), + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => Error::Auth(describe( + "the admin action is not authorized or the licence denies it", + )), + StatusCode::NOT_FOUND + if code + .as_deref() + .is_some_and(|code| KNOWN_NOT_FOUND_CODES.contains(&code)) => + { + Error::NotFound(describe("not found")) + } + StatusCode::NOT_FOUND => Error::UnsupportedFeature(UNSUPPORTED_MESSAGE.to_string()), + StatusCode::CONFLICT => Error::Conflict(describe("a backfill job is already running")), + StatusCode::NOT_IMPLEMENTED => { + Error::UnsupportedFeature(describe("the provider is not compiled into this server")) + } + StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS => { + Error::Network(describe("the server asked for a retry")) + } + status if status.is_server_error() => Error::Network(describe("server error")), + _ => Error::General(describe("unexpected response")), + } +} + +/// Bound and de-control a server-authored string before it reaches a terminal. +fn sanitize_message(message: &str) -> String { + message + .chars() + .filter(|c| !c.is_control()) + .take(MAX_ERROR_MESSAGE_CHARS) + .collect::() + .trim() + .to_string() +} + +#[async_trait] +impl OnDemandMigrationApi for AdminClient { + async fn set_on_demand_migration( + &self, + bucket: &str, + config: &OnDemandMigrationConfigRequest, + dry_run: bool, + ) -> Result { + validate_local_bucket(bucket)?; + let body = config.to_wire_json()?; + let query: &[(&str, &str)] = if dry_run { &[("dry-run", "true")] } else { &[] }; + self.on_demand_migration_json(Method::PUT, bucket, "", query, Some(body)) + .await + } + + async fn get_on_demand_migration(&self, bucket: &str) -> Result { + self.on_demand_migration_json(Method::GET, bucket, "", &[], None) + .await + } + + async fn delete_on_demand_migration(&self, bucket: &str) -> Result<()> { + self.on_demand_migration_request(Method::DELETE, bucket, "", &[], None) + .await + .map(|_| ()) + } + + async fn on_demand_migration_status(&self, bucket: &str) -> Result { + self.on_demand_migration_json(Method::GET, bucket, "/status", &[], None) + .await + } + + async fn start_on_demand_migration_backfill( + &self, + bucket: &str, + request: &BackfillStartRequest, + ) -> Result { + request.validate()?; + let body = Zeroizing::new(serde_json::to_vec(request)?); + self.on_demand_migration_json( + Method::POST, + bucket, + "/backfill", + &[("op", "start")], + Some(body), + ) + .await + } + + async fn cancel_on_demand_migration_backfill(&self, bucket: &str) -> Result { + self.on_demand_migration_json(Method::POST, bucket, "/backfill", &[("op", "cancel")], None) + .await + } + + async fn on_demand_migration_backfill_status(&self, bucket: &str) -> Result { + self.on_demand_migration_json(Method::GET, bucket, "/backfill", &[], None) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn status_codes_map_to_the_documented_exit_classes() { + let cases: &[(u16, &str, i32)] = &[ + ( + 400, + r#"{"Code":"InvalidArgument","Message":"bad endpoint"}"#, + 2, + ), + ( + 400, + r#"{"Code":"OnDemandMigrationDisabled","Message":"off"}"#, + 2, + ), + ( + 400, + r#"{"Code":"OnDemandMigrationSourceUnreachable","Message":"connect"}"#, + 3, + ), + (401, "", 4), + (403, r#"{"Code":"AccessDenied","Message":"licence"}"#, 4), + ( + 404, + r#"{"Code":"NoSuchConfiguration","Message":"unset"}"#, + 5, + ), + (404, r#"{"Code":"NoSuchBucket","Message":"missing"}"#, 5), + ( + 404, + r#"{"Code":"NoSuchBackfillJob","Message":"never ran"}"#, + 5, + ), + (404, "", 7), + (404, "not found", 7), + ( + 409, + r#"{"Code":"OnDemandMigrationBackfillRunning","Message":"busy"}"#, + 6, + ), + (501, r#"{"Code":"OnDemandMigrationBackendNotCompiled"}"#, 7), + (429, "", 3), + (503, "", 3), + (418, "", 1), + ]; + for (status, body, exit_code) in cases { + let error = map_on_demand_migration_error(StatusCode::from_u16(*status).unwrap(), body); + assert_eq!(error.exit_code(), *exit_code, "{status} {body}"); + } + } + + #[test] + fn route_absence_uses_the_fixed_unsupported_message() { + let error = map_on_demand_migration_error(StatusCode::NOT_FOUND, ""); + assert_eq!( + error.to_string(), + format!("Unsupported feature: {UNSUPPORTED_MESSAGE}") + ); + } + + #[test] + fn server_messages_are_bounded_and_de_controlled() { + let body = format!( + r#"{{"Code":"InvalidArgument","Message":"line\u001b[31mred\n{}"}}"#, + "x".repeat(2000) + ); + let text = map_on_demand_migration_error(StatusCode::BAD_REQUEST, &body).to_string(); + assert!(!text.contains('\u{1b}')); + assert!(!text.contains('\n')); + assert!(text.len() < 700); + assert!(text.contains("HTTP 400 InvalidArgument")); + } +} + +#[cfg(test)] +mod transport_tests; diff --git a/crates/s3/src/admin/on_demand_migration/transport_tests.rs b/crates/s3/src/admin/on_demand_migration/transport_tests.rs new file mode 100644 index 00000000..23dd7b04 --- /dev/null +++ b/crates/s3/src/admin/on_demand_migration/transport_tests.rs @@ -0,0 +1,327 @@ +//! Wire-level tests against a local HTTP server: signing, routes, query +//! encoding, the plaintext body, and the fixture-pinned responses. + +use super::*; +use rc_core::Alias; +use rc_core::admin::{ + PathStyle, SkipExisting, SourceCredentialsRequest, SourceProvider, SourceRequest, TlsRequest, +}; +use serde_json::Value; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +const SET_REQUEST: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/set_request.json"); +const SET_RESPONSE: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/set_response.json"); +const GET_RESPONSE: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/get_response.json"); +const STATUS: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/status.json"); +const BACKFILL_JOB: &str = + include_str!("../../../../core/tests/fixtures/on_demand_migration/backfill_job.json"); + +struct Captured { + method: String, + target: String, + headers: String, + body: Vec, +} + +async fn server( + responses: Vec<(u16, String)>, +) -> (AdminClient, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let mut requests = Vec::new(); + for (status, body) in responses { + let (mut stream, _) = + tokio::time::timeout(std::time::Duration::from_secs(10), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut bytes = Vec::new(); + let header_end = loop { + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).await.unwrap(); + assert!(n > 0); + bytes.extend_from_slice(&buf[..n]); + if let Some(end) = bytes.windows(4).position(|v| v == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&bytes[..end]); + let length = headers + .lines() + .find_map(|l| { + l.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|s| s.trim().parse::().unwrap()) + }) + .unwrap_or(0); + if bytes.len() >= end + 4 + length { + break end + 4; + } + } + }; + let headers = String::from_utf8_lossy(&bytes[..header_end]).into_owned(); + let mut request_line = headers.lines().next().unwrap().split_whitespace(); + requests.push(Captured { + method: request_line.next().unwrap().to_string(), + target: request_line.next().unwrap().to_string(), + headers: headers.clone(), + body: bytes[header_end..].to_vec(), + }); + let payload = if status == 204 { String::new() } else { body }; + stream + .write_all( + format!( + "HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + payload.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + } + requests + }); + let mut client = + AdminClient::new(&Alias::new("a", &endpoint, "test-access", "test-secret")).unwrap(); + client.http_client = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(std::time::Duration::from_secs(5)) + .build() + .unwrap(); + (client, task) +} + +fn fixture_request() -> OnDemandMigrationConfigRequest { + let mut request = OnDemandMigrationConfigRequest::new(SourceRequest { + provider: SourceProvider::Minio, + endpoint: Some("https://source.example.com:9000".into()), + region: "us-east-1".into(), + bucket: "legacy-photos".into(), + path_style: PathStyle::Auto, + credentials: Some(SourceCredentialsRequest { + access_key: "AKIASOURCE".into(), + secret_key: Zeroizing::new("sourceSecretKey123".into()), + session_token: None, + }), + tls: TlsRequest::default(), + }); + request.filter.source_prefix = Some("photos/".into()); + request +} + +#[tokio::test] +async fn set_signs_a_put_with_the_fixture_body_and_dry_run_query() { + let (client, server) = server(vec![(200, SET_RESPONSE.to_string())]).await; + let result = client + .set_on_demand_migration("photos", &fixture_request(), true) + .await + .unwrap(); + assert_eq!(result.bucket, "photos"); + assert!(result.probe.unwrap().reachable); + let requests = server.await.unwrap(); + let request = &requests[0]; + assert_eq!(request.method, "PUT"); + assert_eq!( + request.target, + "/rustfs/admin/v3/on-demand-migration/photos?dry-run=true" + ); + let lower = request.headers.to_ascii_lowercase(); + assert!(lower.contains("authorization: aws4-hmac-sha256")); + assert!(lower.contains("content-type: application/json")); + let body: Value = serde_json::from_slice(&request.body).unwrap(); + let expected: Value = serde_json::from_str(SET_REQUEST).unwrap(); + assert_eq!(body, expected); +} + +#[tokio::test] +async fn set_without_dry_run_has_no_query_and_encodes_the_bucket() { + let (client, server) = server(vec![(200, SET_RESPONSE.to_string())]).await; + client + .set_on_demand_migration("my.bucket", &fixture_request(), false) + .await + .unwrap(); + let requests = server.await.unwrap(); + assert_eq!( + requests[0].target, + "/rustfs/admin/v3/on-demand-migration/my.bucket" + ); +} + +#[tokio::test] +async fn get_and_status_use_their_routes_and_parse_fixtures() { + let (client, server) = server(vec![ + (200, GET_RESPONSE.to_string()), + (200, STATUS.to_string()), + ]) + .await; + let config = client.get_on_demand_migration("photos").await.unwrap(); + assert_eq!(config.updated_at.as_deref(), Some("2026-09-02T10:00:00Z")); + let status = client.on_demand_migration_status("photos").await.unwrap(); + assert_eq!(status.served_by_source_ratio, None); + assert_eq!(status.queue_depth, 1); + let requests = server.await.unwrap(); + assert_eq!(requests[0].method, "GET"); + assert_eq!( + requests[0].target, + "/rustfs/admin/v3/on-demand-migration/photos" + ); + assert_eq!( + requests[1].target, + "/rustfs/admin/v3/on-demand-migration/photos/status" + ); +} + +#[tokio::test] +async fn delete_accepts_no_content() { + let (client, server) = server(vec![(204, String::new())]).await; + client.delete_on_demand_migration("photos").await.unwrap(); + let requests = server.await.unwrap(); + assert_eq!(requests[0].method, "DELETE"); + assert!(requests[0].body.is_empty()); +} + +#[tokio::test] +async fn backfill_routes_carry_the_op_selector_and_optional_body() { + let (client, server) = server(vec![ + (200, BACKFILL_JOB.to_string()), + (200, BACKFILL_JOB.to_string()), + (200, BACKFILL_JOB.to_string()), + ]) + .await; + let request = BackfillStartRequest { + prefix: Some("photos/".into()), + skip_existing: Some(SkipExisting::EtagOrSize), + dry_run: true, + }; + let started = client + .start_on_demand_migration_backfill("photos", &request) + .await + .unwrap(); + assert_eq!(started.job.unwrap().state, "running"); + client + .cancel_on_demand_migration_backfill("photos") + .await + .unwrap(); + client + .on_demand_migration_backfill_status("photos") + .await + .unwrap(); + let requests = server.await.unwrap(); + assert_eq!(requests[0].method, "POST"); + assert_eq!( + requests[0].target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill?op=start" + ); + let body: Value = serde_json::from_slice(&requests[0].body).unwrap(); + assert_eq!( + body, + serde_json::json!({"prefix":"photos/","skip_existing":"etag_or_size","dry_run":true}) + ); + assert_eq!(requests[1].method, "POST"); + assert_eq!( + requests[1].target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill?op=cancel" + ); + assert!(requests[1].body.is_empty()); + assert_eq!(requests[2].method, "GET"); + assert_eq!( + requests[2].target, + "/rustfs/admin/v3/on-demand-migration/photos/backfill" + ); +} + +#[tokio::test] +async fn route_absence_is_reported_as_unsupported_not_as_not_found() { + let (client, server) = server(vec![(404, String::new())]).await; + let error = client.get_on_demand_migration("photos").await.unwrap_err(); + assert_eq!(error.exit_code(), 7); + assert!(error.to_string().contains(UNSUPPORTED_MESSAGE)); + server.await.unwrap(); +} + +#[tokio::test] +async fn unset_configuration_is_not_found() { + let (client, server) = server(vec![( + 404, + r#"{"Code":"NoSuchConfiguration","Message":"on-demand migration is not configured for bucket photos"}"#.into(), + )]) + .await; + let error = client.get_on_demand_migration("photos").await.unwrap_err(); + assert_eq!(error.exit_code(), 5); + assert!(error.to_string().contains("NoSuchConfiguration")); + server.await.unwrap(); +} + +#[tokio::test] +async fn backfill_conflict_and_source_unreachable_keep_their_classes() { + let (client, server) = server(vec![ + ( + 409, + r#"{"Code":"OnDemandMigrationBackfillRunning","Message":"a backfill job is already running"}"#.into(), + ), + ( + 400, + r#"{"Code":"OnDemandMigrationSourceUnreachable","Message":"connect"}"#.into(), + ), + (403, r#"{"Code":"AccessDenied","Message":"licence"}"#.into()), + ]) + .await; + let conflict = client + .start_on_demand_migration_backfill("photos", &BackfillStartRequest::default()) + .await + .unwrap_err(); + assert_eq!(conflict.exit_code(), 6); + let unreachable = client + .set_on_demand_migration("photos", &fixture_request(), true) + .await + .unwrap_err(); + assert_eq!(unreachable.exit_code(), 3); + let licence = client + .set_on_demand_migration("photos", &fixture_request(), false) + .await + .unwrap_err(); + assert_eq!(licence.exit_code(), 4); + server.await.unwrap(); +} + +#[tokio::test] +async fn error_bodies_never_echo_the_admin_credentials() { + let (client, server) = server(vec![( + 400, + r#"{"Code":"InvalidArgument","Message":"signed with test-secret by test-access"}"#.into(), + )]) + .await; + let error = client.get_on_demand_migration("photos").await.unwrap_err(); + let text = error.to_string(); + assert!(!text.contains("test-secret")); + assert!(!text.contains("test-access")); + assert!(text.contains("[REDACTED]")); + server.await.unwrap(); +} + +#[tokio::test] +async fn invalid_bucket_or_config_never_reaches_the_network() { + let (client, server) = server(vec![]).await; + assert_eq!( + client + .get_on_demand_migration("a/b") + .await + .unwrap_err() + .exit_code(), + 2 + ); + let mut request = fixture_request(); + request.source.endpoint = None; + assert_eq!( + client + .set_on_demand_migration("photos", &request, false) + .await + .unwrap_err() + .exit_code(), + 2 + ); + assert!(server.await.unwrap().is_empty()); +} diff --git a/docs/reference/rc/admin.md b/docs/reference/rc/admin.md index 03800e3b..6d52ab75 100644 --- a/docs/reference/rc/admin.md +++ b/docs/reference/rc/admin.md @@ -2,7 +2,7 @@ ## Purpose -The `rc admin` operation manages the RustFS Admin API, including scanner and storage diagnostics, bounded realtime metrics, KMS inspection and key lifecycle management, cluster information, healing, pools, expansion, decommissioning, rebalance workflows, IAM users, policies, groups, service accounts, site replication, and service control. +The `rc admin` operation manages the RustFS Admin API, including scanner and storage diagnostics, bounded realtime metrics, KMS inspection and key lifecycle management, cluster information, healing, pools, expansion, decommissioning, rebalance workflows, IAM users, policies, groups, service accounts, per-bucket on-demand migration, site replication, and service control. `rc admin` does not implement the MinIO Admin API. MinIO aliases remain available to S3 data commands, but MinIO administrative operations require a MinIO-compatible admin client. @@ -62,6 +62,10 @@ rc admin diagnostics client-devnull [--size ] [--timeout ... rc admin config module-switch ... rc admin bucket-metadata ... +rc admin bucket migration set / --provider [--endpoint URL] --region --source-bucket [OPTIONS] [--dry-run] +rc admin bucket migration / [--watch [--interval SECONDS]] +rc admin bucket migration backfill start / [--prefix P] [--skip-existing always|etag_or_size] [--dry-run] +rc admin bucket migration backfill / [--watch [--interval SECONDS]] rc admin replicate add [...] rc admin replicate [OPTIONS] rc admin replicate edit --site [EDIT OPTIONS] --yes @@ -95,6 +99,7 @@ rc admin replicate remove <--all|--site > | `service` | Control the server process: restart, stop, freeze, unfreeze. | | `config` | Inspect, plan, export, and mutate RustFS server configuration. | | `bucket-metadata` | Export or import validated per-bucket configuration archives. | +| `bucket migration` | Configure, inspect, and backfill On-Demand Migration from an external S3-compatible source bucket. | | `replicate` | Manage site replication across clusters. | ## Account and Two-Factor Workflow @@ -772,6 +777,49 @@ Server exports redact replication-target credentials. The client never prints ar An import is one bounded PUT and is never automatically retried. A transport failure or server error can mean that only part of the archive was applied; inspect every selected bucket before deciding whether to retry. Successful JSON output uses output schema v3 with one `admin_operations` result per selected bucket. +## On-Demand Migration Workflow + +`rc admin bucket migration` manages RustFS On-Demand Migration: a bucket names an external S3-compatible source bucket, a GET that misses locally is served from that source and stored locally in the same pass, and a background backfill job pulls the rest. It is the RustFS equivalent of Cloudflare R2 Sippy or Tigris shadow buckets. The server-side operations guide is `docs/operations/on-demand-migration.md` in `rustfs/rustfs`; the wire contract is pinned by the fixtures vendored under `crates/core/tests/fixtures/on_demand_migration/`. + +Every command takes the local bucket as `/`. + +| Command | Description | +| --- | --- | +| `rc admin bucket migration set / --provider

[--endpoint URL] --region --source-bucket [--prefix P] [--source-prefix SP] [--access-key AK [--secret-key SK] \| --public] [--path-style auto\|path\|virtual] [--skip-tls-verify] [--ca-cert FILE] [--head proxy\|local_only] [--range-get serve_and_backfill\|serve_only] [--source-error propagate\|not_found] [--no-preserve-etag] [--copy-tags] [--no-events] [--inline-max-bytes N] [--max-concurrent-pulls N] [--dry-run]` | Validate the configuration, probe the source (`HeadBucket` plus a one-key listing) and save it. `--dry-run` sends `PUT ...?dry-run=true`, which validates and probes without saving. | +| `rc admin bucket migration get /` | Print the saved configuration as a table. Credentials are shown as `REDACTED`. | +| `rc admin bucket migration rm /` | Remove the configuration. Idempotent; objects already pulled stay in place. | +| `rc admin bucket migration status / [--watch] [--interval SECONDS]` | Print the answering node's runtime status: source-hit ratio, migrated bytes, in-flight and queued pulls, breaker state, request and failure counters, and the last source error. | +| `rc admin bucket migration backfill start / [--prefix P] [--skip-existing always\|etag_or_size] [--dry-run]` | Start the background job that walks the source listing and pulls what is missing locally. `--dry-run` lists and counts without queuing anything. | +| `rc admin bucket migration backfill cancel /` | Ask the running job to stop at its next checkpoint. | +| `rc admin bucket migration backfill status / [--watch] [--interval SECONDS]` | Print the job checkpoint. With `--watch`, refresh one progress line every `--interval` seconds (default 2) until the job reaches a terminal state, then print the final checkpoint. | + +### Secret handling + +`set` needs the source secret key whenever `--access-key` is given. It is taken, in order, from `--secret-key`, from the `RC_ODM_SECRET_KEY` environment variable, or from a hidden terminal prompt. Prefer the variable or the prompt: a flag value lands in shell history and in `ps` output. The prompt is only offered when standard input is a terminal and output is human-readable; with `--json` or in a script the variable is required and a missing one is a usage error before any request is sent. + +The server returns every credential as the placeholder `REDACTED`, and `set` replaces the configuration wholesale rather than merging into it. `rc` therefore refuses the placeholder as a secret: editing an existing configuration means passing the real secret again. `--public` configures anonymous access and is mutually exclusive with the credential flags. + +`--endpoint` must be `scheme://host[:port]` with no path, query, fragment or embedded userinfo; it is optional only for `--provider aws`, where the server derives it from `--region`. `--ca-cert` reads a PEM bundle of at most 64 KiB. Arguments are validated before the secret is read, so a typo never costs a prompt. + +### Output + +Human output prints one aligned key/value table per document. `status` renders `served_by_source_ratio` exactly as the server reports it: a percentage when present and an em dash (`—`) when the server returns `null`, never zero, because a missing ratio and a zero ratio mean different things. `--json` wraps every result in output schema v3 with `type: on_demand_migration` and `data: {operation, bucket, result}`, where `result` is the server document with credentials redacted. `--watch` with `--json` emits one compact record per refresh on stdout; without `--json`, the backfill progress line is written to stderr so stdout carries only the final document. + +Every response field is optional on the client with the server default, so an older server that omits a field still parses, and unknown fields from a newer server are ignored. + +### Exit codes + +| Condition | Code | +|---|---| +| Malformed target, missing or conflicting flags, invalid endpoint or CA file, configuration rejected by the server (`InvalidArgument`), module switch off (`OnDemandMigrationDisabled`) | usage (2) | +| Source unreachable during the probe (`OnDemandMigrationSourceUnreachable`), transport failure, 5xx | network (3) | +| Not authorized, or the licence denies the entitlement (`AccessDenied`) | authentication (4) | +| No configuration (`NoSuchConfiguration`), no such bucket, no backfill job recorded | not found (5) | +| A backfill job already holds the lease (409 `OnDemandMigrationBackfillRunning`) | conflict (6) | +| Route family absent (the server predates on-demand migration), or the provider was excluded at build time (501) | unsupported (7) | + +A 404 that does not carry one of the route family's own error codes means the whole feature is missing from the server; `rc` prints `server does not support on-demand migration` and exits 7 rather than treating the bucket as unconfigured. Writes are never automatically retried: a `set` probes the source and a backfill start takes a lease, so a repeated request is a second decision. + ## Site Replication Workflow `rc admin replicate` manages multi-cluster site replication. Peer sites are given as configured alias names; their endpoints and credentials are resolved from the local alias store, so every participating site needs an alias with root credentials before running `add`. diff --git a/schemas/output_v3.json b/schemas/output_v3.json index d24710fa..91ada115 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -26,6 +26,7 @@ "enum": [ "capabilities", "table_catalog", + "on_demand_migration", "versioned_objects", "locks", "multipart_uploads", @@ -67,6 +68,18 @@ "generated_at": { "$ref": "#/definitions/nullableTimestamp" } } }, + "onDemandMigrationSuccessData": { + "type": "object", + "required": ["operation", "bucket", "result"], + "properties": { + "operation": { + "type": "string", + "enum": ["set", "get", "remove", "status", "backfill_start", "backfill_cancel", "backfill_status"] + }, + "bucket": { "type": "string" }, + "result": { "type": "object" } + } + }, "successEnvelope": { "type": "object", "required": ["schema_version", "type", "status", "data"], @@ -2053,6 +2066,17 @@ { "properties": { "type": { "const": "table_catalog" }, "data": { "type": "object" } } } ] }, + { + "allOf": [ + { "$ref": "#/definitions/successEnvelope" }, + { + "properties": { + "type": { "const": "on_demand_migration" }, + "data": { "$ref": "#/definitions/onDemandMigrationSuccessData" } + } + } + ] + }, { "$ref": "#/definitions/errorOutput" } ] }