mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-05 19:55:37 +00:00
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/<version> 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
This commit is contained in:
Zhengchao An
@@ -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"}
|
||||
@@ -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}}
|
||||
@@ -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"}}
|
||||
@@ -0,0 +1 @@
|
||||
{"configured":true,"enabled":true,"module_enabled":false}
|
||||
+19
-135
@@ -412,7 +412,7 @@ impl AdminClient {
|
||||
self.execute(request).await
|
||||
}
|
||||
|
||||
fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result<reqwest::Url, AdminClientError> {
|
||||
pub(crate) fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result<reqwest::Url, AdminClientError> {
|
||||
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<T: for<'de> Deserialize<'de>>(&self, request: reqwest::Request) -> Result<T, AdminClientError> {
|
||||
pub(crate) async fn execute<T: for<'de> Deserialize<'de>>(&self, request: reqwest::Request) -> Result<T, AdminClientError> {
|
||||
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<String> {
|
||||
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<Mutex>`. Deliberately dependency-free —
|
||||
/// the assertions only need the raw request bytes.
|
||||
struct TestServer {
|
||||
addr: std::net::SocketAddr,
|
||||
requests: Arc<Mutex<Vec<RecordedRequest>>>,
|
||||
}
|
||||
|
||||
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<Mutex<Vec<RecordedRequest>>> = 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<usize> {
|
||||
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<RecordedRequest> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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::*;
|
||||
|
||||
@@ -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<String>,
|
||||
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<OnDemandMigrationCredentials>,
|
||||
#[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<String>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
|
||||
pub struct OnDemandMigrationFilter {
|
||||
#[serde(default)]
|
||||
pub prefix: Option<String>,
|
||||
#[serde(default)]
|
||||
pub source_prefix: Option<String>,
|
||||
}
|
||||
|
||||
#[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<u64>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
/// `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<String>,
|
||||
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<OnDemandMigrationSetResponse, AdminClientError> {
|
||||
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<OnDemandMigrationGetResponse, AdminClientError> {
|
||||
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<OnDemandMigrationStatus, AdminClientError> {
|
||||
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<T: Serialize + for<'de> 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:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String> {
|
||||
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<Mutex>`. Deliberately dependency-free —
|
||||
/// the assertions only need the raw request bytes.
|
||||
pub(crate) struct TestServer {
|
||||
pub(crate) addr: std::net::SocketAddr,
|
||||
requests: Arc<Mutex<Vec<RecordedRequest>>>,
|
||||
}
|
||||
|
||||
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<Mutex<Vec<RecordedRequest>>> = 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<usize> {
|
||||
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<RecordedRequest> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user