mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-02 22:27:45 +08:00
test(e2e): extend fake S3 target as an on-demand migration source (#7068)
* test(e2e): extend fake S3 target as an on-demand migration source Add ListObjectsV2 paging, Range GET/HEAD, unversioned buckets, standard and user metadata replay, ResponseStatus/TruncateBodyAt/Stall fault actions, Range/User-Agent/prefix/continuation-token journal fields, count_requests, direct seeding, and a configurable object cap to the programmable fake S3 target, and add the on_demand_migration e2e harness (OdmTestEnv, admin wrappers, source seeding, local-state assertions, second RustFS source) with its self-test. * test(ci): refresh darwin e2e-full selection for ODM harness
This commit is contained in:
Zhengchao An
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=ef914ec0b8daa9c2c5e52f501d339914662f42d6f6ed9d33877d56b97adf16f9
|
||||
sha256-darwin=9dccb0cd537cf79ae70c1c20e8281d36d03f2f09f81142a5341e26e3dc18709d
|
||||
sha256-linux=a8a816d7bb0e7cb5632b1863b33794bcb9fc7e765f150aa5e1bf16518e28dfb4
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
# Programmable fake S3 target
|
||||
|
||||
This module is the shared failure-injection boundary for replication end-to-end tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it.
|
||||
This module is the shared failure-injection boundary for replication end-to-end tests and the programmable external source for on-demand-migration (ODM) tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it.
|
||||
|
||||
`FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script.
|
||||
|
||||
Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
|
||||
Supported data operations are HeadBucket, GetBucketVersioning, ListObjectsV2, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets created with `create_bucket` are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
|
||||
|
||||
Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions. Each record also journals a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
|
||||
`create_bucket_with_mode(name, BucketMode::Unversioned)` models a plain migration source: PUT overwrites in place, DELETE removes the key without a delete marker, GetBucketVersioning reports no status, and no `x-amz-version-id` is returned by PUT, GET, HEAD, tagging, or multipart completion. The only `versionId` such a bucket accepts is `null`; any other value is rejected with `InvalidArgument`. The mode is fixed at creation.
|
||||
|
||||
The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type at 1 KiB. A PUT or uploaded part is capped at 64 MiB; a completed multipart object and all stored object/part data are capped at 128 MiB. Body drain, body-permit waits, delay, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.
|
||||
ListObjectsV2 lists current versions only (a key whose newest version is a delete marker is hidden) in byte order and supports `prefix`, `delimiter`, `max-keys` (clamped to 1000), `start-after`, and `continuation-token`; common prefixes count toward `max-keys`, `IsTruncated` / `NextContinuationToken` / `KeyCount` follow S3, and continuation tokens are opaque. `encoding-type` and `fetch-owner` are accepted but ignored, and ListObjects (v1) is not implemented. GET and HEAD honor `Range` in the `bytes=first-last`, `bytes=first-`, and `bytes=-suffix` forms with a 206 status, exact `Content-Range`, and `Accept-Ranges: bytes`; unsatisfiable ranges answer 416 `InvalidRange` with `Content-Range: bytes */<length>`. PUT and CreateMultipartUpload accept `Content-Type`, `Content-Encoding`, `Content-Disposition`, `Content-Language`, `Cache-Control`, `Expires`, and `x-amz-meta-*` (names stored lowercased), and HEAD/GET replay them verbatim together with `Last-Modified` and the ETag (hex MD5 for single PUTs, `<md5-of-part-md5s>-<parts>` for multipart objects). `put_seed_object` stores an object directly, bypassing the wire, the fault script, and the journal, so a source can be seeded without polluting the assertions a scenario later makes.
|
||||
|
||||
Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
|
||||
|
||||
The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type and each standard object header at 1 KiB. By default a PUT or uploaded part is capped at 64 MiB and a completed multipart object and all stored object/part data are capped at 128 MiB; `FakeS3Target::start_with_options(FakeS3TargetOptions { max_object_bytes })` raises the object cap up to 256 MiB, and the total budget then becomes twice the object cap (never below 128 MiB). Body drain, body-permit waits, delay, stall, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -23,10 +23,17 @@ pub mod common;
|
||||
#[cfg(test)]
|
||||
pub mod chaos;
|
||||
|
||||
// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8).
|
||||
// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8)
|
||||
// and on-demand-migration source scenarios (backlog#2151).
|
||||
#[cfg(test)]
|
||||
pub mod fake_s3_target;
|
||||
|
||||
// On-demand migration (backlog#2147): shared two-server environment, admin
|
||||
// wrappers, and the harness self-test (backlog#2151). Behavior scenarios are
|
||||
// added by later ODM tasks.
|
||||
#[cfg(test)]
|
||||
pub mod on_demand_migration;
|
||||
|
||||
// Socket-level network fault-injection proxy for black-box cluster tests
|
||||
// (backlog#1325 network fault-injection block): latency / blackhole / one-way
|
||||
// partition on the wire between nodes. Serves #1312/#1319 (lock-plane one-way
|
||||
|
||||
452
crates/e2e_test/src/on_demand_migration/common.rs
Normal file
452
crates/e2e_test/src/on_demand_migration/common.rs
Normal file
@@ -0,0 +1,452 @@
|
||||
// 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.
|
||||
|
||||
//! Shared environment for on-demand migration (ODM) end-to-end tests.
|
||||
//!
|
||||
//! [`OdmTestEnv`] pairs one RustFS server under test with one in-process
|
||||
//! programmable S3 source ([`FakeS3Target`]). Admin calls target the route
|
||||
//! convention fixed by the tracking plan
|
||||
//! (`/rustfs/admin/v3/on-demand-migration/{bucket}`, JSON bodies); the
|
||||
//! server side lands with ODM-07, so until then the wrappers compile but are
|
||||
//! not exercised by the harness self-test.
|
||||
|
||||
use crate::common::{RustFSTestEnvironment, signed_request};
|
||||
use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FakeS3TargetOptions, SeedMetadata};
|
||||
use aws_config::retry::RetryConfig;
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::config::{Credentials, Region};
|
||||
use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
|
||||
use bytes::Bytes;
|
||||
use serde::Serialize;
|
||||
use std::fmt;
|
||||
|
||||
pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
|
||||
|
||||
/// Module switch the server reads at startup (`false` before GA). The harness
|
||||
/// turns it on so scenario tests exercise the feature without repeating it.
|
||||
pub const ODM_MODULE_SWITCH_ENV: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED";
|
||||
/// Admin route prefix; the bucket name is appended as one path segment.
|
||||
pub const ODM_ADMIN_ROUTE: &str = "/rustfs/admin/v3/on-demand-migration";
|
||||
/// Region the fake source is addressed with (it accepts any SigV4 region).
|
||||
pub const FAKE_SOURCE_REGION: &str = "us-east-1";
|
||||
|
||||
/// Wire form of the bucket-level ODM configuration (ODM-01 model). Every
|
||||
/// field is public so a scenario can tweak one knob and serialize the rest
|
||||
/// with the documented defaults.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct OdmSourceSpec {
|
||||
pub version: u32,
|
||||
pub enabled: bool,
|
||||
pub source: OdmSource,
|
||||
pub filter: OdmFilter,
|
||||
pub policy: OdmPolicy,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct OdmSource {
|
||||
pub provider: String,
|
||||
pub endpoint: String,
|
||||
pub region: String,
|
||||
pub bucket: String,
|
||||
pub path_style: String,
|
||||
pub credentials: Option<OdmCredentials>,
|
||||
pub tls: OdmTls,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
pub struct OdmCredentials {
|
||||
pub access_key: String,
|
||||
pub secret_key: String,
|
||||
pub session_token: Option<String>,
|
||||
}
|
||||
|
||||
impl fmt::Debug for OdmCredentials {
|
||||
/// Test logs are captured into CI artifacts; keep the secret out of them.
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("OdmCredentials")
|
||||
.field("access_key", &self.access_key)
|
||||
.field("secret_key", &"REDACTED")
|
||||
.field("session_token", &self.session_token.as_ref().map(|_| "REDACTED"))
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize)]
|
||||
pub struct OdmTls {
|
||||
pub skip_verify: bool,
|
||||
pub ca_cert_pem: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize)]
|
||||
pub struct OdmFilter {
|
||||
pub prefix: Option<String>,
|
||||
pub source_prefix: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct OdmPolicy {
|
||||
pub head: String,
|
||||
pub range_get: String,
|
||||
pub source_error: String,
|
||||
pub respect_local_delete_marker: bool,
|
||||
pub preserve_etag: bool,
|
||||
pub copy_tags: bool,
|
||||
pub emit_events: bool,
|
||||
pub negative_cache_ttl_secs: u64,
|
||||
pub inline_max_bytes: u64,
|
||||
pub multipart_part_size_bytes: u64,
|
||||
pub max_concurrent_pulls: u32,
|
||||
pub pull_queue_capacity: u32,
|
||||
pub source_timeout: OdmSourceTimeout,
|
||||
pub bandwidth_limit_bytes_per_sec: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct OdmSourceTimeout {
|
||||
pub connect_ms: u64,
|
||||
pub first_byte_ms: u64,
|
||||
pub idle_ms: u64,
|
||||
}
|
||||
|
||||
impl Default for OdmPolicy {
|
||||
/// The ODM-01 defaults verbatim.
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
head: "proxy".to_string(),
|
||||
range_get: "serve_and_backfill".to_string(),
|
||||
source_error: "propagate".to_string(),
|
||||
respect_local_delete_marker: true,
|
||||
preserve_etag: true,
|
||||
copy_tags: false,
|
||||
emit_events: true,
|
||||
negative_cache_ttl_secs: 30,
|
||||
inline_max_bytes: 16 * 1024 * 1024,
|
||||
multipart_part_size_bytes: 64 * 1024 * 1024,
|
||||
max_concurrent_pulls: 8,
|
||||
pull_queue_capacity: 1024,
|
||||
source_timeout: OdmSourceTimeout {
|
||||
connect_ms: 5_000,
|
||||
first_byte_ms: 15_000,
|
||||
idle_ms: 30_000,
|
||||
},
|
||||
bandwidth_limit_bytes_per_sec: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl OdmSourceSpec {
|
||||
/// Enabled configuration pointing at a bucket on the fake source with the
|
||||
/// fixture credentials, path-style addressing, and default policy.
|
||||
pub fn for_fake_source(source: &FakeS3Target, source_bucket: impl Into<String>) -> Self {
|
||||
Self::new(
|
||||
"s3",
|
||||
source.endpoint(),
|
||||
FAKE_SOURCE_REGION,
|
||||
source_bucket,
|
||||
FAKE_ACCESS_KEY,
|
||||
FAKE_SECRET_KEY,
|
||||
)
|
||||
}
|
||||
|
||||
/// Enabled configuration pointing at a bucket on a second RustFS server
|
||||
/// (see [`start_source_rustfs`]).
|
||||
pub fn for_rustfs_source(source: &RustFSTestEnvironment, source_bucket: impl Into<String>) -> Self {
|
||||
Self::new(
|
||||
"rustfs",
|
||||
&source.url,
|
||||
FAKE_SOURCE_REGION,
|
||||
source_bucket,
|
||||
&source.access_key,
|
||||
&source.secret_key,
|
||||
)
|
||||
}
|
||||
|
||||
fn new(
|
||||
provider: &str,
|
||||
endpoint: &str,
|
||||
region: &str,
|
||||
source_bucket: impl Into<String>,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
) -> Self {
|
||||
Self {
|
||||
version: 1,
|
||||
enabled: true,
|
||||
source: OdmSource {
|
||||
provider: provider.to_string(),
|
||||
endpoint: endpoint.to_string(),
|
||||
region: region.to_string(),
|
||||
bucket: source_bucket.into(),
|
||||
path_style: "path".to_string(),
|
||||
credentials: Some(OdmCredentials {
|
||||
access_key: access_key.to_string(),
|
||||
secret_key: secret_key.to_string(),
|
||||
session_token: None,
|
||||
}),
|
||||
tls: OdmTls::default(),
|
||||
},
|
||||
filter: OdmFilter::default(),
|
||||
policy: OdmPolicy::default(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn to_json(&self) -> serde_json::Value {
|
||||
serde_json::to_value(self).expect("ODM source spec serializes")
|
||||
}
|
||||
}
|
||||
|
||||
/// Backfill job control (ODM-12 route shape).
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum BackfillOp {
|
||||
Start(BackfillRequest),
|
||||
Cancel,
|
||||
Status,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize)]
|
||||
pub struct BackfillRequest {
|
||||
pub prefix: Option<String>,
|
||||
pub skip_existing: Option<String>,
|
||||
pub dry_run: bool,
|
||||
}
|
||||
|
||||
/// Status plus raw body of an admin call, so a scenario can assert on the
|
||||
/// HTTP status first and only then parse the JSON.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AdminResponse {
|
||||
pub status: u16,
|
||||
pub body: String,
|
||||
}
|
||||
|
||||
impl AdminResponse {
|
||||
pub fn json(&self) -> Result<serde_json::Value, BoxError> {
|
||||
Ok(serde_json::from_str(&self.body)?)
|
||||
}
|
||||
}
|
||||
|
||||
/// One object to seed into the source.
|
||||
#[derive(Clone)]
|
||||
pub struct SeedObject {
|
||||
pub key: String,
|
||||
pub body: Bytes,
|
||||
pub metadata: SeedMetadata,
|
||||
}
|
||||
|
||||
impl SeedObject {
|
||||
pub fn new(key: impl Into<String>, body: impl Into<Bytes>) -> Self {
|
||||
Self {
|
||||
key: key.into(),
|
||||
body: body.into(),
|
||||
metadata: SeedMetadata::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_metadata(mut self, metadata: SeedMetadata) -> Self {
|
||||
self.metadata = metadata;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
/// RustFS under test plus its fake S3 source.
|
||||
pub struct OdmTestEnv {
|
||||
pub rustfs: RustFSTestEnvironment,
|
||||
pub source: FakeS3Target,
|
||||
/// S3 client for the RustFS under test.
|
||||
pub client: Client,
|
||||
}
|
||||
|
||||
impl OdmTestEnv {
|
||||
/// Start a fake source with default limits and a RustFS server with the
|
||||
/// ODM module switch enabled.
|
||||
pub async fn start() -> Result<Self, BoxError> {
|
||||
Self::start_with_options(FakeS3TargetOptions::default()).await
|
||||
}
|
||||
|
||||
pub async fn start_with_options(options: FakeS3TargetOptions) -> Result<Self, BoxError> {
|
||||
let source = FakeS3Target::start_with_options(options).await?;
|
||||
let mut rustfs = RustFSTestEnvironment::new().await?;
|
||||
rustfs
|
||||
.start_rustfs_server_with_env(vec![], &[(ODM_MODULE_SWITCH_ENV, "true")])
|
||||
.await?;
|
||||
let client = rustfs.create_s3_client();
|
||||
Ok(Self { rustfs, source, client })
|
||||
}
|
||||
|
||||
/// S3 client addressing the fake source directly, for assertions on the
|
||||
/// source's own state. Retries are off so a scripted fault is consumed by
|
||||
/// exactly the request the test issued.
|
||||
pub fn source_client(&self) -> Client {
|
||||
fake_source_client(&self.source)
|
||||
}
|
||||
|
||||
/// Enabled ODM configuration for `source_bucket` on the fake source.
|
||||
pub fn fake_source_spec(&self, source_bucket: impl Into<String>) -> OdmSourceSpec {
|
||||
OdmSourceSpec::for_fake_source(&self.source, source_bucket)
|
||||
}
|
||||
|
||||
/// `PUT /rustfs/admin/v3/on-demand-migration/{bucket}` with the JSON spec.
|
||||
pub async fn configure_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result<AdminResponse, BoxError> {
|
||||
self.admin(http::Method::PUT, &format!("/{bucket}"), Some(spec.to_json()))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Same as [`Self::configure_source`] with `dry-run=true`: validate and
|
||||
/// probe without persisting.
|
||||
pub async fn validate_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result<AdminResponse, BoxError> {
|
||||
self.admin(http::Method::PUT, &format!("/{bucket}?dry-run=true"), Some(spec.to_json()))
|
||||
.await
|
||||
}
|
||||
|
||||
/// `GET .../{bucket}`: redacted configuration, 404 when unconfigured.
|
||||
pub async fn get_config(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
|
||||
self.admin(http::Method::GET, &format!("/{bucket}"), None).await
|
||||
}
|
||||
|
||||
/// `DELETE .../{bucket}`: remove the configuration (idempotent).
|
||||
pub async fn disable(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
|
||||
self.admin(http::Method::DELETE, &format!("/{bucket}"), None).await
|
||||
}
|
||||
|
||||
/// `GET .../{bucket}/status`: runtime snapshot.
|
||||
pub async fn status(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
|
||||
self.admin(http::Method::GET, &format!("/{bucket}/status"), None).await
|
||||
}
|
||||
|
||||
/// Backfill control: `POST .../{bucket}/backfill?op=start|cancel` or
|
||||
/// `GET .../{bucket}/backfill` for the checkpoint.
|
||||
pub async fn backfill(&self, bucket: &str, op: BackfillOp) -> Result<AdminResponse, BoxError> {
|
||||
match op {
|
||||
BackfillOp::Start(request) => {
|
||||
self.admin(
|
||||
http::Method::POST,
|
||||
&format!("/{bucket}/backfill?op=start"),
|
||||
Some(serde_json::to_value(request)?),
|
||||
)
|
||||
.await
|
||||
}
|
||||
BackfillOp::Cancel => {
|
||||
self.admin(http::Method::POST, &format!("/{bucket}/backfill?op=cancel"), None)
|
||||
.await
|
||||
}
|
||||
BackfillOp::Status => self.admin(http::Method::GET, &format!("/{bucket}/backfill"), None).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn admin(
|
||||
&self,
|
||||
method: http::Method,
|
||||
path_and_query: &str,
|
||||
body: Option<serde_json::Value>,
|
||||
) -> Result<AdminResponse, BoxError> {
|
||||
let url = format!("{}{ODM_ADMIN_ROUTE}{path_and_query}", self.rustfs.url);
|
||||
let body = body.map(|value| serde_json::to_vec(&value)).transpose()?;
|
||||
let content_type = body.is_some().then_some("application/json");
|
||||
let response = signed_request(method, &url, &self.rustfs.access_key, &self.rustfs.secret_key, body, content_type).await?;
|
||||
Ok(AdminResponse {
|
||||
status: response.status().as_u16(),
|
||||
body: response.text().await?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Store objects directly in the fake source (no wire traffic, no journal
|
||||
/// entries). Returns the ETags in input order.
|
||||
pub fn seed_source(&self, source_bucket: &str, objects: &[SeedObject]) -> Vec<String> {
|
||||
objects
|
||||
.iter()
|
||||
.map(|object| {
|
||||
self.source
|
||||
.put_seed_object(source_bucket, object.key.clone(), object.body.clone(), &object.metadata)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Whether `key` is listed by the RustFS under test. Listing is served from
|
||||
/// local state only, so this does not trigger a migration the way GET or
|
||||
/// HEAD would.
|
||||
pub async fn local_key_listed(&self, bucket: &str, key: &str) -> Result<bool, BoxError> {
|
||||
let listed = self
|
||||
.client
|
||||
.list_objects_v2()
|
||||
.bucket(bucket)
|
||||
.prefix(key)
|
||||
.max_keys(1)
|
||||
.send()
|
||||
.await?;
|
||||
Ok(listed.contents().iter().any(|object| object.key() == Some(key)))
|
||||
}
|
||||
|
||||
/// Panics unless `key` is stored locally with exactly `expected` bytes.
|
||||
/// Presence is checked through listing first so a missing object fails
|
||||
/// here instead of being pulled from the source by the GET.
|
||||
pub async fn assert_local_present(&self, bucket: &str, key: &str, expected: &[u8]) {
|
||||
assert!(
|
||||
self.local_key_listed(bucket, key)
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")),
|
||||
"{bucket}/{key} must be present locally"
|
||||
);
|
||||
let body = self
|
||||
.client
|
||||
.get_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("GET {bucket}/{key} failed: {error}"))
|
||||
.body
|
||||
.collect()
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("reading {bucket}/{key} failed: {error}"))
|
||||
.into_bytes();
|
||||
assert_eq!(body.as_ref(), expected, "{bucket}/{key} local content mismatch");
|
||||
}
|
||||
|
||||
/// Panics if `key` is listed locally.
|
||||
pub async fn assert_local_absent(&self, bucket: &str, key: &str) {
|
||||
assert!(
|
||||
!self
|
||||
.local_key_listed(bucket, key)
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")),
|
||||
"{bucket}/{key} must be absent locally"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// S3 client for the fake source with retries disabled (see
|
||||
/// [`OdmTestEnv::source_client`]).
|
||||
pub fn fake_source_client(source: &FakeS3Target) -> Client {
|
||||
let credentials = Credentials::new(FAKE_ACCESS_KEY, FAKE_SECRET_KEY, None, None, "odm-fake-source");
|
||||
Client::from_conf(
|
||||
aws_sdk_s3::Config::builder()
|
||||
.credentials_provider(credentials)
|
||||
.region(Region::new(FAKE_SOURCE_REGION))
|
||||
.endpoint_url(source.endpoint())
|
||||
.force_path_style(true)
|
||||
.behavior_version_latest()
|
||||
.retry_config(RetryConfig::standard().with_max_attempts(1))
|
||||
.http_client(SmithyHttpClientBuilder::new().build_http())
|
||||
.build(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Start a second, fully independent RustFS process (own port, data
|
||||
/// directory, and default credentials) to act as a real S3 source. It is
|
||||
/// spawned the same way `reliant::tiering` starts its cold tier; the process
|
||||
/// is stopped and its directory removed when the returned environment drops.
|
||||
pub async fn start_source_rustfs() -> Result<RustFSTestEnvironment, BoxError> {
|
||||
let mut source = RustFSTestEnvironment::new().await?;
|
||||
source.start_rustfs_server_without_cleanup(vec![]).await?;
|
||||
Ok(source)
|
||||
}
|
||||
606
crates/e2e_test/src/on_demand_migration/harness_self_test.rs
Normal file
606
crates/e2e_test/src/on_demand_migration/harness_self_test.rs
Normal file
@@ -0,0 +1,606 @@
|
||||
// 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.
|
||||
|
||||
//! Self-test of the ODM harness (rustfs/backlog#2151): the fake source's
|
||||
//! migration-facing surface (ListObjectsV2 paging, `Range`, unversioned
|
||||
//! buckets, metadata replay, fault actions) and the two-server environment.
|
||||
//! No ODM behavior is exercised here.
|
||||
|
||||
use super::common::{OdmTestEnv, SeedObject, fake_source_client, start_source_rustfs};
|
||||
use crate::fake_s3_target::{BucketMode, FakeS3Target, FakeS3TargetOptions, FaultAction, Operation, SeedMetadata};
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::error::ProvideErrorMetadata;
|
||||
use aws_sdk_s3::primitives::{ByteStream, DateTime};
|
||||
use bytes::Bytes;
|
||||
use std::collections::BTreeSet;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
|
||||
|
||||
const SOURCE_BUCKET: &str = "odm-source";
|
||||
|
||||
/// Position-dependent payload so a misaligned range read is caught.
|
||||
fn payload(len: usize) -> Bytes {
|
||||
(0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
|
||||
}
|
||||
|
||||
async fn fake_source() -> Result<(FakeS3Target, Client), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let source = FakeS3Target::start().await?;
|
||||
source.create_bucket(SOURCE_BUCKET);
|
||||
let client = fake_source_client(&source);
|
||||
Ok((source, client))
|
||||
}
|
||||
|
||||
/// Full ListObjectsV2 traversal. Returns `(keys, common prefixes, pages)` and
|
||||
/// checks the page shape on the way: every page except the last is full and
|
||||
/// truncated, the last carries no continuation token.
|
||||
async fn list_all(
|
||||
client: &Client,
|
||||
prefix: Option<&str>,
|
||||
delimiter: Option<&str>,
|
||||
start_after: Option<&str>,
|
||||
max_keys: i32,
|
||||
) -> Result<(Vec<String>, Vec<String>, usize), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let mut keys = Vec::new();
|
||||
let mut prefixes = Vec::new();
|
||||
let mut pages = 0usize;
|
||||
let mut token: Option<String> = None;
|
||||
loop {
|
||||
let page = client
|
||||
.list_objects_v2()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.set_prefix(prefix.map(str::to_string))
|
||||
.set_delimiter(delimiter.map(str::to_string))
|
||||
.set_start_after(start_after.map(str::to_string))
|
||||
.max_keys(max_keys)
|
||||
.set_continuation_token(token.clone())
|
||||
.send()
|
||||
.await?;
|
||||
pages += 1;
|
||||
let page_keys: Vec<String> = page
|
||||
.contents()
|
||||
.iter()
|
||||
.filter_map(|object| object.key().map(str::to_string))
|
||||
.collect();
|
||||
let page_prefixes: Vec<String> = page
|
||||
.common_prefixes()
|
||||
.iter()
|
||||
.filter_map(|common| common.prefix().map(str::to_string))
|
||||
.collect();
|
||||
let entries = page_keys.len() + page_prefixes.len();
|
||||
assert_eq!(page.key_count(), Some(entries as i32), "KeyCount must count keys and prefixes");
|
||||
assert_eq!(page.continuation_token(), token.as_deref(), "the request token must be echoed");
|
||||
keys.extend(page_keys);
|
||||
prefixes.extend(page_prefixes);
|
||||
if page.is_truncated() == Some(true) {
|
||||
assert_eq!(entries as i32, max_keys, "every truncated page must be full");
|
||||
token = Some(
|
||||
page.next_continuation_token()
|
||||
.expect("truncated page must carry a continuation token")
|
||||
.to_string(),
|
||||
);
|
||||
} else {
|
||||
assert!(page.next_continuation_token().is_none(), "final page must not carry a token");
|
||||
return Ok((keys, prefixes, pages));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_list_objects_v2_paginates_with_delimiter() -> TestResult {
|
||||
let (source, client) = fake_source().await?;
|
||||
let mut expected_keys = BTreeSet::new();
|
||||
for directory in 0..30 {
|
||||
for file in 0..30 {
|
||||
expected_keys.insert(format!("d{directory:02}/k{file:03}"));
|
||||
}
|
||||
}
|
||||
for index in 0..100 {
|
||||
expected_keys.insert(format!("top-{index:03}"));
|
||||
}
|
||||
assert_eq!(expected_keys.len(), 1000);
|
||||
for key in &expected_keys {
|
||||
source.put_seed_object(SOURCE_BUCKET, key.clone(), Bytes::from(key.clone()), &SeedMetadata::new());
|
||||
}
|
||||
// A key whose current version is a delete marker must stay hidden.
|
||||
client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("hidden/marker")
|
||||
.body(ByteStream::from_static(b"gone"))
|
||||
.send()
|
||||
.await?;
|
||||
client
|
||||
.delete_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("hidden/marker")
|
||||
.send()
|
||||
.await?;
|
||||
let expected_sorted: Vec<String> = expected_keys.iter().cloned().collect();
|
||||
let expected_prefixes: Vec<String> = (0..30).map(|directory| format!("d{directory:02}/")).collect();
|
||||
let expected_top: Vec<String> = (0..100).map(|index| format!("top-{index:03}")).collect();
|
||||
|
||||
// Flat traversal in byte order, 1000 keys in pages of 7.
|
||||
let (keys, prefixes, pages) = list_all(&client, None, None, None, 7).await?;
|
||||
assert_eq!(keys, expected_sorted);
|
||||
assert!(prefixes.is_empty());
|
||||
assert_eq!(pages, 143);
|
||||
|
||||
// Delimiter folding: 30 common prefixes then 100 top-level keys, pages of 7.
|
||||
let (keys, prefixes, pages) = list_all(&client, None, Some("/"), None, 7).await?;
|
||||
assert_eq!(prefixes, expected_prefixes);
|
||||
assert_eq!(keys, expected_top);
|
||||
assert_eq!(pages, 19);
|
||||
|
||||
// Empty prefix equals no prefix.
|
||||
let (keys, _, _) = list_all(&client, Some(""), None, None, 1000).await?;
|
||||
assert_eq!(keys, expected_sorted);
|
||||
|
||||
// No match: empty, not truncated, no token.
|
||||
let (keys, prefixes, pages) = list_all(&client, Some("zzz/"), Some("/"), None, 7).await?;
|
||||
assert!(keys.is_empty() && prefixes.is_empty());
|
||||
assert_eq!(pages, 1);
|
||||
let (keys, _, _) = list_all(&client, Some("hidden/"), None, None, 7).await?;
|
||||
assert!(keys.is_empty(), "a current delete marker must hide its key");
|
||||
|
||||
// Exact page boundary: 30 keys under one directory, max-keys=30 -> one
|
||||
// untruncated page.
|
||||
let (keys, prefixes, pages) = list_all(&client, Some("d05/"), Some("/"), None, 30).await?;
|
||||
assert_eq!(keys.len(), 30);
|
||||
assert!(prefixes.is_empty());
|
||||
assert_eq!(pages, 1);
|
||||
|
||||
// start-after skips keys at or before the marker.
|
||||
let (keys, _, _) = list_all(&client, None, None, Some("top-097"), 1000).await?;
|
||||
assert_eq!(keys, ["top-098", "top-099"]);
|
||||
|
||||
// max-keys is clamped to 1000; exactly 1000 keys fit in one page.
|
||||
let (keys, _, pages) = list_all(&client, None, None, None, 5000).await?;
|
||||
assert_eq!(keys.len(), 1000);
|
||||
assert_eq!(pages, 1);
|
||||
|
||||
let listings: Vec<_> = source
|
||||
.requests()
|
||||
.into_iter()
|
||||
.filter(|record| record.operation == Operation::ListObjectsV2)
|
||||
.collect();
|
||||
assert!(listings.len() >= 143 + 19);
|
||||
assert!(listings.iter().any(|record| record.prefix.as_deref() == Some("d05/")));
|
||||
assert!(
|
||||
listings.iter().any(|record| record.continuation_token.is_some()),
|
||||
"resumed pages must journal their continuation token"
|
||||
);
|
||||
assert!(listings.iter().all(|record| record.user_agent.is_some()));
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_range_get_variants_and_416() -> TestResult {
|
||||
let (source, client) = fake_source().await?;
|
||||
let body = payload(1000);
|
||||
source.put_seed_object(SOURCE_BUCKET, "ranged", body.clone(), &SeedMetadata::new());
|
||||
|
||||
for (range, expected_range, expected_slice) in [
|
||||
("bytes=10-19", "bytes 10-19/1000", &body[10..20]),
|
||||
("bytes=990-", "bytes 990-999/1000", &body[990..]),
|
||||
("bytes=-5", "bytes 995-999/1000", &body[995..]),
|
||||
("bytes=0-5000", "bytes 0-999/1000", &body[..]),
|
||||
] {
|
||||
let output = client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("ranged")
|
||||
.range(range)
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(output.content_range(), Some(expected_range), "{range}");
|
||||
assert_eq!(output.accept_ranges(), Some("bytes"), "{range}");
|
||||
assert_eq!(output.content_length(), Some(expected_slice.len() as i64), "{range}");
|
||||
let collected = output.body.collect().await?.into_bytes();
|
||||
assert_eq!(collected.as_ref(), expected_slice, "{range}");
|
||||
}
|
||||
let head = client
|
||||
.head_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("ranged")
|
||||
.range("bytes=10-19")
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(head.content_range(), Some("bytes 10-19/1000"));
|
||||
assert_eq!(head.content_length(), Some(10));
|
||||
|
||||
for range in ["bytes=1000-", "bytes=-0"] {
|
||||
let error = client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("ranged")
|
||||
.range(range)
|
||||
.send()
|
||||
.await
|
||||
.expect_err("unsatisfiable range must fail");
|
||||
let response = error.raw_response().expect("416 must retain the raw response");
|
||||
assert_eq!(response.status().as_u16(), 416, "{range}");
|
||||
assert_eq!(response.headers().get("content-range"), Some("bytes */1000"), "{range}");
|
||||
assert_eq!(error.code(), Some("InvalidRange"), "{range}");
|
||||
}
|
||||
|
||||
let ranged = source
|
||||
.requests()
|
||||
.into_iter()
|
||||
.find(|record| record.operation == Operation::GetObject && record.range.as_deref() == Some("bytes=10-19"))
|
||||
.expect("the Range header must be journaled verbatim");
|
||||
assert_eq!(ranged.key.as_deref(), Some("ranged"));
|
||||
assert!(source.count_requests(Operation::GetObject, "ranged") >= 6);
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_unversioned_bucket_overwrites_and_deletes() -> TestResult {
|
||||
let (source, client) = fake_source().await?;
|
||||
source.create_bucket_with_mode("plain-source", BucketMode::Unversioned);
|
||||
let versioning = client.get_bucket_versioning().bucket("plain-source").send().await?;
|
||||
assert!(versioning.status().is_none(), "unversioned bucket must report no versioning status");
|
||||
|
||||
let first = client
|
||||
.put_object()
|
||||
.bucket("plain-source")
|
||||
.key("doc")
|
||||
.body(ByteStream::from_static(b"first"))
|
||||
.send()
|
||||
.await?;
|
||||
assert!(first.version_id().is_none());
|
||||
let second = client
|
||||
.put_object()
|
||||
.bucket("plain-source")
|
||||
.key("doc")
|
||||
.body(ByteStream::from_static(b"second"))
|
||||
.send()
|
||||
.await?;
|
||||
assert!(second.version_id().is_none());
|
||||
let get = client.get_object().bucket("plain-source").key("doc").send().await?;
|
||||
assert!(get.version_id().is_none(), "GET must not return x-amz-version-id");
|
||||
assert_eq!(get.body.collect().await?.into_bytes().as_ref(), b"second");
|
||||
let head = client.head_object().bucket("plain-source").key("doc").send().await?;
|
||||
assert!(head.version_id().is_none(), "HEAD must not return x-amz-version-id");
|
||||
assert_eq!(source.stored_versions("plain-source", "doc").len(), 1, "overwrite must replace in place");
|
||||
|
||||
let deleted = client.delete_object().bucket("plain-source").key("doc").send().await?;
|
||||
assert!(deleted.delete_marker().is_none() && deleted.version_id().is_none());
|
||||
let missing = client
|
||||
.get_object()
|
||||
.bucket("plain-source")
|
||||
.key("doc")
|
||||
.send()
|
||||
.await
|
||||
.expect_err("deleted object must be gone");
|
||||
assert_eq!(missing.raw_response().map(|response| response.status().as_u16()), Some(404));
|
||||
assert_eq!(missing.code(), Some("NoSuchKey"));
|
||||
let missing_head = client
|
||||
.head_object()
|
||||
.bucket("plain-source")
|
||||
.key("doc")
|
||||
.send()
|
||||
.await
|
||||
.expect_err("deleted object must fail HEAD");
|
||||
assert_eq!(missing_head.raw_response().map(|response| response.status().as_u16()), Some(404));
|
||||
assert!(source.stored_versions("plain-source", "doc").is_empty(), "DELETE must not leave a marker");
|
||||
|
||||
// The versioned bucket on the same target keeps its version ids.
|
||||
let versioned = client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("doc")
|
||||
.body(ByteStream::from_static(b"versioned"))
|
||||
.send()
|
||||
.await?;
|
||||
assert!(versioned.version_id().is_some());
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_replays_standard_and_user_metadata() -> TestResult {
|
||||
let (source, client) = fake_source().await?;
|
||||
let body = payload(4096);
|
||||
let expected_etag = format!("\"{}\"", {
|
||||
use md5::Digest as _;
|
||||
hex_simd::encode_to_string(md5::Md5::digest(&body), hex_simd::AsciiCase::Lower)
|
||||
});
|
||||
// 2026-01-01T00:00:00Z rendered as an HTTP date by the SDK.
|
||||
let expires = DateTime::from_secs(1_767_225_600);
|
||||
client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("meta")
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.content_type("application/x-odm")
|
||||
.content_encoding("gzip")
|
||||
.content_disposition("attachment; filename=\"meta.bin\"")
|
||||
.content_language("en-US")
|
||||
.cache_control("max-age=60")
|
||||
.expires(expires)
|
||||
.metadata("Foo-Bar", "mixed case name")
|
||||
.metadata("UPPER", "upper name")
|
||||
.metadata("already-lower", "lower name")
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
let head = client.head_object().bucket(SOURCE_BUCKET).key("meta").send().await?;
|
||||
let get = client.get_object().bucket(SOURCE_BUCKET).key("meta").send().await?;
|
||||
for (label, content_type, content_encoding, content_disposition, content_language, cache_control, expires_string, e_tag) in [
|
||||
(
|
||||
"HEAD",
|
||||
head.content_type(),
|
||||
head.content_encoding(),
|
||||
head.content_disposition(),
|
||||
head.content_language(),
|
||||
head.cache_control(),
|
||||
head.expires_string(),
|
||||
head.e_tag(),
|
||||
),
|
||||
(
|
||||
"GET",
|
||||
get.content_type(),
|
||||
get.content_encoding(),
|
||||
get.content_disposition(),
|
||||
get.content_language(),
|
||||
get.cache_control(),
|
||||
get.expires_string(),
|
||||
get.e_tag(),
|
||||
),
|
||||
] {
|
||||
assert_eq!(content_type, Some("application/x-odm"), "{label}");
|
||||
assert_eq!(content_encoding, Some("gzip"), "{label}");
|
||||
assert_eq!(content_disposition, Some("attachment; filename=\"meta.bin\""), "{label}");
|
||||
assert_eq!(content_language, Some("en-US"), "{label}");
|
||||
assert_eq!(cache_control, Some("max-age=60"), "{label}");
|
||||
assert_eq!(expires_string, Some("Thu, 01 Jan 2026 00:00:00 GMT"), "{label}");
|
||||
assert_eq!(e_tag, Some(expected_etag.as_str()), "{label}");
|
||||
}
|
||||
for metadata in [head.metadata(), get.metadata()] {
|
||||
let metadata = metadata.expect("user metadata must be replayed");
|
||||
assert_eq!(metadata.get("foo-bar").map(String::as_str), Some("mixed case name"));
|
||||
assert_eq!(metadata.get("upper").map(String::as_str), Some("upper name"));
|
||||
assert_eq!(metadata.get("already-lower").map(String::as_str), Some("lower name"));
|
||||
assert!(!metadata.contains_key("Foo-Bar") && !metadata.contains_key("UPPER"));
|
||||
}
|
||||
assert!(head.last_modified().is_some());
|
||||
assert_eq!(head.last_modified(), get.last_modified());
|
||||
assert_eq!(head.content_length(), Some(4096));
|
||||
assert_eq!(get.body.collect().await?.into_bytes(), body);
|
||||
|
||||
// Seeded objects replay the same way.
|
||||
let seeded_etag = source.put_seed_object(
|
||||
SOURCE_BUCKET,
|
||||
"seeded",
|
||||
Bytes::from_static(b"seeded"),
|
||||
&SeedMetadata::new()
|
||||
.content_type("text/plain")
|
||||
.content_encoding("identity")
|
||||
.cache_control("no-store")
|
||||
.user_metadata("Origin", "seed"),
|
||||
);
|
||||
let seeded = client.head_object().bucket(SOURCE_BUCKET).key("seeded").send().await?;
|
||||
assert_eq!(seeded.e_tag(), Some(format!("\"{seeded_etag}\"").as_str()));
|
||||
assert_eq!(seeded.content_type(), Some("text/plain"));
|
||||
assert_eq!(seeded.content_encoding(), Some("identity"));
|
||||
assert_eq!(seeded.cache_control(), Some("no-store"));
|
||||
assert_eq!(
|
||||
seeded
|
||||
.metadata()
|
||||
.and_then(|metadata| metadata.get("origin"))
|
||||
.map(String::as_str),
|
||||
Some("seed")
|
||||
);
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_fault_actions_truncate_stall_and_status() -> TestResult {
|
||||
let (source, client) = fake_source().await?;
|
||||
let body = payload(4096);
|
||||
source.put_seed_object(SOURCE_BUCKET, "faulty", body.clone(), &SeedMetadata::new());
|
||||
|
||||
// TruncateBodyAt: headers promise 4096 bytes, the body ends after 100.
|
||||
source.inject_for_key(Operation::GetObject, "faulty", FaultAction::TruncateBodyAt(100), 1);
|
||||
let truncated = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
|
||||
assert_eq!(truncated.content_length(), Some(4096));
|
||||
let short_read = truncated
|
||||
.body
|
||||
.collect()
|
||||
.await
|
||||
.expect_err("a truncated body must fail to collect");
|
||||
let short_read = short_read.to_string();
|
||||
assert!(!short_read.is_empty());
|
||||
|
||||
// ResponseStatus: arbitrary status with the matching S3 error code.
|
||||
for (code, expected_code) in [
|
||||
(429u16, "SlowDown"),
|
||||
(404, "NoSuchKey"),
|
||||
(500, "InternalError"),
|
||||
(503, "ServiceUnavailable"),
|
||||
] {
|
||||
source.inject(Operation::GetObject, FaultAction::ResponseStatus(code), 1);
|
||||
let error = client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("faulty")
|
||||
.send()
|
||||
.await
|
||||
.expect_err("scripted status must fail");
|
||||
assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(code));
|
||||
assert_eq!(error.code(), Some(expected_code));
|
||||
}
|
||||
|
||||
// Stall: the fully computed response is held before its first byte.
|
||||
source.inject(Operation::HeadObject, FaultAction::Stall(Duration::from_millis(400)), 1);
|
||||
let started = Instant::now();
|
||||
let stalled = client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
|
||||
assert!(started.elapsed() >= Duration::from_millis(350), "stall must delay the first byte");
|
||||
assert_eq!(stalled.content_length(), Some(4096));
|
||||
let unstalled_started = Instant::now();
|
||||
client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
|
||||
assert!(unstalled_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
|
||||
|
||||
// The object is intact once the script is drained.
|
||||
let intact = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
|
||||
assert_eq!(intact.body.collect().await?.into_bytes(), body);
|
||||
|
||||
assert_eq!(source.count_requests(Operation::GetObject, "faulty"), 6);
|
||||
assert_eq!(source.count_requests(Operation::HeadObject, "faulty"), 2);
|
||||
assert_eq!(source.count_requests(Operation::GetObject, "other"), 0);
|
||||
let records = source.requests();
|
||||
assert!(
|
||||
records.iter().all(|record| record
|
||||
.user_agent
|
||||
.as_deref()
|
||||
.is_some_and(|agent| agent.contains("aws-sdk-rust"))),
|
||||
"the SDK user agent must be journaled"
|
||||
);
|
||||
assert!(
|
||||
records
|
||||
.iter()
|
||||
.any(|record| record.fault == Some(FaultAction::TruncateBodyAt(100)))
|
||||
);
|
||||
assert!(
|
||||
records
|
||||
.iter()
|
||||
.any(|record| record.fault == Some(FaultAction::Stall(Duration::from_millis(400))))
|
||||
);
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fake_source_raised_object_cap_accepts_large_put() -> TestResult {
|
||||
let source = FakeS3Target::start_with_options(FakeS3TargetOptions {
|
||||
max_object_bytes: 96 * 1024 * 1024,
|
||||
})
|
||||
.await?;
|
||||
source.create_bucket(SOURCE_BUCKET);
|
||||
let client = fake_source_client(&source);
|
||||
let len = 64 * 1024 * 1024 + 1;
|
||||
client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("large")
|
||||
.body(ByteStream::from(vec![7u8; len]))
|
||||
.send()
|
||||
.await?;
|
||||
let head = client.head_object().bucket(SOURCE_BUCKET).key("large").send().await?;
|
||||
assert_eq!(head.content_length(), Some(len as i64));
|
||||
let tail = client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("large")
|
||||
.range("bytes=-1")
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(tail.content_range(), Some(format!("bytes {}-{}/{len}", len - 1, len - 1).as_str()));
|
||||
source.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn odm_env_starts_rustfs_and_fake_source() -> TestResult {
|
||||
let env = OdmTestEnv::start().await?;
|
||||
env.source.create_bucket(SOURCE_BUCKET);
|
||||
let local_bucket = "odm-local";
|
||||
env.rustfs.create_test_bucket(local_bucket).await?;
|
||||
|
||||
let etags = env.seed_source(
|
||||
SOURCE_BUCKET,
|
||||
&[
|
||||
SeedObject::new("seed/a", Bytes::from_static(b"alpha")),
|
||||
SeedObject::new("seed/b", Bytes::from_static(b"beta"))
|
||||
.with_metadata(SeedMetadata::new().content_type("text/plain").user_metadata("Kind", "seed")),
|
||||
],
|
||||
);
|
||||
assert_eq!(etags.len(), 2);
|
||||
assert!(env.source.requests().is_empty(), "seeding must not touch the journal");
|
||||
let source_client = env.source_client();
|
||||
let seeded = source_client.head_object().bucket(SOURCE_BUCKET).key("seed/b").send().await?;
|
||||
assert_eq!(seeded.content_type(), Some("text/plain"));
|
||||
assert_eq!(seeded.e_tag(), Some(format!("\"{}\"", etags[1]).as_str()));
|
||||
assert_eq!(env.source.count_requests(Operation::HeadObject, "seed/b"), 1);
|
||||
|
||||
env.assert_local_absent(local_bucket, "seed/a").await;
|
||||
env.client
|
||||
.put_object()
|
||||
.bucket(local_bucket)
|
||||
.key("seed/a")
|
||||
.body(ByteStream::from_static(b"alpha"))
|
||||
.send()
|
||||
.await?;
|
||||
env.assert_local_present(local_bucket, "seed/a", b"alpha").await;
|
||||
env.assert_local_absent(local_bucket, "seed/b").await;
|
||||
|
||||
let spec = env.fake_source_spec(SOURCE_BUCKET).to_json();
|
||||
assert_eq!(spec["version"], 1);
|
||||
assert_eq!(spec["enabled"], true);
|
||||
assert_eq!(spec["source"]["provider"], "s3");
|
||||
assert_eq!(spec["source"]["endpoint"], env.source.endpoint());
|
||||
assert_eq!(spec["source"]["bucket"], SOURCE_BUCKET);
|
||||
assert_eq!(spec["source"]["credentials"]["secret_key"], "fake-secret");
|
||||
assert_eq!(spec["policy"]["source_timeout"]["first_byte_ms"], 15_000);
|
||||
assert!(spec["policy"]["bandwidth_limit_bytes_per_sec"].is_null());
|
||||
let debug = format!("{:?}", env.fake_source_spec(SOURCE_BUCKET));
|
||||
assert!(!debug.contains("fake-secret"), "Debug output must redact the secret");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn start_source_rustfs_round_trips_put_get() -> TestResult {
|
||||
let env = OdmTestEnv::start().await?;
|
||||
let source = start_source_rustfs().await?;
|
||||
assert_ne!(source.url, env.rustfs.url, "the source must be a separate instance");
|
||||
|
||||
source.create_test_bucket(SOURCE_BUCKET).await?;
|
||||
let source_client = source.create_s3_client();
|
||||
let body = payload(70_000);
|
||||
let put = source_client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("real/object")
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.content_type("application/octet-stream")
|
||||
.send()
|
||||
.await?;
|
||||
assert!(put.e_tag().is_some());
|
||||
let get = source_client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key("real/object")
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(get.content_type(), Some("application/octet-stream"));
|
||||
assert_eq!(get.body.collect().await?.into_bytes(), body);
|
||||
|
||||
let visible_to_primary = env
|
||||
.client
|
||||
.list_buckets()
|
||||
.send()
|
||||
.await?
|
||||
.buckets()
|
||||
.iter()
|
||||
.any(|bucket| bucket.name() == Some(SOURCE_BUCKET));
|
||||
assert!(!visible_to_primary, "the two servers must not share state");
|
||||
let spec = super::common::OdmSourceSpec::for_rustfs_source(&source, SOURCE_BUCKET).to_json();
|
||||
assert_eq!(spec["source"]["provider"], "rustfs");
|
||||
assert_eq!(spec["source"]["endpoint"], source.url);
|
||||
Ok(())
|
||||
}
|
||||
24
crates/e2e_test/src/on_demand_migration/mod.rs
Normal file
24
crates/e2e_test/src/on_demand_migration/mod.rs
Normal file
@@ -0,0 +1,24 @@
|
||||
// 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 (ODM) end-to-end suite (rustfs/backlog#2147).
|
||||
//!
|
||||
//! `common` is the shared environment: one RustFS under test, one programmable
|
||||
//! fake S3 source, admin-API wrappers, seeding and local-state assertions.
|
||||
//! `harness_self_test` proves the harness itself; ODM behavior scenarios are
|
||||
//! separate modules wired by later tasks.
|
||||
|
||||
pub mod common;
|
||||
|
||||
mod harness_self_test;
|
||||
Reference in New Issue
Block a user