From a5bde8b0af22d17111b98adb9fabe1dbd45bde26 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 3 Sep 2026 01:58:49 +0800 Subject: [PATCH] feat(admin): add on-demand migration bucket admin API (#7076) * feat(ecstore): add on-demand migration bucket config model Introduce OnDemandMigrationConfig (deny_unknown_fields, version 1) with typed validation, credential redaction, a secret-free Debug impl, and the OnceLock publish hook the runtime registers into. Exported through the api facade. * feat(ecstore): persist on-demand migration config in bucket metadata Store the config as a RustFS extension entry (on-demand-migration.json) with its update time in .metadata.bin, add the typed BucketMetadataSys accessor, and publish the config through the hook on every cache-install path alongside the durability sync. * refactor(ecstore): extract shared remote S3 client builder Move the aws_sdk_s3 client construction out of bucket_target_sys into bucket/remote_s3_client.rs: endpoint assembly, credential provider, path-style selection, custom CA / skip-TLS transports and the outbound SSRF gate now build from a neutral RemoteS3EndpointSpec so replication targets and the upcoming on-demand migration source client share one policy. Replication builds its client through From<&BucketTarget>; the gate keeps its relaxed semantics (private allowed, loopback only behind RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET) verbatim. The builder also gains optional connect/read timeouts and a User-Agent suffix interceptor, both unset for replication. Refs rustfs/backlog#2149 * feat(ecstore): add on-demand migration SourceClient Add bucket/on_demand_migration/source_client.rs on top of the shared remote S3 builder: HEAD, ranged streaming GET, ListObjectsV2 with source-prefix mapping, GetObjectTagging and an admin probe. Every request carries the x-rustfs-/x-minio-source-proxy-request anti-loop markers and a RustFS-OnDemandMigration/ User-Agent suffix; SSE-C source objects are rejected as unsupported. SourceError classifies SDK failures (not found, access denied, throttled, timeout, connect, server error) with retryability and a stable metrics label. Debug output redacts credentials. Refs rustfs/backlog#2149 * docs(operations): point outbound policy at shared remote S3 client builder * chore: integrate ODM-01 and ODM-02 as B1 base (fix facade merge) * feat(admin): add on-demand migration bucket admin API Add the management plane for On-Demand Migration (ODM-07, rustfs/backlog#2154): PUT/GET/DELETE /v3/on-demand-migration/{bucket}, PUT ?dry-run=true, and a GET .../status skeleton. - PUT authorizes SetBucketOnDemandMigration, checks the bucket, the RUSTFS_ON_DEMAND_MIGRATION_ENABLED switch and the license, validates the ODM-01 config against local endpoints and replication targets, probes the source with SourceClient::probe(), then persists through the incarnation gate and asks peers to reload. Responses carry the redacted config and a probe summary; probe failures name only the error class. - GET answers 404 NoSuchConfiguration when unset; DELETE is idempotent (204). - New AdminAction variants admin:SetBucketOnDemandMigration and admin:GetBucketOnDemandMigration, route policy matrix rows, registration and MinIO alias coverage, and a doc row for the extra handler gates. - rustfs-madmin gains on_demand_migration wire types and client methods; golden fixtures under crates/madmin/fixtures/on_demand_migration/ are asserted byte-for-byte by both the handler and the client tests. Anonymous sources still map to a 400 naming source.credentials until the runtime slice adds the credential-less path. * refactor(admin): route on-demand migration handler errors through the s3 facade --- .../on_demand_migration/get_response.json | 1 + .../on_demand_migration/set_request.json | 1 + .../on_demand_migration/set_response.json | 1 + .../fixtures/on_demand_migration/status.json | 1 + crates/madmin/src/client.rs | 154 +-- crates/madmin/src/lib.rs | 5 + crates/madmin/src/on_demand_migration.rs | 535 +++++++ crates/madmin/src/test_support.rs | 154 +++ crates/policy/src/policy/action.rs | 28 + .../admin-route-action-snapshot.md | 11 + rustfs/src/admin/handlers/mod.rs | 5 + .../src/admin/handlers/on_demand_migration.rs | 1232 +++++++++++++++++ rustfs/src/admin/mod.rs | 7 +- rustfs/src/admin/route_policy.rs | 57 + rustfs/src/admin/route_registration_test.rs | 16 + rustfs/src/admin/storage_api.rs | 38 +- rustfs/src/storage/storage_api.rs | 2 +- 17 files changed, 2106 insertions(+), 142 deletions(-) create mode 100644 crates/madmin/fixtures/on_demand_migration/get_response.json create mode 100644 crates/madmin/fixtures/on_demand_migration/set_request.json create mode 100644 crates/madmin/fixtures/on_demand_migration/set_response.json create mode 100644 crates/madmin/fixtures/on_demand_migration/status.json create mode 100644 crates/madmin/src/on_demand_migration.rs create mode 100644 crates/madmin/src/test_support.rs create mode 100644 rustfs/src/admin/handlers/on_demand_migration.rs diff --git a/crates/madmin/fixtures/on_demand_migration/get_response.json b/crates/madmin/fixtures/on_demand_migration/get_response.json new file mode 100644 index 000000000..7aed6ae57 --- /dev/null +++ b/crates/madmin/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","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/madmin/fixtures/on_demand_migration/set_request.json b/crates/madmin/fixtures/on_demand_migration/set_request.json new file mode 100644 index 000000000..30fefe38a --- /dev/null +++ b/crates/madmin/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","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/madmin/fixtures/on_demand_migration/set_response.json b/crates/madmin/fixtures/on_demand_migration/set_response.json new file mode 100644 index 000000000..cae01f223 --- /dev/null +++ b/crates/madmin/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","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/madmin/fixtures/on_demand_migration/status.json b/crates/madmin/fixtures/on_demand_migration/status.json new file mode 100644 index 000000000..848abd355 --- /dev/null +++ b/crates/madmin/fixtures/on_demand_migration/status.json @@ -0,0 +1 @@ +{"configured":true,"enabled":true,"module_enabled":false} diff --git a/crates/madmin/src/client.rs b/crates/madmin/src/client.rs index 4f7c6fb66..6662946b9 100644 --- a/crates/madmin/src/client.rs +++ b/crates/madmin/src/client.rs @@ -412,7 +412,7 @@ impl AdminClient { self.execute(request).await } - fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result { + pub(crate) fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result { let mut url = self .endpoint .join(&format!("{}{}", self.api_prefix.trim_end_matches('/'), path)) @@ -430,7 +430,7 @@ impl AdminClient { /// then hand the signed headers to the HTTP client. The signature covers /// method, path, query, and an unsigned-payload marker — the same shape /// RustFS itself sends for peer admin calls. - async fn sign_and_build( + pub(crate) async fn sign_and_build( &self, method: Method, url: reqwest::Url, @@ -479,7 +479,7 @@ impl AdminClient { Ok(request) } - async fn execute Deserialize<'de>>(&self, request: reqwest::Request) -> Result { + pub(crate) async fn execute Deserialize<'de>>(&self, request: reqwest::Request) -> Result { let response = self.http.execute(request).await?; let status = response.status(); let bytes = response.bytes().await?; @@ -493,6 +493,20 @@ impl AdminClient { message: err.to_string(), }) } + + /// Execute a request whose success answer carries no body (`204`). + pub(crate) async fn execute_no_content(&self, request: reqwest::Request) -> Result<(), AdminClientError> { + let response = self.http.execute(request).await?; + let status = response.status(); + if !status.is_success() { + let bytes = response.bytes().await?; + return Err(AdminClientError::HttpStatus { + status: status.as_u16(), + body: String::from_utf8_lossy(&bytes).into_owned(), + }); + } + Ok(()) + } } /// Response of [`AdminClient::heal_stop`]: cancelling a single tokened task @@ -518,7 +532,7 @@ fn heal_path(bucket: Option<&str>, prefix: Option<&str>) -> String { /// Encode a single path segment (slashes are content, not separators, inside /// bucket/prefix path params). -fn percent_encode_path_segment(segment: &str) -> String { +pub(crate) fn percent_encode_path_segment(segment: &str) -> String { let mut out = String::with_capacity(segment.len()); for byte in segment.bytes() { match byte { @@ -535,8 +549,8 @@ mod tests { AdminClient, AdminClientError, BackgroundHealStatus, HealOpts, HealScanMode, HealStartSuccess, HealTaskStatus, ScannerStatus, heal_path, percent_encode_path_segment, }; + use crate::test_support::TestServer; use serde_json::json; - use std::sync::{Arc, Mutex}; #[test] fn heal_paths_cover_root_bucket_and_prefix() { @@ -734,134 +748,4 @@ mod tests { let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); assert!(matches!(client.scanner_status().await.unwrap_err(), AdminClientError::Decode { .. })); } - - /// One recorded request, parsed off the wire with the minimum needed for - /// assertions: method, path, query, headers, body. - #[derive(Debug, Clone)] - struct RecordedRequest { - method: String, - path: String, - query: String, - headers: Vec<(String, String)>, - body: String, - } - - impl RecordedRequest { - fn header(&self, name: &str) -> Option { - self.headers - .iter() - .find(|(key, _)| key.eq_ignore_ascii_case(name)) - .map(|(_, value)| value.clone()) - } - } - - /// Minimal HTTP/1.1 server: one canned response per connection, every - /// request recorded behind an `Arc`. Deliberately dependency-free — - /// the assertions only need the raw request bytes. - struct TestServer { - addr: std::net::SocketAddr, - requests: Arc>>, - } - - impl TestServer { - async fn spawn(response_body: &'static str, status: u16) -> Self { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind ephemeral port"); - let addr = listener.local_addr().expect("local addr"); - let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); - - let recorded = requests.clone(); - tokio::spawn(async move { - let reason = if status == 200 { "OK" } else { "Forbidden" }; - let response = format!( - "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response_body}", - response_body.len() - ); - // Each request is a fresh connection (connection: close); a - // bounded loop serves every call a test makes while letting - // the task exit instead of lingering for the whole process. - for _ in 0..16 { - let Ok((mut stream, _)) = listener.accept().await else { - break; - }; - let mut buffer = Vec::with_capacity(2048); - let mut chunk = [0u8; 2048]; - // Read headers plus content-length body, or stop on close. - loop { - if let Some(end) = find_header_end(&buffer) { - let content_length = extract_content_length(&buffer[..end]); - if buffer.len() >= end + content_length { - break; - } - } - let n = match stream.read(&mut chunk).await { - Ok(0) | Err(_) => break, - Ok(n) => n, - }; - buffer.extend_from_slice(&chunk[..n]); - if buffer.len() > 64 * 1024 { - break; - } - } - if let Some(request) = parse_request(&buffer) { - recorded.lock().expect("recorded lock").push(request); - } - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.shutdown().await; - } - }); - - Self { addr, requests } - } - - fn recorded(&self) -> RecordedRequest { - self.requests - .lock() - .expect("recorded lock") - .last() - .cloned() - .expect("the client call must have produced one recorded request") - } - } - - fn find_header_end(buffer: &[u8]) -> Option { - buffer.windows(4).position(|window| window == b"\r\n\r\n").map(|pos| pos + 4) - } - - fn extract_content_length(headers: &[u8]) -> usize { - let text = String::from_utf8_lossy(headers).to_ascii_lowercase(); - text.lines() - .find_map(|line| line.strip_prefix("content-length:")) - .and_then(|value| value.trim().parse().ok()) - .unwrap_or(0) - } - - fn parse_request(raw: &[u8]) -> Option { - let end = find_header_end(raw)?; - let head = String::from_utf8_lossy(&raw[..end]); - let body = String::from_utf8_lossy(&raw[end..]).into_owned(); - let mut lines = head.lines(); - let request_line = lines.next()?; - let mut parts = request_line.split_whitespace(); - let method = parts.next()?.to_string(); - let target = parts.next()?.to_string(); - let (path, query) = match target.split_once('?') { - Some((path, query)) => (path.to_string(), query.to_string()), - None => (target, String::new()), - }; - let headers = lines - .filter_map(|line| line.split_once(':')) - .map(|(name, value)| (name.trim().to_string(), value.trim().to_string())) - .collect(); - Some(RecordedRequest { - method, - path, - query, - headers, - body, - }) - } } diff --git a/crates/madmin/src/lib.rs b/crates/madmin/src/lib.rs index d4a4dfc6a..d66514418 100644 --- a/crates/madmin/src/lib.rs +++ b/crates/madmin/src/lib.rs @@ -20,6 +20,7 @@ pub mod health; pub mod info_commands; pub mod metrics; pub mod net; +pub mod on_demand_migration; pub mod policy; pub mod service_commands; pub mod site_replication; @@ -27,10 +28,14 @@ pub mod trace; pub mod user; pub mod utils; +#[cfg(test)] +pub(crate) mod test_support; + pub use account::*; pub use client::*; pub use group::*; pub use info_commands::*; +pub use on_demand_migration::*; pub use policy::*; pub use site_replication::*; pub use user::*; diff --git a/crates/madmin/src/on_demand_migration.rs b/crates/madmin/src/on_demand_migration.rs new file mode 100644 index 000000000..ffbf60ee9 --- /dev/null +++ b/crates/madmin/src/on_demand_migration.rs @@ -0,0 +1,535 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! On-Demand Migration admin API contract (ODM-07, rustfs/backlog#2154). +//! +//! Wire types for `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}` and +//! `GET .../status`, mirroring the server's config model +//! (`crates/ecstore/src/bucket/on_demand_migration/config.rs`) and handler +//! responses (`rustfs/src/admin/handlers/on_demand_migration.rs`). The SDK +//! owns its own copies, madmin-go style; the fixtures under +//! `fixtures/on_demand_migration/` are the contract both sides pin +//! byte-for-byte, so field order, defaults and `null` handling here must +//! match the server exactly. + +use crate::client::{AdminClient, AdminClientError, percent_encode_path_segment}; +use http::Method; +use serde::{Deserialize, Serialize}; +use std::fmt; + +/// Config schema version this client speaks. +pub const ON_DEMAND_MIGRATION_CONFIG_VERSION: u32 = 1; + +/// Query flag that validates and probes a config without saving it. +const DRY_RUN_QUERY: &str = "dry-run"; + +/// Bucket-level on-demand migration configuration (request body of the +/// `PUT`, redacted copy in every response). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationConfig { + #[serde(default = "default_version")] + pub version: u32, + #[serde(default = "default_true")] + pub enabled: bool, + pub source: OnDemandMigrationSource, + #[serde(default)] + pub filter: OnDemandMigrationFilter, + #[serde(default)] + pub policy: OnDemandMigrationPolicy, +} + +impl OnDemandMigrationConfig { + /// A config with the documented defaults for everything but the source. + pub fn new(source: OnDemandMigrationSource) -> Self { + Self { + version: ON_DEMAND_MIGRATION_CONFIG_VERSION, + enabled: true, + source, + filter: OnDemandMigrationFilter::default(), + policy: OnDemandMigrationPolicy::default(), + } + } +} + +/// The external S3-compatible source bucket. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSource { + pub provider: OnDemandMigrationProvider, + /// `http(s)://host[:port]`; optional only for `aws`, where it derives from `region`. + #[serde(default)] + pub endpoint: Option, + pub region: String, + pub bucket: String, + #[serde(default)] + pub path_style: OnDemandMigrationPathStyle, + /// `None` means anonymous access to a public source bucket. + #[serde(default)] + pub credentials: Option, + #[serde(default)] + pub tls: OnDemandMigrationTls, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OnDemandMigrationProvider { + S3, + Aws, + Minio, + Rustfs, + R2, + Gcs, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OnDemandMigrationPathStyle { + #[default] + Auto, + Path, + Virtual, +} + +/// Static source credentials. `Debug` never prints the secret or the +/// session token; responses carry them as `REDACTED`. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationCredentials { + pub access_key: String, + pub secret_key: String, + #[serde(default)] + pub session_token: Option, +} + +impl fmt::Debug for OnDemandMigrationCredentials { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OnDemandMigrationCredentials") + .field("access_key", &self.access_key) + .field("secret_key", &"REDACTED") + .field("session_token", &self.session_token.as_ref().map(|_| "REDACTED")) + .finish() + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct OnDemandMigrationTls { + #[serde(default)] + pub skip_verify: bool, + #[serde(default)] + pub ca_cert_pem: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct OnDemandMigrationFilter { + #[serde(default)] + pub prefix: Option, + #[serde(default)] + pub source_prefix: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OnDemandMigrationHeadPolicy { + #[default] + Proxy, + LocalOnly, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OnDemandMigrationRangeGetPolicy { + #[default] + ServeAndBackfill, + ServeOnly, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OnDemandMigrationSourceErrorPolicy { + #[default] + Propagate, + NotFound, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSourceTimeout { + #[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 OnDemandMigrationSourceTimeout { + fn default() -> Self { + Self { + connect_ms: default_connect_ms(), + first_byte_ms: default_first_byte_ms(), + idle_ms: default_idle_ms(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationPolicy { + #[serde(default)] + pub head: OnDemandMigrationHeadPolicy, + #[serde(default)] + pub range_get: OnDemandMigrationRangeGetPolicy, + #[serde(default)] + pub source_error: OnDemandMigrationSourceErrorPolicy, + #[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: OnDemandMigrationSourceTimeout, + #[serde(default)] + pub bandwidth_limit_bytes_per_sec: Option, +} + +impl Default for OnDemandMigrationPolicy { + fn default() -> Self { + Self { + head: OnDemandMigrationHeadPolicy::default(), + range_get: OnDemandMigrationRangeGetPolicy::default(), + source_error: OnDemandMigrationSourceErrorPolicy::default(), + 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: OnDemandMigrationSourceTimeout::default(), + bandwidth_limit_bytes_per_sec: None, + } + } +} + +const MIB: u64 = 1024 * 1024; + +fn default_version() -> u32 { + ON_DEMAND_MIGRATION_CONFIG_VERSION +} +fn default_true() -> bool { + true +} +fn default_negative_cache_ttl_secs() -> u64 { + 30 +} +fn default_inline_max_bytes() -> u64 { + 16 * MIB +} +fn default_multipart_part_size_bytes() -> u64 { + 64 * MIB +} +fn default_max_concurrent_pulls() -> u32 { + 8 +} +fn default_pull_queue_capacity() -> u32 { + 1024 +} +fn default_connect_ms() -> u64 { + 5000 +} +fn default_first_byte_ms() -> u64 { + 15_000 +} +fn default_idle_ms() -> u64 { + 30_000 +} + +/// What the source answered during `PUT` validation. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationProbe { + pub reachable: bool, + pub listable: bool, + #[serde(default)] + pub sample_key: Option, +} + +/// `PUT` response: the redacted config plus the probe summary. `updated_at` +/// is `None` for a dry run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationSetResponse { + pub bucket: String, + pub dry_run: bool, + pub config: OnDemandMigrationConfig, + #[serde(default)] + pub updated_at: Option, + pub probe: OnDemandMigrationProbe, +} + +/// `GET` response: the redacted config and its RFC 3339 save time. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationGetResponse { + pub bucket: String, + pub config: OnDemandMigrationConfig, + pub updated_at: String, +} + +/// `GET .../status` response. Runtime counters are added by later ODM tasks. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationStatus { + pub configured: bool, + pub enabled: bool, + pub module_enabled: bool, +} + +fn config_path(bucket: &str) -> String { + format!("/v3/on-demand-migration/{}", percent_encode_path_segment(bucket)) +} + +impl AdminClient { + /// Configure the on-demand migration source of `bucket`. With `dry_run` + /// the server validates and probes the source but saves nothing. + pub async fn set_on_demand_migration( + &self, + bucket: &str, + config: &OnDemandMigrationConfig, + dry_run: bool, + ) -> Result { + let body = serde_json::to_vec(config).map_err(|err| AdminClientError::Decode { + message: err.to_string(), + })?; + let mut query = Vec::new(); + if dry_run { + query.push((DRY_RUN_QUERY, "true".to_string())); + } + let url = self.url_for(&config_path(bucket), &query)?; + let request = self.sign_and_build(Method::PUT, url, body, Some("application/json")).await?; + self.execute(request).await + } + + /// Read the (redacted) on-demand migration config of `bucket`. A bucket + /// without one answers HTTP 404 `NoSuchConfiguration`. + pub async fn get_on_demand_migration(&self, bucket: &str) -> Result { + self.get_json(&config_path(bucket)).await + } + + /// Clear the on-demand migration config of `bucket`; already-pulled + /// objects stay. Idempotent: a bucket without a config still answers 204. + pub async fn delete_on_demand_migration(&self, bucket: &str) -> Result<(), AdminClientError> { + let url = self.url_for(&config_path(bucket), &[])?; + let request = self.sign_and_build(Method::DELETE, url, Vec::new(), None).await?; + self.execute_no_content(request).await + } + + /// Read the on-demand migration status of `bucket`. + pub async fn on_demand_migration_status(&self, bucket: &str) -> Result { + self.get_json(&format!("{}/status", config_path(bucket))).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::TestServer; + + const SET_REQUEST_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/set_request.json"); + const SET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/set_response.json"); + const GET_RESPONSE_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/get_response.json"); + const STATUS_FIXTURE: &str = include_str!("../fixtures/on_demand_migration/status.json"); + + fn round_trip Deserialize<'de>>(fixture: &str) -> T { + let value: T = serde_json::from_str(fixture.trim()).expect("fixture decodes"); + let reserialized = serde_json::to_string(&value).expect("fixture re-encodes"); + assert_eq!( + reserialized, + fixture.trim(), + "client wire shape must reproduce the server fixture byte for byte" + ); + value + } + + #[test] + fn config_fixture_round_trips_byte_for_byte() { + let config: OnDemandMigrationConfig = round_trip(SET_REQUEST_FIXTURE); + assert_eq!(config.version, ON_DEMAND_MIGRATION_CONFIG_VERSION); + assert_eq!(config.source.provider, OnDemandMigrationProvider::Minio); + assert_eq!(config.source.path_style, OnDemandMigrationPathStyle::Auto); + assert_eq!(config.filter.source_prefix.as_deref(), Some("photos/")); + assert_eq!( + config.policy, + OnDemandMigrationPolicy::default(), + "fixture policy is the documented default" + ); + assert_eq!(config.source.tls, OnDemandMigrationTls::default()); + } + + #[test] + fn set_response_fixture_round_trips_and_is_redacted() { + let response: OnDemandMigrationSetResponse = round_trip(SET_RESPONSE_FIXTURE); + assert_eq!(response.bucket, "photos"); + assert!(!response.dry_run); + assert_eq!(response.updated_at.as_deref(), Some("2026-09-02T10:00:00Z")); + assert_eq!(response.probe.sample_key.as_deref(), Some("photos/2024/01.jpg")); + let credentials = response.config.source.credentials.expect("credentials present"); + assert_eq!(credentials.secret_key, "REDACTED"); + assert!(!format!("{credentials:?}").contains("sourceSecretKey123")); + } + + #[test] + fn get_response_and_status_fixtures_round_trip() { + let response: OnDemandMigrationGetResponse = round_trip(GET_RESPONSE_FIXTURE); + assert_eq!(response.updated_at, "2026-09-02T10:00:00Z"); + let status: OnDemandMigrationStatus = round_trip(STATUS_FIXTURE); + assert!(status.configured && status.enabled && !status.module_enabled); + } + + #[test] + fn minimal_config_expands_to_the_server_defaults() { + let config = OnDemandMigrationConfig::new(OnDemandMigrationSource { + provider: OnDemandMigrationProvider::Minio, + endpoint: Some("https://source.example.com:9000".to_string()), + region: "us-east-1".to_string(), + bucket: "legacy-photos".to_string(), + path_style: OnDemandMigrationPathStyle::Auto, + credentials: Some(OnDemandMigrationCredentials { + access_key: "AKIASOURCE".to_string(), + secret_key: "sourceSecretKey123".to_string(), + session_token: None, + }), + tls: OnDemandMigrationTls::default(), + }); + let mut expected: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).expect("fixture"); + expected.filter.source_prefix = None; + assert_eq!(config, expected); + + // A client-side minimal document decodes with the same defaults. + let minimal: OnDemandMigrationConfig = serde_json::from_str( + r#"{"source":{"provider":"s3","endpoint":"https://s.example","region":"us-east-1","bucket":"b"}}"#, + ) + .expect("minimal decodes"); + assert!(minimal.enabled); + assert_eq!(minimal.policy.max_concurrent_pulls, 8); + assert!(minimal.source.credentials.is_none()); + } + + #[test] + fn credentials_debug_never_prints_secrets() { + let credentials = OnDemandMigrationCredentials { + access_key: "AKIASOURCE".to_string(), + secret_key: "sourceSecretKey123".to_string(), + session_token: Some("token-value".to_string()), + }; + let rendered = format!("{credentials:?}"); + assert!(rendered.contains("AKIASOURCE")); + assert!(!rendered.contains("sourceSecretKey123")); + assert!(!rendered.contains("token-value")); + } + + #[tokio::test] + async fn set_sends_the_config_as_the_signed_put_body() { + let server = TestServer::spawn(SET_RESPONSE_FIXTURE, 200).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + let config: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).unwrap(); + + let response = client + .set_on_demand_migration("photos", &config, false) + .await + .expect("set decodes"); + assert_eq!(response.config.source.credentials.unwrap().secret_key, "REDACTED"); + + let request = server.recorded(); + assert_eq!(request.method, "PUT"); + assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos"); + assert_eq!(request.query, "", "dry-run must not be sent unless requested"); + assert_eq!(request.header("content-type").as_deref(), Some("application/json")); + assert!( + request + .header("authorization") + .is_some_and(|auth| auth.starts_with("AWS4-HMAC-SHA256")) + ); + assert_eq!(request.body, SET_REQUEST_FIXTURE.trim(), "the body is the canonical config document"); + } + + #[tokio::test] + async fn dry_run_adds_the_query_flag_and_tolerates_a_null_timestamp() { + let dry_run_body = SET_RESPONSE_FIXTURE + .trim() + .replace(r#""dry_run":false"#, r#""dry_run":true"#) + .replace(r#""updated_at":"2026-09-02T10:00:00Z""#, r#""updated_at":null"#); + let leaked: &'static str = Box::leak(dry_run_body.into_boxed_str()); + let server = TestServer::spawn(leaked, 200).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + let config: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).unwrap(); + + let response = client + .set_on_demand_migration("my bucket", &config, true) + .await + .expect("dry run decodes"); + assert!(response.dry_run); + assert_eq!(response.updated_at, None); + + let request = server.recorded(); + assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/my%20bucket"); + assert_eq!(request.query, "dry-run=true"); + } + + #[tokio::test] + async fn get_and_status_use_the_registered_routes() { + let server = TestServer::spawn(GET_RESPONSE_FIXTURE, 200).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + let response = client.get_on_demand_migration("photos").await.expect("get decodes"); + assert_eq!(response.updated_at, "2026-09-02T10:00:00Z"); + let request = server.recorded(); + assert_eq!(request.method, "GET"); + assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos"); + + let server = TestServer::spawn(STATUS_FIXTURE, 200).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + let status = client.on_demand_migration_status("photos").await.expect("status decodes"); + assert!(status.configured); + let request = server.recorded(); + assert_eq!(request.method, "GET"); + assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos/status"); + } + + #[tokio::test] + async fn delete_accepts_an_empty_204_and_surfaces_other_statuses() { + let server = TestServer::spawn("", 204).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + client.delete_on_demand_migration("photos").await.expect("204 is success"); + let request = server.recorded(); + assert_eq!(request.method, "DELETE"); + assert_eq!(request.path, "/rustfs/admin/v3/on-demand-migration/photos"); + + let server = TestServer::spawn(r#"{"code":"NoSuchConfiguration","message":"not configured"}"#, 404).await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + match client.get_on_demand_migration("photos").await.unwrap_err() { + AdminClientError::HttpStatus { status, body } => { + assert_eq!(status, 404); + assert!(body.contains("NoSuchConfiguration")); + } + other => panic!("expected HttpStatus, got {other:?}"), + } + } +} diff --git a/crates/madmin/src/test_support.rs b/crates/madmin/src/test_support.rs new file mode 100644 index 000000000..5452eaffa --- /dev/null +++ b/crates/madmin/src/test_support.rs @@ -0,0 +1,154 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Test-only HTTP server shared by the admin client modules: one canned +//! response per connection, every request recorded for assertions. + +use std::sync::{Arc, Mutex}; + +/// One recorded request, parsed off the wire with the minimum needed for +/// assertions: method, path, query, headers, body. +#[derive(Debug, Clone)] +pub(crate) struct RecordedRequest { + pub(crate) method: String, + pub(crate) path: String, + pub(crate) query: String, + pub(crate) headers: Vec<(String, String)>, + pub(crate) body: String, +} + +impl RecordedRequest { + pub(crate) fn header(&self, name: &str) -> Option { + self.headers + .iter() + .find(|(key, _)| key.eq_ignore_ascii_case(name)) + .map(|(_, value)| value.clone()) + } +} + +/// Minimal HTTP/1.1 server: one canned response per connection, every +/// request recorded behind an `Arc`. Deliberately dependency-free — +/// the assertions only need the raw request bytes. +pub(crate) struct TestServer { + pub(crate) addr: std::net::SocketAddr, + requests: Arc>>, +} + +impl TestServer { + pub(crate) async fn spawn(response_body: &'static str, status: u16) -> Self { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + + let recorded = requests.clone(); + tokio::spawn(async move { + let reason = match status { + 200 => "OK", + 204 => "No Content", + 400 => "Bad Request", + 404 => "Not Found", + _ => "Forbidden", + }; + let response = format!( + "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response_body}", + response_body.len() + ); + // Each request is a fresh connection (connection: close); a + // bounded loop serves every call a test makes while letting + // the task exit instead of lingering for the whole process. + for _ in 0..16 { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + let mut buffer = Vec::with_capacity(2048); + let mut chunk = [0u8; 2048]; + // Read headers plus content-length body, or stop on close. + loop { + if let Some(end) = find_header_end(&buffer) { + let content_length = extract_content_length(&buffer[..end]); + if buffer.len() >= end + content_length { + break; + } + } + let n = match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => n, + }; + buffer.extend_from_slice(&chunk[..n]); + if buffer.len() > 64 * 1024 { + break; + } + } + if let Some(request) = parse_request(&buffer) { + recorded.lock().expect("recorded lock").push(request); + } + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } + }); + + Self { addr, requests } + } + + pub(crate) fn recorded(&self) -> RecordedRequest { + self.requests + .lock() + .expect("recorded lock") + .last() + .cloned() + .expect("the client call must have produced one recorded request") + } +} + +fn find_header_end(buffer: &[u8]) -> Option { + buffer.windows(4).position(|window| window == b"\r\n\r\n").map(|pos| pos + 4) +} + +fn extract_content_length(headers: &[u8]) -> usize { + let text = String::from_utf8_lossy(headers).to_ascii_lowercase(); + text.lines() + .find_map(|line| line.strip_prefix("content-length:")) + .and_then(|value| value.trim().parse().ok()) + .unwrap_or(0) +} + +fn parse_request(raw: &[u8]) -> Option { + let end = find_header_end(raw)?; + let head = String::from_utf8_lossy(&raw[..end]); + let body = String::from_utf8_lossy(&raw[end..]).into_owned(); + let mut lines = head.lines(); + let request_line = lines.next()?; + let mut parts = request_line.split_whitespace(); + let method = parts.next()?.to_string(); + let target = parts.next()?.to_string(); + let (path, query) = match target.split_once('?') { + Some((path, query)) => (path.to_string(), query.to_string()), + None => (target, String::new()), + }; + let headers = lines + .filter_map(|line| line.split_once(':')) + .map(|(name, value)| (name.trim().to_string(), value.trim().to_string())) + .collect(); + Some(RecordedRequest { + method, + path, + query, + headers, + body, + }) +} diff --git a/crates/policy/src/policy/action.rs b/crates/policy/src/policy/action.rs index 1a3438219..f0076aa34 100644 --- a/crates/policy/src/policy/action.rs +++ b/crates/policy/src/policy/action.rs @@ -465,6 +465,12 @@ pub enum AdminAction { SetBucketTargetAction, #[strum(serialize = "admin:GetBucketTarget")] GetBucketTargetAction, + /// Configure, validate or clear a bucket's on-demand migration source. + #[strum(serialize = "admin:SetBucketOnDemandMigration")] + SetBucketOnDemandMigrationAction, + /// Read a bucket's on-demand migration configuration and status. + #[strum(serialize = "admin:GetBucketOnDemandMigration")] + GetBucketOnDemandMigrationAction, #[strum(serialize = "admin:GetMetrics")] GetMetricsAction, #[strum(serialize = "admin:ReplicationDiff")] @@ -623,6 +629,8 @@ impl AdminAction { | AdminAction::SetBucketQuotaAdminAction | AdminAction::SetBucketTargetAction | AdminAction::GetBucketTargetAction + | AdminAction::SetBucketOnDemandMigrationAction + | AdminAction::GetBucketOnDemandMigrationAction | AdminAction::GetMetricsAction | AdminAction::ReplicationDiff | AdminAction::GetReplicationMetricsAction @@ -835,6 +843,26 @@ mod tests { assert!(AdminAction::GetMetricsAction.is_valid()); } + #[test] + fn test_bucket_on_demand_migration_admin_actions_are_valid() { + let set_action = AdminAction::try_from("admin:SetBucketOnDemandMigration").expect("parse set action"); + let get_action = AdminAction::try_from("admin:GetBucketOnDemandMigration").expect("parse get action"); + + assert_eq!(set_action, AdminAction::SetBucketOnDemandMigrationAction); + assert_eq!(get_action, AdminAction::GetBucketOnDemandMigrationAction); + assert!(set_action.is_valid()); + assert!(get_action.is_valid()); + assert_eq!(<&str>::from(set_action), "admin:SetBucketOnDemandMigration"); + assert_eq!(<&str>::from(get_action), "admin:GetBucketOnDemandMigration"); + + // `admin:*` must cover the new actions without a per-action listing, + // while a read-only grant must not confer the write action. + let all_admin = Action::AdminAction(AdminAction::AllAdminActions); + assert!(all_admin.is_match(&Action::AdminAction(set_action))); + assert!(all_admin.is_match(&Action::AdminAction(get_action))); + assert!(!Action::AdminAction(get_action).is_match(&Action::AdminAction(set_action))); + } + #[test] fn test_table_catalog_admin_action_is_valid() { let get_action = AdminAction::try_from("admin:GetTableCatalog").expect("Should parse GetTableCatalog action"); diff --git a/docs/architecture/admin-route-action-snapshot.md b/docs/architecture/admin-route-action-snapshot.md index b73fdcd1e..b6b18aa47 100644 --- a/docs/architecture/admin-route-action-snapshot.md +++ b/docs/architecture/admin-route-action-snapshot.md @@ -25,3 +25,14 @@ Router-level credential checks (`S3Router::check_access`) are bypassed only for: - console assets (`/favicon.ico`, `/rustfs/console...`), only while the console is enabled. Every other admin route requires credentials at the router and a precise `AdminAction` or `S3Action` check in the handler (metrics routes, for example, authorize `GetMetricsAction`). The MinIO alias contract is specified in [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md). + +## Gated Bucket Feature Routes + +Some bucket-scoped routes add gates after the `AdminAction` check. The gates are enforced in the handler, so they are invisible to the route matrix and listed here instead. + +| Route | Actions | Extra gates after authorization | +|---|---|---| +| `PUT`/`DELETE /rustfs/admin/v3/on-demand-migration/{bucket}` (`?dry-run=true` validates and probes without saving) | `SetBucketOnDemandMigrationAction` (`admin:SetBucketOnDemandMigration`) | bucket must exist (`NoSuchBucket`); `PUT` also requires the `RUSTFS_ON_DEMAND_MIGRATION_ENABLED` module switch (`OnDemandMigrationDisabled`, 400) and the server license (`license_check()`, same mapping as object zip downloads); the source must answer HEAD + a one-key list (`OnDemandMigrationSourceUnreachable`, 400). Handler: `rustfs/src/admin/handlers/on_demand_migration.rs` | +| `GET /rustfs/admin/v3/on-demand-migration/{bucket}` and `GET .../{bucket}/status` | `GetBucketOnDemandMigrationAction` (`admin:GetBucketOnDemandMigration`) | bucket must exist; reads work while the module switch is off so operators can inspect a disabled deployment; `GET` answers `NoSuchConfiguration` (404) when nothing is configured | + +Responses on these routes carry the redacted configuration (`secret_key` and `session_token` replaced by `REDACTED`); the wire shape is pinned by the fixtures under `crates/madmin/fixtures/on_demand_migration/`, shared by the server handler tests and the `rustfs-madmin` client tests. diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 024b2676e..288f1849f 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -49,6 +49,7 @@ mod notify_runtime_access; pub mod object_data_cache; pub mod object_zip_download; pub mod oidc; +pub mod on_demand_migration; pub mod plugins_catalog; pub mod plugins_instances; pub mod policies; @@ -129,6 +130,10 @@ mod tests { let _list_extension_instances = extensions::ListExtensionInstancesHandler {}; let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {}; let _create_object_zip_download = object_zip_download::CreateObjectZipDownloadHandler {}; + let _set_on_demand_migration = on_demand_migration::SetBucketOnDemandMigrationHandler {}; + let _get_on_demand_migration = on_demand_migration::GetBucketOnDemandMigrationHandler {}; + let _delete_on_demand_migration = on_demand_migration::DeleteBucketOnDemandMigrationHandler {}; + let _on_demand_migration_status = on_demand_migration::GetBucketOnDemandMigrationStatusHandler {}; let _list_plugin_instances = plugins_instances::ListPluginInstancesHandler {}; let _get_plugin_instance = plugins_instances::GetPluginInstanceHandler {}; let _put_plugin_instance = plugins_instances::PutPluginInstanceHandler {}; diff --git a/rustfs/src/admin/handlers/on_demand_migration.rs b/rustfs/src/admin/handlers/on_demand_migration.rs new file mode 100644 index 000000000..c6a521ab4 --- /dev/null +++ b/rustfs/src/admin/handlers/on_demand_migration.rs @@ -0,0 +1,1232 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Per-bucket On-Demand Migration admin handlers (ODM-07, rustfs/backlog#2154). +//! +//! A bucket can name an external S3-compatible source whose objects are +//! pulled on first access. This module is the management plane only: +//! `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}` configure, read and +//! clear the source, `?dry-run=true` validates and probes without saving, and +//! `GET .../status` reports the switch state. The data plane, counters and +//! backfill live in later ODM tasks and extend the same routes. +//! +//! Credentials in the request body are never echoed: every response carries +//! the `redacted()` config, probe failures name only the error class, and no +//! log line carries the config. + +use crate::admin::auth::authorize_admin_request; +use crate::admin::handlers::admin_json_response; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::{ + AppContext, app_context_from_req, current_deployment_id, current_endpoints_handle, current_notification_system_for_context, + object_store_from_req, +}; +use crate::admin::storage_api::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG; +use crate::admin::storage_api::bucket::metadata_sys; +use crate::admin::storage_api::bucket::on_demand_migration::source_client::{ + SourceClient, SourceClientSpec, SourceError, SourceProbe, SourceProvider, SourceTimeouts, +}; +use crate::admin::storage_api::bucket::on_demand_migration::{ + OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, ValidationContext, +}; +use crate::admin::storage_api::bucket::remote_s3_client::{PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError}; +use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketOptions}; +use crate::admin::storage_api::error::StorageError; +use crate::admin::storage_api::s3::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, error as admin_s3_error}; +use crate::admin::utils::{extract_query_params, read_compatible_admin_body}; +use crate::error::ApiError; +use crate::license::license_check; +use crate::server::ADMIN_PREFIX; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_credentials::Credentials; +use rustfs_policy::policy::action::{Action, AdminAction}; +use serde::Serialize; +use std::num::NonZeroU64; +use std::sync::Arc; +use std::time::Duration; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; +use tracing::{info, warn}; + +const LOG_COMPONENT_ADMIN: &str = "admin"; +const LOG_SUBSYSTEM_ON_DEMAND_MIGRATION: &str = "bucket_on_demand_migration"; +const EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG: &str = "admin_bucket_on_demand_migration_config"; + +const ROUTE_PATH: &str = "/v3/on-demand-migration/{bucket}"; +const STATUS_ROUTE_PATH: &str = "/v3/on-demand-migration/{bucket}/status"; +const DRY_RUN_QUERY: &str = "dry-run"; + +/// Error code returned when the module switch is off and a write is attempted. +pub(crate) const ERR_CODE_MODULE_DISABLED: &str = "OnDemandMigrationDisabled"; +/// Error code returned when the source bucket did not answer the probe. +pub(crate) const ERR_CODE_SOURCE_UNREACHABLE: &str = "OnDemandMigrationSourceUnreachable"; +/// Error code returned by `GET` when the bucket has no configuration. +pub(crate) const ERR_CODE_NO_SUCH_CONFIGURATION: &str = "NoSuchConfiguration"; + +/// The published switch is `RUSTFS_ON_DEMAND_MIGRATION_ENABLED`, owned by +/// ODM-05 in `module_switches.rs`. This is the only read of it in the admin +/// plane so the orchestrator can swap the call for the published predicate. +const ENV_ON_DEMAND_MIGRATION_ENABLED: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED"; + +fn module_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_ON_DEMAND_MIGRATION_ENABLED, false) +} + +/// What the source answered during `PUT` validation. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ProbeSummary { + /// `HeadBucket` succeeded. + pub reachable: bool, + /// A one-key `ListObjectsV2` succeeded. + pub listable: bool, + /// The first key the listing returned, if the source bucket is not empty. + pub sample_key: Option, +} + +impl From for ProbeSummary { + fn from(probe: SourceProbe) -> Self { + Self { + reachable: true, + listable: true, + sample_key: probe.sample_object.map(|object| object.key), + } + } +} + +#[derive(Debug, Serialize)] +pub(crate) struct SetBucketOnDemandMigrationResponse { + pub bucket: String, + pub dry_run: bool, + pub config: OnDemandMigrationConfig, + /// `null` for a dry run: nothing was saved. + pub updated_at: Option, + pub probe: ProbeSummary, +} + +#[derive(Debug, Serialize)] +pub(crate) struct GetBucketOnDemandMigrationResponse { + pub bucket: String, + pub config: OnDemandMigrationConfig, + pub updated_at: String, +} + +#[derive(Debug, Serialize)] +pub(crate) struct BucketOnDemandMigrationStatus { + pub configured: bool, + pub enabled: bool, + pub module_enabled: bool, +} + +pub struct SetBucketOnDemandMigrationHandler; +pub struct GetBucketOnDemandMigrationHandler; +pub struct DeleteBucketOnDemandMigrationHandler; +pub struct GetBucketOnDemandMigrationStatusHandler; + +pub fn register_on_demand_migration_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::PUT, + format!("{ADMIN_PREFIX}{ROUTE_PATH}").as_str(), + AdminOperation(&SetBucketOnDemandMigrationHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}{ROUTE_PATH}").as_str(), + AdminOperation(&GetBucketOnDemandMigrationHandler {}), + )?; + r.insert( + Method::DELETE, + format!("{ADMIN_PREFIX}{ROUTE_PATH}").as_str(), + AdminOperation(&DeleteBucketOnDemandMigrationHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}{STATUS_ROUTE_PATH}").as_str(), + AdminOperation(&GetBucketOnDemandMigrationStatusHandler {}), + )?; + Ok(()) +} + +fn custom_error(code: &'static str, status: StatusCode, message: String) -> S3Error { + let mut err = S3Error::with_message(S3ErrorCode::Custom(code.into()), message); + err.set_status_code(status); + err +} + +fn bucket_from_params(params: &Params<'_, '_>) -> S3Result { + let bucket = params.get("bucket").unwrap_or("").to_string(); + if bucket.is_empty() { + return Err(admin_s3_error(S3ErrorCode::InvalidRequest, "bucket name is required")); + } + Ok(bucket) +} + +/// Authorize with `action` and confirm the bucket exists (404 otherwise). +async fn authorize_for_bucket(req: &S3Request, action: AdminAction, bucket: &str) -> S3Result { + let cred = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?; + + let Some(store) = object_store_from_req(req) else { + return Err(admin_s3_error(S3ErrorCode::InternalError, "object store is not initialized")); + }; + store + .get_bucket_info(bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + Ok(cred) +} + +/// Same mapping as the object zip download: a denied entitlement is +/// `AccessDenied`, anything else is an internal failure whose detail stays in +/// the log. +fn license_gate() -> S3Result<()> { + license_check().map_err(|err| match err.kind() { + std::io::ErrorKind::PermissionDenied => admin_s3_error(S3ErrorCode::AccessDenied, format!("{err}")), + _ => { + tracing::error!( + event = EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + error = %err, + "license check failed" + ); + admin_s3_error(S3ErrorCode::InternalError, "License validation failed") + } + }) +} + +fn parse_config(body: &[u8]) -> S3Result { + if body.is_empty() { + return Err(admin_s3_error( + S3ErrorCode::InvalidRequest, + "request body is required: an on-demand migration config JSON", + )); + } + OnDemandMigrationConfig::from_json(body).map_err(|err| match err { + OnDemandMigrationConfigError::Malformed(reason) => { + admin_s3_error(S3ErrorCode::InvalidArgument, format!("invalid on-demand migration config: {reason}")) + } + other => admin_s3_error(S3ErrorCode::InvalidArgument, format!("{other}")), + }) +} + +fn is_dry_run(req: &S3Request) -> bool { + extract_query_params(&req.uri) + .get(DRY_RUN_QUERY) + .is_some_and(|value| value.eq_ignore_ascii_case("true")) +} + +/// Every endpoint of this deployment, as `scheme://host:port`, so a source +/// naming one of them with the same bucket is rejected as a self-reference. +/// Single-node local-disk layouts carry no host and contribute nothing; the +/// outbound endpoint policy still refuses loopback sources for them. +fn local_endpoints() -> Vec { + let mut endpoints: Vec = current_endpoints_handle() + .map(|pools| { + pools + .as_ref() + .iter() + .flat_map(|pool| pool.endpoints.as_ref().iter()) + .map(|endpoint| endpoint.grid_host()) + .filter(|host| !host.is_empty()) + .collect() + }) + .unwrap_or_default(); + endpoints.sort(); + endpoints.dedup(); + endpoints +} + +/// `(endpoint URL, target bucket)` of every replication target of `bucket`; +/// a bucket that never had targets configured has none. +async fn replication_target_endpoints(bucket: &str) -> S3Result> { + let targets = match metadata_sys::list_bucket_targets(bucket).await { + Ok(targets) => targets, + Err(StorageError::ConfigNotFound) => return Ok(Vec::new()), + Err(err) => { + return Err(admin_s3_error( + S3ErrorCode::InternalError, + format!("failed to read replication targets: {err}"), + )); + } + }; + Ok(targets + .targets + .iter() + .filter_map(|target| target.url().ok().map(|url| (url.to_string(), target.target_bucket.clone()))) + .collect()) +} + +async fn validate_config(bucket: &str, config: &OnDemandMigrationConfig) -> S3Result<()> { + let deployment_id = current_deployment_id().unwrap_or_default(); + let local_endpoints = local_endpoints(); + let replication_targets = replication_target_endpoints(bucket).await?; + config + .validate(ValidationContext { + local_bucket: bucket, + local_deployment_id: &deployment_id, + local_endpoints: &local_endpoints, + replication_target_endpoints: &replication_targets, + }) + .map_err(|err| admin_s3_error(S3ErrorCode::InvalidArgument, format!("{err}"))) +} + +fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider { + use crate::admin::storage_api::bucket::on_demand_migration::Provider; + match config.source.provider { + Provider::S3 => SourceProvider::S3, + Provider::Aws => SourceProvider::Aws, + Provider::Minio => SourceProvider::Minio, + Provider::Rustfs => SourceProvider::Rustfs, + Provider::R2 => SourceProvider::R2, + Provider::Gcs => SourceProvider::Gcs, + } +} + +/// Map the persisted config onto the client builder's spec. `first_byte_ms` +/// bounds the read timeout: the probe only issues HEAD and a one-key list. +pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClientSpec { + let source = &config.source; + let timeout = &config.policy.source_timeout; + SourceClientSpec { + endpoint: source.effective_endpoint(), + region: source.effective_region().to_string(), + bucket: source.bucket.clone(), + source_prefix: config.filter.source_prefix.clone(), + provider: source_provider(config), + path_style: match source.path_style { + PathStyle::Auto => RemotePathStyle::Auto, + PathStyle::Path => RemotePathStyle::Path, + PathStyle::Virtual => RemotePathStyle::VirtualHost, + }, + credentials: source.credentials.as_ref().map(|credentials| RemoteCredentials { + access_key: credentials.access_key.clone(), + secret_key: credentials.secret_key.clone(), + session_token: credentials.session_token.clone(), + expiration: None, + account_id: String::new(), + }), + skip_tls_verify: source.tls.skip_verify, + ca_cert_pem: source.tls.ca_cert_pem.clone(), + timeouts: SourceTimeouts { + connect: Duration::from_millis(timeout.connect_ms), + read: Duration::from_millis(timeout.first_byte_ms), + }, + bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new), + } +} + +/// Builder failures are input errors: the endpoint policy, the CA PEM or the +/// credentials the operator supplied. Anonymous sources are not wired yet +/// (ODM-05 adds the credential-less path), so `MissingCredentials` is a 400 +/// naming the field instead of an opaque internal error. +fn client_build_error(err: RemoteS3ClientError) -> S3Error { + match err { + RemoteS3ClientError::MissingCredentials => admin_s3_error( + S3ErrorCode::InvalidArgument, + "source.credentials is required: anonymous sources are not supported yet", + ), + other => admin_s3_error(S3ErrorCode::InvalidArgument, format!("source client cannot be built: {other}")), + } +} + +/// Only the error class crosses the boundary: SDK messages can carry the +/// signed request, including the endpoint host and query. +pub(crate) fn probe_error(err: &SourceError) -> S3Error { + custom_error( + ERR_CODE_SOURCE_UNREACHABLE, + StatusCode::BAD_REQUEST, + format!("source bucket probe failed: {}", err.class_label()), + ) +} + +async fn probe_source(bucket: &str, config: &OnDemandMigrationConfig) -> S3Result { + let spec = source_client_spec(config); + let client = SourceClient::new(&spec).await.map_err(client_build_error)?; + match client.probe().await { + Ok(probe) => Ok(ProbeSummary::from(probe)), + Err(err) => { + warn!( + event = EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + bucket = %bucket, + probe_error_class = err.class_label(), + "on-demand migration source probe failed" + ); + Err(probe_error(&err)) + } + } +} + +fn format_updated_at(updated_at: OffsetDateTime) -> S3Result { + updated_at + .format(&Rfc3339) + .map_err(|err| admin_s3_error(S3ErrorCode::InternalError, format!("failed to format timestamp: {err}"))) +} + +/// Ask every peer to reload the bucket metadata so the new source takes +/// effect cluster-wide before the periodic refresh. A peer that does not +/// answer is a warning: the refresh loop converges it. +pub(crate) async fn reload_peers(context: Option<&AppContext>, bucket: &str) -> Result<(), String> { + let Some(notification_sys) = current_notification_system_for_context(context) else { + return Ok(()); + }; + notification_sys + .load_bucket_metadata(bucket) + .await + .map_err(|err| err.to_string()) +} + +fn notify_peers_reload(context: Option>, bucket: String, operation: &'static str) { + tokio::spawn(async move { + if let Err(error) = reload_peers(context.as_deref(), &bucket).await { + warn!( + event = EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + bucket = %bucket, + error = %error, + "failed to notify peers after {operation}" + ); + } + }); +} + +#[async_trait::async_trait] +impl Operation for SetBucketOnDemandMigrationHandler { + #[tracing::instrument(skip_all)] + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let bucket = bucket_from_params(¶ms)?; + let cred = authorize_for_bucket(&req, AdminAction::SetBucketOnDemandMigrationAction, &bucket).await?; + + if !module_enabled() { + return Err(custom_error( + ERR_CODE_MODULE_DISABLED, + StatusCode::BAD_REQUEST, + format!("on-demand migration is disabled: set {ENV_ON_DEMAND_MIGRATION_ENABLED}=true"), + )); + } + license_gate()?; + + let dry_run = is_dry_run(&req); + let path = req.uri.path().to_string(); + let context = app_context_from_req(&req); + + // Taken before the (possibly slow) probe: a bucket recreated meanwhile + // must not receive this config. + let expected_incarnation_id = metadata_sys::capture_bucket_metadata_incarnation(&bucket) + .await + .map_err(|err| admin_s3_error(S3ErrorCode::InternalError, format!("failed to capture bucket incarnation: {err}")))?; + + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, &path, &cred.secret_key).await?; + let config = parse_config(&body)?; + validate_config(&bucket, &config).await?; + let probe = probe_source(&bucket, &config).await?; + + let updated_at = if dry_run { + None + } else { + let json = config + .to_json() + .map_err(|err| admin_s3_error(S3ErrorCode::InternalError, format!("failed to encode config: {err}")))?; + let updated_at = + metadata_sys::update_if_incarnation(&bucket, BUCKET_ON_DEMAND_MIGRATION_CONFIG, json, expected_incarnation_id) + .await + .map_err(|err| { + admin_s3_error(S3ErrorCode::InternalError, format!("failed to save on-demand migration config: {err}")) + })?; + info!( + event = EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + bucket = %bucket, + enabled = config.enabled, + "on-demand migration config set" + ); + notify_peers_reload(context, bucket.clone(), "set on-demand migration config"); + Some(format_updated_at(updated_at)?) + }; + + let response = SetBucketOnDemandMigrationResponse { + bucket, + dry_run, + config: config.redacted(), + updated_at, + probe, + }; + admin_json_response(&path, &cred.secret_key, StatusCode::OK, &response) + } +} + +#[async_trait::async_trait] +impl Operation for GetBucketOnDemandMigrationHandler { + #[tracing::instrument(skip_all)] + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let bucket = bucket_from_params(¶ms)?; + let cred = authorize_for_bucket(&req, AdminAction::GetBucketOnDemandMigrationAction, &bucket).await?; + + let Some((config, updated_at)) = metadata_sys::get_on_demand_migration_config(&bucket).await.map_err(|err| { + admin_s3_error(S3ErrorCode::InternalError, format!("failed to read on-demand migration config: {err}")) + })? + else { + return Err(custom_error( + ERR_CODE_NO_SUCH_CONFIGURATION, + StatusCode::NOT_FOUND, + format!("on-demand migration is not configured for bucket {bucket}"), + )); + }; + + let response = GetBucketOnDemandMigrationResponse { + bucket, + config: config.redacted(), + updated_at: format_updated_at(updated_at)?, + }; + admin_json_response(req.uri.path(), &cred.secret_key, StatusCode::OK, &response) + } +} + +#[async_trait::async_trait] +impl Operation for DeleteBucketOnDemandMigrationHandler { + #[tracing::instrument(skip_all)] + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let bucket = bucket_from_params(¶ms)?; + authorize_for_bucket(&req, AdminAction::SetBucketOnDemandMigrationAction, &bucket).await?; + + let expected_incarnation_id = metadata_sys::capture_bucket_metadata_incarnation(&bucket) + .await + .map_err(|err| admin_s3_error(S3ErrorCode::InternalError, format!("failed to capture bucket incarnation: {err}")))?; + + // Idempotent: clearing an absent config still rewrites the metadata + // (fresh timestamp) and answers 204. Pulled objects stay in place. + metadata_sys::delete_if_incarnation(&bucket, BUCKET_ON_DEMAND_MIGRATION_CONFIG, expected_incarnation_id) + .await + .map_err(|err| { + admin_s3_error(S3ErrorCode::InternalError, format!("failed to clear on-demand migration config: {err}")) + })?; + + info!( + event = EVENT_ADMIN_ON_DEMAND_MIGRATION_CONFIG, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + bucket = %bucket, + "on-demand migration config cleared" + ); + notify_peers_reload(app_context_from_req(&req), bucket, "clear on-demand migration config"); + + Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty()))) + } +} + +#[async_trait::async_trait] +impl Operation for GetBucketOnDemandMigrationStatusHandler { + #[tracing::instrument(skip_all)] + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let bucket = bucket_from_params(¶ms)?; + let cred = authorize_for_bucket(&req, AdminAction::GetBucketOnDemandMigrationAction, &bucket).await?; + + let config = metadata_sys::get_on_demand_migration_config(&bucket).await.map_err(|err| { + admin_s3_error(S3ErrorCode::InternalError, format!("failed to read on-demand migration config: {err}")) + })?; + + let status = BucketOnDemandMigrationStatus { + configured: config.is_some(), + enabled: config.is_some_and(|(config, _)| config.enabled), + module_enabled: module_enabled(), + }; + admin_json_response(req.uri.path(), &cred.secret_key, StatusCode::OK, &status) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use http::Uri; + + /// `matchit::Params` cannot be built by hand; route a sample path the way + /// the admin router does. + pub(super) fn bucket_router() -> matchit::Router<()> { + let mut router = matchit::Router::new(); + router.insert("/{bucket}", ()).expect("bucket route"); + router + } + + /// Wire fixtures shared with the `rustfs-madmin` client tests: both sides + /// must reproduce these bytes exactly. + const SET_REQUEST_FIXTURE: &str = include_str!("../../../../crates/madmin/fixtures/on_demand_migration/set_request.json"); + const SET_RESPONSE_FIXTURE: &str = include_str!("../../../../crates/madmin/fixtures/on_demand_migration/set_response.json"); + const GET_RESPONSE_FIXTURE: &str = include_str!("../../../../crates/madmin/fixtures/on_demand_migration/get_response.json"); + const STATUS_FIXTURE: &str = include_str!("../../../../crates/madmin/fixtures/on_demand_migration/status.json"); + const FIXTURE_UPDATED_AT: &str = "2026-09-02T10:00:00Z"; + + fn fixture_config() -> OnDemandMigrationConfig { + OnDemandMigrationConfig::from_json(SET_REQUEST_FIXTURE.trim().as_bytes()).expect("fixture config parses") + } + + #[test] + fn parse_config_rejects_empty_body_unknown_fields_and_malformed_json() { + assert_eq!(parse_config(b"").unwrap_err().code(), &S3ErrorCode::InvalidRequest); + assert_eq!(parse_config(b"{").unwrap_err().code(), &S3ErrorCode::InvalidArgument); + + let unknown = + br#"{"source":{"provider":"s3","endpoint":"https://s.example","region":"us-east-1","bucket":"b"},"bogus":1}"#; + let err = parse_config(unknown).unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + assert!( + err.message().unwrap_or_default().contains("bogus"), + "message must name the offending field" + ); + + assert!(parse_config(SET_REQUEST_FIXTURE.trim().as_bytes()).is_ok()); + } + + #[test] + fn set_request_fixture_is_the_plaintext_wire_shape() { + let config = fixture_config(); + assert_eq!( + config.source.credentials.as_ref().map(|c| c.secret_key.as_str()), + Some("sourceSecretKey123") + ); + // A client sends the exact bytes the server re-serializes: the fixture is canonical. + let reserialized = String::from_utf8(config.to_json().expect("serialize")).expect("utf-8"); + assert_eq!(reserialized, SET_REQUEST_FIXTURE.trim()); + } + + #[test] + fn set_response_matches_madmin_golden_fixture() { + let response = SetBucketOnDemandMigrationResponse { + bucket: "photos".to_string(), + dry_run: false, + config: fixture_config().redacted(), + updated_at: Some(FIXTURE_UPDATED_AT.to_string()), + probe: ProbeSummary { + reachable: true, + listable: true, + sample_key: Some("photos/2024/01.jpg".to_string()), + }, + }; + let json = serde_json::to_string(&response).expect("serialize"); + assert_eq!(json, SET_RESPONSE_FIXTURE.trim()); + assert!(!json.contains("sourceSecretKey123"), "responses must never echo the secret key"); + assert!(json.contains(r#""secret_key":"REDACTED""#)); + } + + #[test] + fn get_response_matches_madmin_golden_fixture() { + let response = GetBucketOnDemandMigrationResponse { + bucket: "photos".to_string(), + config: fixture_config().redacted(), + updated_at: FIXTURE_UPDATED_AT.to_string(), + }; + assert_eq!(serde_json::to_string(&response).expect("serialize"), GET_RESPONSE_FIXTURE.trim()); + } + + #[test] + fn status_matches_madmin_golden_fixture() { + let status = BucketOnDemandMigrationStatus { + configured: true, + enabled: true, + module_enabled: false, + }; + assert_eq!(serde_json::to_string(&status).expect("serialize"), STATUS_FIXTURE.trim()); + } + + #[test] + fn updated_at_uses_rfc3339_utc() { + let ts = OffsetDateTime::from_unix_timestamp(1_788_343_200).expect("timestamp"); + assert_eq!(format_updated_at(ts).expect("format"), FIXTURE_UPDATED_AT); + } + + #[test] + fn source_client_spec_maps_provider_path_style_credentials_and_timeouts() { + let config = fixture_config(); + let spec = source_client_spec(&config); + + assert_eq!(spec.endpoint, "https://source.example.com:9000"); + assert_eq!(spec.region, "us-east-1"); + assert_eq!(spec.bucket, "legacy-photos"); + assert_eq!(spec.source_prefix.as_deref(), Some("photos/")); + assert_eq!(spec.provider, SourceProvider::Minio); + assert_eq!(spec.path_style, RemotePathStyle::Auto); + let credentials = spec.credentials.as_ref().expect("credentials mapped"); + assert_eq!(credentials.access_key, "AKIASOURCE"); + assert_eq!(credentials.secret_key, "sourceSecretKey123"); + assert_eq!(credentials.session_token, None); + assert_eq!(spec.timeouts.connect, Duration::from_millis(5000)); + assert_eq!(spec.timeouts.read, Duration::from_millis(15_000)); + assert_eq!(spec.bandwidth_limit, None); + + let mut virtual_host = config.clone(); + virtual_host.source.path_style = PathStyle::Virtual; + virtual_host.policy.bandwidth_limit_bytes_per_sec = Some(1 << 20); + let spec = source_client_spec(&virtual_host); + assert_eq!(spec.path_style, RemotePathStyle::VirtualHost); + assert_eq!(spec.bandwidth_limit, NonZeroU64::new(1 << 20)); + + let mut anonymous = config; + anonymous.source.credentials = None; + assert!(source_client_spec(&anonymous).credentials.is_none()); + } + + #[test] + fn probe_error_names_only_the_error_class() { + let err = probe_error(&SourceError::Connect( + "dispatch failure: GET https://AKIASOURCE:sourceSecretKey123@source.example.com:9000/legacy-photos".to_string(), + )); + assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_SOURCE_UNREACHABLE.into())); + assert_eq!(err.status_code(), Some(StatusCode::BAD_REQUEST)); + let message = err.message().unwrap_or_default(); + assert_eq!(message, "source bucket probe failed: connect"); + assert!(!message.contains("sourceSecretKey123")); + assert!(!message.contains("source.example.com")); + + let denied = probe_error(&SourceError::AccessDenied); + assert_eq!(denied.message(), Some("source bucket probe failed: access_denied")); + } + + #[test] + fn missing_credentials_is_a_400_naming_the_field() { + let err = client_build_error(RemoteS3ClientError::MissingCredentials); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + assert!(err.message().unwrap_or_default().contains("source.credentials")); + } + + #[test] + fn module_switch_defaults_off_and_reads_the_env() { + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, None::<&str>, || assert!(!module_enabled())); + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("true"), || assert!(module_enabled())); + temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("false"), || assert!(!module_enabled())); + } + + #[test] + fn dry_run_query_requires_a_literal_true() { + let request = |uri: &'static str| S3Request { + input: Body::empty(), + method: Method::PUT, + uri: Uri::from_static(uri), + headers: http::HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + assert!(is_dry_run(&request("/rustfs/admin/v3/on-demand-migration/b?dry-run=true"))); + assert!(is_dry_run(&request("/rustfs/admin/v3/on-demand-migration/b?dry-run=TRUE"))); + assert!(!is_dry_run(&request("/rustfs/admin/v3/on-demand-migration/b?dry-run=1"))); + assert!(!is_dry_run(&request("/rustfs/admin/v3/on-demand-migration/b"))); + } + + #[cfg(not(feature = "license"))] + #[test] + fn license_gate_always_passes_without_the_license_feature() { + assert!(license_gate().is_ok()); + } + + /// Strict builds refuse writes until a license is installed; the mapping + /// is the one the object zip download uses. + #[cfg(feature = "license")] + #[test] + #[serial_test::serial] + fn license_gate_rejects_writes_without_a_license() { + let err = license_gate().expect_err("strict build without a license must refuse"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn handlers_reject_requests_without_credentials_before_touching_storage() { + let request = |method: Method, uri: &'static str| S3Request { + input: Body::empty(), + method, + uri: Uri::from_static(uri), + headers: http::HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + let router = bucket_router(); + let params = router.at("/photos").expect("route matches").params; + let handlers: [(&dyn Operation, Method, &'static str); 4] = [ + ( + &SetBucketOnDemandMigrationHandler {}, + Method::PUT, + "/rustfs/admin/v3/on-demand-migration/photos", + ), + ( + &GetBucketOnDemandMigrationHandler {}, + Method::GET, + "/rustfs/admin/v3/on-demand-migration/photos", + ), + ( + &DeleteBucketOnDemandMigrationHandler {}, + Method::DELETE, + "/rustfs/admin/v3/on-demand-migration/photos", + ), + ( + &GetBucketOnDemandMigrationStatusHandler {}, + Method::GET, + "/rustfs/admin/v3/on-demand-migration/photos/status", + ), + ]; + for (handler, method, uri) in handlers { + let err = handler + .call(request(method, uri), params.clone()) + .await + .expect_err("a request without credentials must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(err.message(), Some("get cred failed")); + } + } + + #[tokio::test] + async fn handlers_reject_an_empty_bucket_path_param() { + let request = S3Request { + input: Body::empty(), + method: Method::GET, + uri: Uri::from_static("/rustfs/admin/v3/on-demand-migration/"), + headers: http::HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + let err = GetBucketOnDemandMigrationHandler {} + .call(request, Params::new()) + .await + .expect_err("an empty bucket must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(err.message(), Some("bucket name is required")); + } +} + +/// Store-backed coverage: one `TestECStoreEnv` per test binary is the rule +/// (ambient globals), so every scenario runs from a single test body. +#[cfg(all(test, not(feature = "license")))] +mod store_tests { + use super::*; + use crate::admin::runtime_sources::{NotificationSystemInterface, publish_test_app_context}; + use crate::admin::storage_api::NotificationSys; + use crate::admin::storage_api::runtime_sources::ECStore; + use crate::admin::storage_api::s3::auth as s3_auth; + use http::{Extensions, HeaderMap, Uri}; + use http_body_util::BodyExt as _; + use rustfs_iam::store::{Store as _, UserType}; + use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq}; + use rustfs_policy::policy::Policy; + use serde_json::Value; + use std::sync::Arc; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + const ROOT_ACCESS_KEY: &str = "ODMADMINROOT"; + const ROOT_SECRET_KEY: &str = "odmAdminRootSecret123"; + const READER_ACCESS_KEY: &str = "odmreader"; + const READER_SECRET_KEY: &str = "odmReaderSecret123"; + const BUCKET: &str = "photos"; + const SOURCE_BUCKET: &str = "legacy-photos"; + + fn admin_request(method: Method, uri: String, body: Vec, access_key: &str, secret_key: &str) -> S3Request { + S3Request { + input: Body::from(body), + method, + uri: Uri::try_from(uri).expect("valid uri"), + headers: HeaderMap::new(), + extensions: Extensions::new(), + credentials: Some(s3_auth::Credentials { + access_key: access_key.to_string(), + secret_key: s3_auth::SecretKey::from(secret_key.to_string()), + }), + region: None, + service: None, + trailing_headers: None, + } + } + + fn root_request(method: Method, uri: String, body: Vec) -> S3Request { + admin_request(method, uri, body, ROOT_ACCESS_KEY, ROOT_SECRET_KEY) + } + + fn bucket_router() -> matchit::Router<()> { + super::tests::bucket_router() + } + + fn bucket_params(router: &matchit::Router<()>) -> Params<'_, 'static> { + router.at(concat!("/", "photos")).expect("route matches").params + } + + fn config_uri(query: &str) -> String { + format!("/rustfs/admin/v3/on-demand-migration/{BUCKET}{query}") + } + + fn config_json(endpoint: &str) -> Vec { + serde_json::json!({ + "source": { + "provider": "minio", + "endpoint": endpoint, + "region": "us-east-1", + "bucket": SOURCE_BUCKET, + "credentials": {"access_key": "AKIASOURCE", "secret_key": "sourceSecretKey123"} + }, + "policy": {"source_timeout": {"connect_ms": 500, "first_byte_ms": 2000, "idle_ms": 2000}} + }) + .to_string() + .into_bytes() + } + + async fn response_json(response: S3Response<(StatusCode, Body)>) -> (StatusCode, Value) { + let (status, body) = response.output; + let bytes = body.collect().await.expect("collect body").to_bytes(); + let value = serde_json::from_slice(&bytes).expect("json body"); + (status, value) + } + + async fn get_config() -> Result<(StatusCode, Value), S3Error> { + let router = bucket_router(); + let response = GetBucketOnDemandMigrationHandler {} + .call(root_request(Method::GET, config_uri(""), Vec::new()), bucket_params(&router)) + .await?; + Ok(response_json(response).await) + } + + async fn status() -> Value { + let router = bucket_router(); + let response = GetBucketOnDemandMigrationStatusHandler {} + .call(root_request(Method::GET, config_uri("/status"), Vec::new()), bucket_params(&router)) + .await + .expect("status is readable"); + response_json(response).await.1 + } + + /// A minimal S3 source: `HEAD /{bucket}` answers 200 and a one-key + /// `ListObjectsV2` returns `sample_key`; everything else is 404. + async fn spawn_fake_source(sample_key: &'static str) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind fake source"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + for _ in 0..32 { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + let mut buffer = Vec::with_capacity(4096); + let mut chunk = [0u8; 4096]; + loop { + if buffer.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => buffer.extend_from_slice(&chunk[..n]), + } + } + let head = String::from_utf8_lossy(&buffer); + let request_line = head.lines().next().unwrap_or_default().to_string(); + let mut parts = request_line.split_whitespace(); + let method = parts.next().unwrap_or_default(); + let target = parts.next().unwrap_or_default(); + let (path, query) = target.split_once('?').unwrap_or((target, "")); + let bucket_path = format!("/{SOURCE_BUCKET}"); + let is_bucket = path.trim_end_matches('/') == bucket_path; + let response = if method == "HEAD" && is_bucket { + "HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".to_string() + } else if method == "GET" && is_bucket && query.contains("list-type=2") { + let body = format!( + "{SOURCE_BUCKET}11false{sample_key}1" + ); + format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) + } else { + eprintln!("fake source: unmatched request line {request_line:?}"); + "HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".to_string() + }; + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } + }); + format!("http://{addr}") + } + + /// A loopback port nothing listens on: the probe must fail with `connect`. + async fn closed_endpoint() -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind probe port"); + let addr = listener.local_addr().expect("local addr"); + drop(listener); + format!("http://{addr}") + } + + struct TestNotificationSystemInterface(Arc); + + impl NotificationSystemInterface for TestNotificationSystemInterface { + fn handle(&self) -> Option> { + Some(self.0.clone()) + } + } + + async fn seed_reader(iam: &rustfs_iam::sys::IamSys) { + let policy = Policy::parse_config( + br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:GetBucketOnDemandMigration"],"Resource":["arn:aws:s3:::*"]}]}"#, + ) + .expect("reader policy parses"); + iam.set_policy("odm-reader", policy).await.expect("store reader policy"); + iam.create_user( + READER_ACCESS_KEY, + &AddOrUpdateUserReq { + secret_key: READER_SECRET_KEY.to_string(), + policy: None, + status: AccountStatus::Enabled, + }, + ) + .await + .expect("create reader"); + iam.policy_db_set(READER_ACCESS_KEY, UserType::Reg, false, "odm-reader") + .await + .expect("attach reader policy"); + } + + async fn build_env(temp: &std::path::Path) -> Arc { + let _ = rustfs_credentials::init_global_action_credentials( + Some(ROOT_ACCESS_KEY.to_string()), + Some(ROOT_SECRET_KEY.to_string()), + ); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(temp) + .disk_count(1) + .build() + .await; + env.make_bucket(BUCKET, false).await; + rustfs_iam::store::object::ObjectStore::new(Arc::clone(&env.ecstore)) + .save_iam_config( + serde_json::json!({"version": 1}), + format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX), + ) + .await + .expect("seed IAM format"); + let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore)) + .await + .expect("build test IAM"); + seed_reader(&iam).await; + + // One unreachable peer: `reload_peers` must report it, proving the + // notification fan-out ran after a successful write. + let mut notification_system = NotificationSys::new(Default::default()).await; + notification_system.peer_clients.push(None); + let context = + AppContext::with_default_interfaces(Arc::clone(&env.ecstore), iam, Arc::new(rustfs_kms::KmsServiceManager::new())) + .with_test_notification_system_interface(Arc::new(TestNotificationSystemInterface(Arc::new( + notification_system, + )))); + publish_test_app_context(Arc::new(context)); + Arc::clone(&env.ecstore) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial] + async fn admin_api_covers_switch_validation_probe_persistence_and_authorization() { + let temp = tempfile::tempdir().expect("odm admin test root"); + let _store = build_env(temp.path()).await; + let router = bucket_router(); + + // Module switch off: reads work, writes are refused with the dedicated code. + temp_env::async_with_vars([(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("false"))], async { + let err = SetBucketOnDemandMigrationHandler {} + .call( + root_request(Method::PUT, config_uri(""), config_json("https://source.example.com")), + bucket_params(&router), + ) + .await + .expect_err("PUT must be refused while the module is disabled"); + assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_MODULE_DISABLED.into())); + assert_eq!(err.status_code(), Some(StatusCode::BAD_REQUEST)); + + let err = get_config().await.expect_err("nothing is configured yet"); + assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_NO_SUCH_CONFIGURATION.into())); + assert_eq!(err.status_code(), Some(StatusCode::NOT_FOUND)); + assert_eq!( + status().await, + serde_json::json!({"configured": false, "enabled": false, "module_enabled": false}) + ); + }) + .await; + + temp_env::async_with_vars( + [ + (ENV_ON_DEMAND_MIGRATION_ENABLED, Some("true")), + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")), + ], + async { + // Unknown bucket: 404 before any body is read. + let missing = router.at("/no-such-bucket").expect("route matches").params; + let err = SetBucketOnDemandMigrationHandler {} + .call( + root_request( + Method::PUT, + "/rustfs/admin/v3/on-demand-migration/no-such-bucket".to_string(), + config_json("https://source.example.com"), + ), + missing, + ) + .await + .expect_err("unknown bucket must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchBucket); + + // Validation failure (empty region) does not persist anything. + let mut invalid: Value = serde_json::from_slice(&config_json("https://source.example.com")).expect("json"); + invalid["source"]["region"] = Value::String(String::new()); + let err = SetBucketOnDemandMigrationHandler {} + .call( + root_request(Method::PUT, config_uri(""), invalid.to_string().into_bytes()), + bucket_params(&router), + ) + .await + .expect_err("invalid config must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + assert!(err.message().unwrap_or_default().contains("region")); + assert!(get_config().await.is_err(), "a rejected PUT must not persist"); + + // Probe failure: 400 with the dedicated code, no secret in the message, nothing persisted. + let closed = closed_endpoint().await; + let err = SetBucketOnDemandMigrationHandler {} + .call(root_request(Method::PUT, config_uri(""), config_json(&closed)), bucket_params(&router)) + .await + .expect_err("unreachable source must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_SOURCE_UNREACHABLE.into())); + assert_eq!(err.status_code(), Some(StatusCode::BAD_REQUEST)); + let rendered = format!("{err:?}"); + assert!(!rendered.contains("sourceSecretKey123"), "probe errors must not carry the secret"); + assert!(get_config().await.is_err(), "a failed probe must not persist"); + + // Dry run: probe succeeds, response is redacted, still nothing persisted. + let source = spawn_fake_source("photos/2024/01.jpg").await; + let response = SetBucketOnDemandMigrationHandler {} + .call( + root_request(Method::PUT, config_uri("?dry-run=true"), config_json(&source)), + bucket_params(&router), + ) + .await + .expect("dry run succeeds against the fake source"); + let (status_code, body) = response_json(response).await; + assert_eq!(status_code, StatusCode::OK); + assert_eq!(body["dry_run"], Value::Bool(true)); + assert_eq!(body["updated_at"], Value::Null); + assert_eq!( + body["probe"], + serde_json::json!({"reachable": true, "listable": true, "sample_key": "photos/2024/01.jpg"}) + ); + assert_eq!(body["config"]["source"]["credentials"]["secret_key"], Value::String("REDACTED".into())); + assert!(get_config().await.is_err(), "a dry run must not persist"); + + // Real PUT persists; GET returns the redacted config and the same timestamp. + let response = SetBucketOnDemandMigrationHandler {} + .call(root_request(Method::PUT, config_uri(""), config_json(&source)), bucket_params(&router)) + .await + .expect("PUT succeeds against the fake source"); + let (status_code, body) = response_json(response).await; + assert_eq!(status_code, StatusCode::OK); + assert_eq!(body["dry_run"], Value::Bool(false)); + let first_updated_at = body["updated_at"].as_str().expect("updated_at is set").to_string(); + + let (status_code, body) = get_config().await.expect("config is readable after PUT"); + assert_eq!(status_code, StatusCode::OK); + assert_eq!(body["bucket"], Value::String(BUCKET.into())); + assert_eq!(body["config"]["source"]["credentials"]["secret_key"], Value::String("REDACTED".into())); + assert_eq!(body["config"]["source"]["credentials"]["access_key"], Value::String("AKIASOURCE".into())); + assert_eq!(body["updated_at"], Value::String(first_updated_at.clone())); + assert_eq!( + status().await, + serde_json::json!({"configured": true, "enabled": true, "module_enabled": true}) + ); + + // The peer fan-out ran: the single unreachable peer is reported. + let context = crate::admin::runtime_sources::current_app_context(); + let err = reload_peers(context.as_deref(), BUCKET) + .await + .expect_err("an unreachable peer must surface from the reload"); + assert!(err.contains("load_bucket_metadata"), "unexpected reload error: {err}"); + + // A second PUT moves the timestamp forward. + tokio::time::sleep(Duration::from_millis(1100)).await; + let response = SetBucketOnDemandMigrationHandler {} + .call(root_request(Method::PUT, config_uri(""), config_json(&source)), bucket_params(&router)) + .await + .expect("second PUT succeeds"); + let (_, body) = response_json(response).await; + let second_updated_at = body["updated_at"].as_str().expect("updated_at is set").to_string(); + assert!(second_updated_at > first_updated_at, "{second_updated_at} must follow {first_updated_at}"); + + // Read-only principal: GET and status answer, PUT and DELETE are 403. + let reader_get = GetBucketOnDemandMigrationHandler {} + .call( + admin_request(Method::GET, config_uri(""), Vec::new(), READER_ACCESS_KEY, READER_SECRET_KEY), + bucket_params(&router), + ) + .await + .expect("reader may GET"); + assert_eq!(reader_get.output.0, StatusCode::OK); + let reader_status = GetBucketOnDemandMigrationStatusHandler {} + .call( + admin_request(Method::GET, config_uri("/status"), Vec::new(), READER_ACCESS_KEY, READER_SECRET_KEY), + bucket_params(&router), + ) + .await + .expect("reader may read status"); + assert_eq!(reader_status.output.0, StatusCode::OK); + let err = SetBucketOnDemandMigrationHandler {} + .call( + admin_request(Method::PUT, config_uri(""), config_json(&source), READER_ACCESS_KEY, READER_SECRET_KEY), + bucket_params(&router), + ) + .await + .expect_err("reader must not PUT"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + let err = DeleteBucketOnDemandMigrationHandler {} + .call( + admin_request(Method::DELETE, config_uri(""), Vec::new(), READER_ACCESS_KEY, READER_SECRET_KEY), + bucket_params(&router), + ) + .await + .expect_err("reader must not DELETE"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + assert!(get_config().await.is_ok(), "denied writes must not change the config"); + + // DELETE clears, answers 204, and is idempotent. + let response = DeleteBucketOnDemandMigrationHandler {} + .call(root_request(Method::DELETE, config_uri(""), Vec::new()), bucket_params(&router)) + .await + .expect("DELETE succeeds"); + assert_eq!(response.output.0, StatusCode::NO_CONTENT); + let err = get_config().await.expect_err("config is gone after DELETE"); + assert_eq!(err.code(), &S3ErrorCode::Custom(ERR_CODE_NO_SUCH_CONFIGURATION.into())); + let metadata = metadata_sys::get(BUCKET).await.expect("bucket metadata"); + assert!(metadata.on_demand_migration_config_json.is_empty()); + assert!(metadata.on_demand_migration_config_updated_at > OffsetDateTime::UNIX_EPOCH); + assert_eq!( + status().await, + serde_json::json!({"configured": false, "enabled": false, "module_enabled": true}) + ); + + let response = DeleteBucketOnDemandMigrationHandler {} + .call(root_request(Method::DELETE, config_uri(""), Vec::new()), bucket_params(&router)) + .await + .expect("a second DELETE is idempotent"); + assert_eq!(response.output.0, StatusCode::NO_CONTENT); + }, + ) + .await; + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index f685f9d4a..e287525a2 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -36,9 +36,9 @@ mod route_registration_test; use handlers::{ account, audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler, extensions, heal, health, idp_compat, ilm_transition, inspect_archive, kms, mfa, module_switch, object_data_cache, - object_zip_download, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, - replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, usage_prefix, - user, + object_zip_download, oidc, on_demand_migration, plugins_catalog, plugins_instances, pools, profile_admin, + quota as quota_handler, rebalance, replication as replication_handler, scanner, site_replication, sts, system, table_catalog, + tier, tls_debug, usage_prefix, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -77,6 +77,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() quota_handler::register_quota_route(r)?; durability_handler::register_durability_route(r)?; + on_demand_migration::register_on_demand_migration_route(r)?; bucket_meta::register_bucket_meta_route(r)?; config_admin::register_config_route(r)?; scanner::register_scanner_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index c6d94a9f6..11adead27 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -38,6 +38,7 @@ const EXPORT_BUCKET_METADATA: AdminActionRef = AdminActionRef::new("ExportBucket const EXPORT_IAM: AdminActionRef = AdminActionRef::new("ExportIAMAction"); const FORCE_UNLOCK: AdminActionRef = AdminActionRef::new("ForceUnlockAdminAction"); const GET_BUCKET_TARGET: AdminActionRef = AdminActionRef::new("GetBucketTargetAction"); +const GET_BUCKET_ON_DEMAND_MIGRATION: AdminActionRef = AdminActionRef::new("GetBucketOnDemandMigrationAction"); const GET_GROUP: AdminActionRef = AdminActionRef::new("GetGroupAdminAction"); const GET_USER: AdminActionRef = AdminActionRef::new("GetUserAdminAction"); const GET_METRICS: AdminActionRef = AdminActionRef::new("GetMetricsAction"); @@ -88,6 +89,7 @@ const SERVER_INFO: AdminActionRef = AdminActionRef::new("ServerInfoAdminAction") const SERVER_UPDATE: AdminActionRef = AdminActionRef::new("ServerUpdateAdminAction"); const SET_BUCKET_QUOTA: AdminActionRef = AdminActionRef::new("SetBucketQuotaAdminAction"); const SET_BUCKET_TARGET: AdminActionRef = AdminActionRef::new("SetBucketTargetAction"); +const SET_BUCKET_ON_DEMAND_MIGRATION: AdminActionRef = AdminActionRef::new("SetBucketOnDemandMigrationAction"); const SET_TABLE: AdminActionRef = AdminActionRef::new("SetTableAction"); const SET_TABLE_BUCKET: AdminActionRef = AdminActionRef::new("SetTableBucketAction"); const SET_TABLE_LIFECYCLE: AdminActionRef = AdminActionRef::new("SetTableLifecycleAction"); @@ -390,6 +392,30 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ CONFIG_UPDATE, RouteRiskLevel::High, ), + admin( + HttpMethod::Put, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + SET_BUCKET_ON_DEMAND_MIGRATION, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + GET_BUCKET_ON_DEMAND_MIGRATION, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Delete, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + SET_BUCKET_ON_DEMAND_MIGRATION, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/on-demand-migration/{bucket}/status", + GET_BUCKET_ON_DEMAND_MIGRATION, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/rustfs/admin/export-bucket-metadata", @@ -2163,6 +2189,37 @@ mod tests { assert_action(HttpMethod::Get, "/rustfs/admin/v3/metrics", GET_METRICS); } + #[test] + fn route_policy_splits_on_demand_migration_into_set_and_get_actions() { + assert_action( + HttpMethod::Put, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + SET_BUCKET_ON_DEMAND_MIGRATION, + ); + assert_action( + HttpMethod::Delete, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + SET_BUCKET_ON_DEMAND_MIGRATION, + ); + assert_action( + HttpMethod::Get, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + GET_BUCKET_ON_DEMAND_MIGRATION, + ); + assert_action( + HttpMethod::Get, + "/rustfs/admin/v3/on-demand-migration/{bucket}/status", + GET_BUCKET_ON_DEMAND_MIGRATION, + ); + // Reads never require the write action, and the routes are not bucket-target routes. + assert_not_action( + HttpMethod::Get, + "/rustfs/admin/v3/on-demand-migration/{bucket}", + SET_BUCKET_ON_DEMAND_MIGRATION, + ); + assert_not_action(HttpMethod::Put, "/rustfs/admin/v3/on-demand-migration/{bucket}", SET_BUCKET_TARGET); + } + #[test] fn route_policy_requires_dedicated_inspect_action_for_encrypted_archive() { assert_action(HttpMethod::Post, "/rustfs/admin/v4/inspect/archive", INSPECT_DATA); diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 9ad2a06de..85f7c029a 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -240,6 +240,14 @@ fn expected_admin_route_matrix() -> Vec { admin_route_sample(Method::PUT, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"), admin_route_sample(Method::GET, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"), admin_route_sample(Method::DELETE, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"), + admin_route_sample(Method::PUT, "/v3/on-demand-migration/{bucket}", "/v3/on-demand-migration/test-bucket"), + admin_route_sample(Method::GET, "/v3/on-demand-migration/{bucket}", "/v3/on-demand-migration/test-bucket"), + admin_route_sample(Method::DELETE, "/v3/on-demand-migration/{bucket}", "/v3/on-demand-migration/test-bucket"), + admin_route_sample( + Method::GET, + "/v3/on-demand-migration/{bucket}/status", + "/v3/on-demand-migration/test-bucket/status", + ), admin_route(Method::GET, "/export-bucket-metadata"), admin_route(Method::GET, "/v3/export-bucket-metadata"), admin_route(Method::PUT, "/import-bucket-metadata"), @@ -1274,6 +1282,10 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/bucket-durability/test-bucket")); assert_route(&router, Method::GET, &admin_path("/v3/bucket-durability/test-bucket")); assert_route(&router, Method::DELETE, &admin_path("/v3/bucket-durability/test-bucket")); + assert_route(&router, Method::PUT, &admin_path("/v3/on-demand-migration/test-bucket")); + assert_route(&router, Method::GET, &admin_path("/v3/on-demand-migration/test-bucket")); + assert_route(&router, Method::DELETE, &admin_path("/v3/on-demand-migration/test-bucket")); + assert_route(&router, Method::GET, &admin_path("/v3/on-demand-migration/test-bucket/status")); assert_route(&router, Method::GET, &admin_path("/export-bucket-metadata")); assert_route(&router, Method::GET, &admin_path("/v3/export-bucket-metadata")); @@ -1403,6 +1415,10 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (Method::POST, compat_admin_alias_path("/v3/scanner/cycle-state/reset")), (Method::POST, compat_admin_alias_path("/v3/scanner/usage-state/reset")), (Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")), + (Method::PUT, compat_admin_alias_path("/v3/on-demand-migration/b")), + (Method::GET, compat_admin_alias_path("/v3/on-demand-migration/b")), + (Method::DELETE, compat_admin_alias_path("/v3/on-demand-migration/b")), + (Method::GET, compat_admin_alias_path("/v3/on-demand-migration/b/status")), ] { assert!( router.contains_compatible_route(method.clone(), &path), diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 3bf0e5b1f..7407b7dd8 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -20,8 +20,8 @@ use time::OffsetDateTime; mod ecstore_bucket { pub(crate) use crate::storage::storage_api::ecstore_bucket::{ - bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, object_lock, quota, replication, target, - utils, versioning, versioning_sys, + bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, object_lock, on_demand_migration, quota, + remote_s3_client, replication, target, utils, versioning, versioning_sys, }; } @@ -269,6 +269,7 @@ pub(crate) mod metadata { pub(crate) const BUCKET_TARGETS_FILE: &str = super::ecstore_bucket::metadata::BUCKET_TARGETS_FILE; pub(crate) const BUCKET_VERSIONING_CONFIG: &str = super::ecstore_bucket::metadata::BUCKET_VERSIONING_CONFIG; pub(crate) const BUCKET_DURABILITY_CONFIG: &str = super::ecstore_bucket::metadata::BUCKET_DURABILITY_CONFIG; + pub(crate) const BUCKET_ON_DEMAND_MIGRATION_CONFIG: &str = super::ecstore_bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG; pub(crate) const OBJECT_LOCK_CONFIG: &str = super::ecstore_bucket::metadata::OBJECT_LOCK_CONFIG; pub(crate) type BucketMetadata = super::ecstore_bucket::metadata::BucketMetadata; @@ -282,6 +283,29 @@ pub(crate) mod durability { pub(crate) type BucketDurabilityConfig = super::ecstore_bucket::durability::BucketDurabilityConfig; } +pub(crate) mod on_demand_migration { + pub(crate) type OnDemandMigrationConfig = super::ecstore_bucket::on_demand_migration::OnDemandMigrationConfig; + pub(crate) type OnDemandMigrationConfigError = super::ecstore_bucket::on_demand_migration::OnDemandMigrationConfigError; + pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle; + pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider; + pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>; + + pub(crate) mod source_client { + pub(crate) type SourceClient = super::super::ecstore_bucket::on_demand_migration::source_client::SourceClient; + pub(crate) type SourceClientSpec = super::super::ecstore_bucket::on_demand_migration::source_client::SourceClientSpec; + pub(crate) type SourceError = super::super::ecstore_bucket::on_demand_migration::source_client::SourceError; + pub(crate) type SourceProbe = super::super::ecstore_bucket::on_demand_migration::source_client::SourceProbe; + pub(crate) type SourceProvider = super::super::ecstore_bucket::on_demand_migration::source_client::SourceProvider; + pub(crate) type SourceTimeouts = super::super::ecstore_bucket::on_demand_migration::source_client::SourceTimeouts; + } +} + +pub(crate) mod remote_s3_client { + pub(crate) type PathStyle = super::ecstore_bucket::remote_s3_client::PathStyle; + pub(crate) type RemoteCredentials = super::ecstore_bucket::remote_s3_client::RemoteCredentials; + pub(crate) type RemoteS3ClientError = super::ecstore_bucket::remote_s3_client::RemoteS3ClientError; +} + pub(crate) mod metadata_sys { use std::sync::Arc; @@ -417,6 +441,12 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::get_durability_config(bucket).await } + pub(crate) async fn get_on_demand_migration_config( + bucket: &str, + ) -> Result> { + super::ecstore_bucket::metadata_sys::get_on_demand_migration_config(bucket).await + } + pub(crate) async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> { super::ecstore_bucket::metadata_sys::get_quota_config(bucket).await } @@ -869,7 +899,9 @@ pub(crate) mod bucket { pub(crate) use super::lifecycle; pub(crate) use super::metadata; pub(crate) use super::metadata_sys; + pub(crate) use super::on_demand_migration; pub(crate) use super::quota; + pub(crate) use super::remote_s3_client; pub(crate) use super::replication; pub(crate) use super::target; pub(crate) use super::versioning_sys; @@ -969,7 +1001,7 @@ pub(crate) mod runtime { } pub(crate) mod s3 { - pub(crate) use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header}; + pub(crate) use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, auth, header}; /// Build an `S3Error` without reaching for the `s3s` error macro. /// diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 066780d3b..e503cfbe5 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -408,7 +408,7 @@ pub(crate) mod ecstore_bucket { pub(crate) use rustfs_ecstore::api::bucket::lifecycle::tier_delete_journal::test_util::install_all_v6_fleet_capability_proof; pub(crate) use rustfs_ecstore::api::bucket::{ bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, on_demand_migration, - policy_sys, replication, tagging, target, utils, + policy_sys, remote_s3_client, replication, tagging, target, utils, }; pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys}; }