mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-05 19:55:37 +00:00
feat(ecstore): add a native gcs odm source backend and one backend contract
This commit is contained in:
overtrue
@@ -153,12 +153,13 @@ pub mod bucket {
|
||||
LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup,
|
||||
OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason,
|
||||
PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS,
|
||||
SourceLatencySnapshot, source_client_spec,
|
||||
SourceLatencySnapshot, source_backend_spec, source_client_spec,
|
||||
};
|
||||
pub use crate::bucket::on_demand_migration::{
|
||||
ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
|
||||
OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
|
||||
SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
|
||||
AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
|
||||
ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig,
|
||||
Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig,
|
||||
ValidationContext,
|
||||
};
|
||||
pub use crate::bucket::on_demand_migration::{
|
||||
EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS,
|
||||
@@ -184,9 +185,9 @@ pub mod bucket {
|
||||
}
|
||||
pub mod source_client {
|
||||
pub use crate::bucket::on_demand_migration::source_client::{
|
||||
SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
|
||||
SourceProbe, SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
|
||||
resolve_path_style,
|
||||
AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError,
|
||||
SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse,
|
||||
SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -543,10 +543,9 @@ fn leaf_text(reader: &mut Reader<&[u8]>, end: quick_xml::name::QName<'_>) -> Res
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
|
||||
use crate::bucket::on_demand_migration::source_client::SourceError;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
|
||||
|
||||
const LIST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<EnumerationResults ServiceEndpoint="https://acct.blob.core.windows.net/" ContainerName="legacy">
|
||||
@@ -693,72 +692,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct Recorded {
|
||||
method: String,
|
||||
target: String,
|
||||
headers: Vec<(String, String)>,
|
||||
}
|
||||
|
||||
impl Recorded {
|
||||
fn header(&self, name: &str) -> Option<&str> {
|
||||
self.headers
|
||||
.iter()
|
||||
.find(|(key, _)| key.eq_ignore_ascii_case(name))
|
||||
.map(|(_, value)| value.as_str())
|
||||
}
|
||||
}
|
||||
|
||||
/// One canned HTTP/1.1 response per connection. `Connection: close` keeps
|
||||
/// every request on its own socket so the script order is deterministic.
|
||||
async fn scripted_server(responses: Vec<(u16, Vec<(&'static str, String)>, String)>) -> (Url, Arc<Mutex<Vec<Recorded>>>) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind");
|
||||
let port = listener.local_addr().expect("fixture address").port();
|
||||
let recorded: Arc<Mutex<Vec<Recorded>>> = Arc::new(Mutex::new(Vec::new()));
|
||||
let sink = Arc::clone(&recorded);
|
||||
|
||||
tokio::spawn(async move {
|
||||
for (status, headers, body) in responses {
|
||||
let Ok((mut stream, _)) = listener.accept().await else {
|
||||
return;
|
||||
};
|
||||
let mut request = Vec::new();
|
||||
let mut buffer = [0_u8; 2048];
|
||||
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
|
||||
match stream.read(&mut buffer).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(read) => request.extend_from_slice(&buffer[..read]),
|
||||
}
|
||||
}
|
||||
let text = String::from_utf8_lossy(&request).into_owned();
|
||||
let mut lines = text.lines();
|
||||
let start = lines.next().unwrap_or_default().to_string();
|
||||
let mut parts = start.split_whitespace();
|
||||
let recorded = Recorded {
|
||||
method: parts.next().unwrap_or_default().to_string(),
|
||||
target: parts.next().unwrap_or_default().to_string(),
|
||||
headers: lines
|
||||
.take_while(|line| !line.is_empty())
|
||||
.filter_map(|line| line.split_once(':'))
|
||||
.map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
|
||||
.collect(),
|
||||
};
|
||||
sink.lock().expect("recorder lock").push(recorded);
|
||||
|
||||
let mut response = format!("HTTP/1.1 {status} X\r\nContent-Length: {}\r\nConnection: close\r\n", body.len());
|
||||
for (name, value) in headers {
|
||||
response.push_str(&format!("{name}: {value}\r\n"));
|
||||
}
|
||||
response.push_str("\r\n");
|
||||
response.push_str(&body);
|
||||
let _ = stream.write_all(response.as_bytes()).await;
|
||||
let _ = stream.flush().await;
|
||||
}
|
||||
});
|
||||
|
||||
(Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorded)
|
||||
}
|
||||
|
||||
fn blob_headers() -> Vec<(&'static str, String)> {
|
||||
vec![
|
||||
("ETag", "\"0x8D2F1B0A1B2C3D4\"".to_string()),
|
||||
@@ -774,7 +707,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn head_signs_the_request_and_maps_azure_metadata() {
|
||||
let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await;
|
||||
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, blob_headers(), String::new())]).await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(b"0123456789abcdef0123456789abcdef".to_vec()));
|
||||
|
||||
let head = backend.head("photos/a b.jpg").await.expect("HEAD should map");
|
||||
@@ -806,7 +739,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn sas_credentials_travel_in_the_query_and_never_sign() {
|
||||
let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await;
|
||||
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, blob_headers(), String::new())]).await;
|
||||
let backend = backend(
|
||||
&endpoint,
|
||||
Credential::Sas(vec![
|
||||
@@ -831,7 +764,7 @@ mod tests {
|
||||
async fn get_passes_the_range_through_and_streams_the_body() {
|
||||
let mut headers = blob_headers();
|
||||
headers.push(("Content-Range", "bytes 10-14/100".to_string()));
|
||||
let (endpoint, recorded) = scripted_server(vec![(206, headers, "hello".to_string())]).await;
|
||||
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(206, headers, "hello".to_string())]).await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
|
||||
let range = HTTPRangeSpec {
|
||||
@@ -854,7 +787,7 @@ mod tests {
|
||||
async fn customer_key_blobs_are_refused() {
|
||||
let mut headers = blob_headers();
|
||||
headers.push(("x-ms-encryption-key-sha256", "abc".to_string()));
|
||||
let (endpoint, _) = scripted_server(vec![(200, headers, String::new())]).await;
|
||||
let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
|
||||
let err = backend.head("a.txt").await.expect_err("customer-key blobs are unsupported");
|
||||
@@ -866,8 +799,8 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn list_requests_the_container_and_pages_with_the_marker() {
|
||||
let (endpoint, recorded) = scripted_server(vec![
|
||||
(200, Vec::new(), LIST_PAGE.to_string()),
|
||||
(200, Vec::new(), LAST_PAGE.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), LIST_PAGE.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), LAST_PAGE.to_string()),
|
||||
])
|
||||
.await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
@@ -931,8 +864,11 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn tagging_and_probe_address_the_right_resources() {
|
||||
let (endpoint, recorded) =
|
||||
scripted_server(vec![(200, Vec::new(), TAGS.to_string()), (200, Vec::new(), String::new())]).await;
|
||||
let (endpoint, recorded) = scripted_server(vec![
|
||||
ScriptedResponse::new(200, Vec::new(), TAGS.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), String::new()),
|
||||
])
|
||||
.await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
|
||||
let tags = backend.tagging("a.txt").await.expect("tags should parse");
|
||||
@@ -958,7 +894,7 @@ mod tests {
|
||||
let headers = code
|
||||
.map(|code| vec![(HEADER_ERROR_CODE, code.to_string())])
|
||||
.unwrap_or_default();
|
||||
let (endpoint, _) = scripted_server(vec![(status, headers, String::new())]).await;
|
||||
let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(status, headers, String::new())]).await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
let err = backend.head("a.txt").await.expect_err("{status} must fail");
|
||||
assert_eq!(err.class_label(), expected, "status {status} -> {err:?}");
|
||||
@@ -966,6 +902,90 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
const CONTRACT_LIST_PAGE_ONE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<EnumerationResults ContainerName="legacy">
|
||||
<Blobs>
|
||||
<Blob>
|
||||
<Name>dir/a.txt</Name>
|
||||
<Properties>
|
||||
<Last-Modified>Wed, 21 Oct 2015 07:28:00 GMT</Last-Modified>
|
||||
<Etag>0x8D2F1B0A1B2C3D4</Etag>
|
||||
<Content-Length>5</Content-Length>
|
||||
<AccessTier>Hot</AccessTier>
|
||||
</Properties>
|
||||
</Blob>
|
||||
<BlobPrefix><Name>dir/sub/</Name></BlobPrefix>
|
||||
</Blobs>
|
||||
<NextMarker>cursor-1</NextMarker>
|
||||
</EnumerationResults>"#;
|
||||
|
||||
const CONTRACT_LIST_PAGE_TWO: &str = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<EnumerationResults ContainerName="legacy">
|
||||
<Blobs>
|
||||
<Blob>
|
||||
<Name>dir/b.txt</Name>
|
||||
<Properties>
|
||||
<Last-Modified>Wed, 21 Oct 2015 07:28:00 GMT</Last-Modified>
|
||||
<Content-Length>7</Content-Length>
|
||||
</Properties>
|
||||
</Blob>
|
||||
</Blobs>
|
||||
<NextMarker />
|
||||
</EnumerationResults>"#;
|
||||
|
||||
const CONTRACT_TAGS: &str = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<Tags><TagSet><Tag><Key>env</Key><Value>prod</Value></Tag></TagSet></Tags>"#;
|
||||
|
||||
fn contract_blob_headers() -> Vec<(&'static str, String)> {
|
||||
vec![
|
||||
("ETag", "\"0x8D2F1B0A1B2C3D4\"".to_string()),
|
||||
("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
|
||||
("Content-Type", "text/plain".to_string()),
|
||||
("x-ms-meta-owner", "alice".to_string()),
|
||||
("x-ms-access-tier", "Hot".to_string()),
|
||||
("x-ms-blob-type", "BlockBlob".to_string()),
|
||||
]
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn azure_backend_satisfies_the_shared_backend_contract() {
|
||||
let mut ranged = contract_blob_headers();
|
||||
ranged.push(("Content-Range", "bytes 1-3/5".to_string()));
|
||||
// A HEAD reports the object size with no body, exactly as Azure does.
|
||||
let mut head_only = contract_blob_headers();
|
||||
head_only.push(("Content-Length", "5".to_string()));
|
||||
let (endpoint, _) = scripted_server(vec![
|
||||
ScriptedResponse::new(200, head_only, String::new()),
|
||||
ScriptedResponse::new(200, contract_blob_headers(), "hello".to_string()),
|
||||
ScriptedResponse::new(206, ranged, "ell".to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), CONTRACT_LIST_PAGE_ONE.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), CONTRACT_LIST_PAGE_TWO.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), CONTRACT_TAGS.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), String::new()),
|
||||
ScriptedResponse::new(404, vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())], String::new()),
|
||||
ScriptedResponse::new(
|
||||
403,
|
||||
vec![(HEADER_ERROR_CODE, "AuthorizationPermissionMismatch".to_string())],
|
||||
String::new(),
|
||||
),
|
||||
])
|
||||
.await;
|
||||
let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
|
||||
|
||||
assert_backend_contract(
|
||||
&backend,
|
||||
BackendCapabilities {
|
||||
// Azure's ETag is a concurrency token; the contract requires it
|
||||
// to be carried but never read as a digest.
|
||||
etag_is_opaque: true,
|
||||
// Azure paginates only with an opaque marker.
|
||||
supports_start_after: false,
|
||||
supports_tagging: true,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn transport_failures_never_render_the_request_url() {
|
||||
// Nothing is listening on the reserved port, so the connect fails and
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
// 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.
|
||||
|
||||
//! One contract every [`SourceBackend`] implementation must satisfy.
|
||||
//!
|
||||
//! The migration pipeline talks to a source only through the trait, so a new
|
||||
//! provider is correct exactly when it answers the same questions the same way:
|
||||
//! the same head fields, the same range semantics, the same page shape, the
|
||||
//! same error classes. Each backend supplies a fixture that answers this fixed
|
||||
//! corpus in its own dialect and then runs [`assert_backend_contract`], so a
|
||||
//! provider-specific mapping bug shows up as a contract failure rather than as
|
||||
//! a surprise in the pull pipeline.
|
||||
//!
|
||||
//! Backends differ in two documented ways, declared through
|
||||
//! [`BackendCapabilities`]: whether the provider's ETag is a content digest,
|
||||
//! and whether the provider can resume a listing from a key.
|
||||
|
||||
use super::source_client::{SourceBackend, SourceError, SourceListRequest};
|
||||
use crate::storage_api_contracts::range::HTTPRangeSpec;
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// The single object every fixture serves.
|
||||
pub(super) const OBJECT_KEY: &str = "dir/a.txt";
|
||||
pub(super) const OBJECT_BODY: &[u8] = b"hello";
|
||||
/// MD5 of [`OBJECT_BODY`]; the ETag of the object on a digest provider.
|
||||
pub(super) const OBJECT_MD5: &str = "5d41402abc4b2a76b9719d911017c592";
|
||||
/// The second key the fixture's listing returns, on its second page.
|
||||
pub(super) const SECOND_KEY: &str = "dir/b.txt";
|
||||
pub(super) const COMMON_PREFIX: &str = "dir/sub/";
|
||||
pub(super) const LIST_CURSOR: &str = "cursor-1";
|
||||
/// A key the fixture answers with the provider's "no such object".
|
||||
pub(super) const MISSING_KEY: &str = "missing";
|
||||
/// A key the fixture answers with the provider's "not authorized".
|
||||
pub(super) const FORBIDDEN_KEY: &str = "secret";
|
||||
|
||||
/// Where backends are allowed to differ.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub(super) struct BackendCapabilities {
|
||||
/// The provider's ETag is an opaque token, not a digest of the bytes.
|
||||
pub(super) etag_is_opaque: bool,
|
||||
/// The provider can resume a listing from a key rather than only from an
|
||||
/// opaque cursor.
|
||||
pub(super) supports_start_after: bool,
|
||||
/// The provider has an object-tagging concept at all. GCS does not, and
|
||||
/// answers with an empty map instead of failing a pull.
|
||||
pub(super) supports_tagging: bool,
|
||||
}
|
||||
|
||||
/// Drives `backend` through the shared corpus. Fixtures are scripted in
|
||||
/// request order, so the call order here is part of the contract.
|
||||
pub(super) async fn assert_backend_contract(backend: &dyn SourceBackend, caps: BackendCapabilities) {
|
||||
// 1. HEAD maps the object's shared fields.
|
||||
let head = backend.head(OBJECT_KEY).await.expect("HEAD of the fixture object");
|
||||
assert_eq!(head.size, OBJECT_BODY.len() as u64, "HEAD reports the object size");
|
||||
assert_eq!(head.content_type.as_deref(), Some("text/plain"));
|
||||
assert_eq!(
|
||||
head.user_metadata,
|
||||
HashMap::from([("owner".to_string(), "alice".to_string())]),
|
||||
"user metadata is keyed without the provider prefix"
|
||||
);
|
||||
assert!(head.storage_class.is_some(), "the provider's tier is recorded");
|
||||
assert!(head.last_modified.is_some(), "the provider's timestamp is parsed");
|
||||
assert!(head.sse.is_none(), "the fixture object is not server-side encrypted");
|
||||
assert!(!head.is_multipart_etag);
|
||||
assert_eq!(head.etag_is_opaque, caps.etag_is_opaque);
|
||||
match caps.etag_is_opaque {
|
||||
false => assert_eq!(head.etag.as_deref(), Some(OBJECT_MD5), "a digest ETag is mapped verbatim"),
|
||||
true => assert!(head.etag.is_some(), "an opaque ETag is still recorded"),
|
||||
}
|
||||
|
||||
// 2. An unranged GET streams the whole object and reports no range.
|
||||
let got = backend.get(OBJECT_KEY, None).await.expect("unranged GET");
|
||||
assert_eq!(got.head.size, OBJECT_BODY.len() as u64);
|
||||
assert!(got.content_range.is_none(), "an unranged GET has no content-range");
|
||||
assert_eq!(got.head.etag_is_opaque, caps.etag_is_opaque, "GET and HEAD agree about the ETag");
|
||||
let body = got.body.collect().await.expect("body streams").into_bytes();
|
||||
assert_eq!(body.as_ref(), OBJECT_BODY);
|
||||
|
||||
// 3. A ranged GET returns exactly the requested interval, and `size` is
|
||||
// the length of the returned bytes rather than of the object.
|
||||
let range = HTTPRangeSpec {
|
||||
is_suffix_length: false,
|
||||
start: 1,
|
||||
end: 3,
|
||||
};
|
||||
let got = backend.get(OBJECT_KEY, Some(&range)).await.expect("ranged GET");
|
||||
assert_eq!(got.head.size, 3, "a ranged GET reports the range length");
|
||||
assert_eq!(got.content_range.as_deref(), Some("bytes 1-3/5"));
|
||||
let body = got.body.collect().await.expect("body streams").into_bytes();
|
||||
assert_eq!(body.as_ref(), &OBJECT_BODY[1..=3]);
|
||||
|
||||
// 4. A delimiter listing rolls prefixes up and hands back a cursor.
|
||||
let page = backend
|
||||
.list(&SourceListRequest {
|
||||
prefix: Some("dir/"),
|
||||
delimiter: Some("/"),
|
||||
max_keys: 2,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect("first listing page");
|
||||
assert_eq!(page.objects.len(), 1, "the first page holds one object");
|
||||
assert_eq!(page.objects[0].key, OBJECT_KEY, "listing keys are in the source namespace");
|
||||
assert_eq!(page.objects[0].size, OBJECT_BODY.len() as u64);
|
||||
assert!(page.objects[0].last_modified.is_some());
|
||||
assert_eq!(page.common_prefixes, vec![COMMON_PREFIX.to_string()]);
|
||||
assert!(page.is_truncated);
|
||||
assert_eq!(page.next_continuation_token.as_deref(), Some(LIST_CURSOR));
|
||||
|
||||
// 5. The cursor is passed back verbatim and the last page ends the walk.
|
||||
let page = backend
|
||||
.list(&SourceListRequest {
|
||||
prefix: Some("dir/"),
|
||||
delimiter: Some("/"),
|
||||
continuation_token: Some(LIST_CURSOR),
|
||||
max_keys: 2,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect("second listing page");
|
||||
assert_eq!(page.objects.len(), 1);
|
||||
assert_eq!(page.objects[0].key, SECOND_KEY);
|
||||
assert!(!page.is_truncated);
|
||||
assert!(page.next_continuation_token.is_none(), "a complete listing carries no cursor");
|
||||
|
||||
// 6. Tags come back as a flat map, empty on a provider without tags.
|
||||
let tags = backend.tagging(OBJECT_KEY).await.expect("object tags");
|
||||
match caps.supports_tagging {
|
||||
true => assert_eq!(tags, HashMap::from([("env".to_string(), "prod".to_string())])),
|
||||
false => assert!(tags.is_empty(), "a provider without tags reports none: {tags:?}"),
|
||||
}
|
||||
|
||||
// 7. The probe confirms the bucket or container answers.
|
||||
backend.probe().await.expect("probe of the fixture bucket");
|
||||
|
||||
// 8. A missing object is `NotFound`, and never retried.
|
||||
let err = backend.head(MISSING_KEY).await.expect_err("a missing object must fail");
|
||||
assert!(matches!(err, SourceError::NotFound), "{err:?}");
|
||||
assert_eq!(err.class_label(), "not_found");
|
||||
assert!(!err.is_retryable());
|
||||
|
||||
// 9. A denied object is `AccessDenied`, and never retried.
|
||||
let err = backend.head(FORBIDDEN_KEY).await.expect_err("a denied object must fail");
|
||||
assert!(matches!(err, SourceError::AccessDenied), "{err:?}");
|
||||
assert_eq!(err.class_label(), "access_denied");
|
||||
assert!(!err.is_retryable());
|
||||
|
||||
// 10. A provider without a key cursor must refuse one instead of listing
|
||||
// from the wrong position. This issues no request either way.
|
||||
if !caps.supports_start_after {
|
||||
let err = backend
|
||||
.list(&SourceListRequest {
|
||||
start_after: Some(OBJECT_KEY),
|
||||
max_keys: 1,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect_err("a backend without a key cursor must refuse start_after");
|
||||
assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
|
||||
}
|
||||
}
|
||||
@@ -1024,9 +1024,201 @@ mod tests {
|
||||
"{provider}"
|
||||
);
|
||||
}
|
||||
// The native providers never sign with a region, so "auto" is the
|
||||
// honest value to write for them.
|
||||
for cfg in [azure_cfg(), gcs_native_cfg()] {
|
||||
assert_eq!(cfg.source.region, "auto");
|
||||
cfg.validate(empty_ctx())
|
||||
.unwrap_or_else(|err| panic!("{}: {err}", cfg.source.provider));
|
||||
}
|
||||
assert_eq!(sample().source.effective_region(), "us-west-1");
|
||||
}
|
||||
|
||||
const SERVICE_ACCOUNT_JSON: &str = r#"{"type":"service_account","project_id":"p","client_email":"a@b.iam.gserviceaccount.com","private_key":"-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----\n"}"#;
|
||||
|
||||
fn azure_cfg() -> OnDemandMigrationConfig {
|
||||
let mut cfg = sample();
|
||||
cfg.source.provider = Provider::Azure;
|
||||
cfg.source.endpoint = None;
|
||||
cfg.source.region = "auto".to_string();
|
||||
cfg.source.credentials = None;
|
||||
cfg.source.azure = Some(AzureSourceConfig {
|
||||
account: "legacyaccount".to_string(),
|
||||
account_key: Some("c2VjcmV0LWtleQ==".to_string()),
|
||||
sas_token: None,
|
||||
});
|
||||
cfg
|
||||
}
|
||||
|
||||
fn gcs_native_cfg() -> OnDemandMigrationConfig {
|
||||
let mut cfg = sample();
|
||||
cfg.source.provider = Provider::GcsNative;
|
||||
cfg.source.endpoint = None;
|
||||
cfg.source.region = "auto".to_string();
|
||||
cfg.source.credentials = None;
|
||||
cfg.source.gcs = Some(GcsSourceConfig {
|
||||
service_account_json: SERVICE_ACCOUNT_JSON.to_string(),
|
||||
});
|
||||
cfg
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_providers_derive_their_endpoint_and_round_trip_on_the_wire() {
|
||||
let azure = azure_cfg();
|
||||
assert_eq!(azure.source.effective_endpoint(), "https://legacyaccount.blob.core.windows.net");
|
||||
let gcs = gcs_native_cfg();
|
||||
assert_eq!(gcs.source.effective_endpoint(), "https://storage.googleapis.com");
|
||||
|
||||
for cfg in [azure_cfg(), gcs_native_cfg()] {
|
||||
let json = cfg.to_json().expect("config must serialize");
|
||||
assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg);
|
||||
}
|
||||
// The wire labels are part of the admin contract.
|
||||
assert!(
|
||||
String::from_utf8(azure_cfg().to_json().expect("json"))
|
||||
.expect("utf8")
|
||||
.contains(r#""provider":"azure""#)
|
||||
);
|
||||
assert!(
|
||||
String::from_utf8(gcs_native_cfg().to_json().expect("json"))
|
||||
.expect("utf8")
|
||||
.contains(r#""provider":"gcs_native""#)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_explicit_endpoint_overrides_the_derived_native_one() {
|
||||
// Azurite and fake-gcs-server are addressed this way.
|
||||
let mut cfg = azure_cfg();
|
||||
cfg.source.endpoint = Some("http://azurite.example.com:10000".to_string());
|
||||
cfg.validate(empty_ctx()).expect("an explicit native endpoint is allowed");
|
||||
assert_eq!(cfg.source.effective_endpoint(), "http://azurite.example.com:10000");
|
||||
|
||||
cfg.source.endpoint = Some("http://azurite.example.com:10000/devstoreaccount1".to_string());
|
||||
assert!(
|
||||
matches!(cfg.validate(empty_ctx()), Err(OnDemandMigrationConfigError::InvalidEndpoint(_))),
|
||||
"a native endpoint is still an origin"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_provider_block_belongs_to_exactly_its_own_provider() {
|
||||
let mut cfg = sample();
|
||||
cfg.source.azure = azure_cfg().source.azure;
|
||||
assert_eq!(
|
||||
cfg.validate(empty_ctx()),
|
||||
Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("azure", Provider::S3))
|
||||
);
|
||||
|
||||
let mut cfg = sample();
|
||||
cfg.source.gcs = gcs_native_cfg().source.gcs;
|
||||
assert_eq!(
|
||||
cfg.validate(empty_ctx()),
|
||||
Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("gcs", Provider::S3))
|
||||
);
|
||||
|
||||
let mut cfg = azure_cfg();
|
||||
cfg.source.azure = None;
|
||||
assert_eq!(
|
||||
cfg.validate(empty_ctx()),
|
||||
Err(OnDemandMigrationConfigError::MissingProviderBlock("azure", Provider::Azure))
|
||||
);
|
||||
|
||||
let mut cfg = gcs_native_cfg();
|
||||
cfg.source.gcs = None;
|
||||
assert_eq!(
|
||||
cfg.validate(empty_ctx()),
|
||||
Err(OnDemandMigrationConfigError::MissingProviderBlock("gcs", Provider::GcsNative))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn azure_block_rules() {
|
||||
let with = |account: &str, key: Option<&str>, sas: Option<&str>| {
|
||||
let mut cfg = azure_cfg();
|
||||
cfg.source.azure = Some(AzureSourceConfig {
|
||||
account: account.to_string(),
|
||||
account_key: key.map(str::to_string),
|
||||
sas_token: sas.map(str::to_string),
|
||||
});
|
||||
cfg.validate(empty_ctx())
|
||||
};
|
||||
|
||||
with("legacyaccount", None, Some("sv=2021-08-06&sig=abc%3D")).expect("a SAS token is a complete credential");
|
||||
with("legacyaccount", Some("c2VjcmV0LWtleQ=="), None).expect("an account key is a complete credential");
|
||||
|
||||
for (label, result) in [
|
||||
("empty account", with("", Some("c2VjcmV0LWtleQ=="), None)),
|
||||
// The account becomes the first label of the derived hostname.
|
||||
("account with a dot", with("legacy.account", Some("c2VjcmV0LWtleQ=="), None)),
|
||||
("account with a slash", with("legacy/account", Some("c2VjcmV0LWtleQ=="), None)),
|
||||
("no credential", with("legacyaccount", None, None)),
|
||||
("both credentials", with("legacyaccount", Some("c2VjcmV0LWtleQ=="), Some("sv=1"))),
|
||||
("empty key", with("legacyaccount", Some(""), None)),
|
||||
("key that is not base64", with("legacyaccount", Some("not base64!"), None)),
|
||||
("empty sas", with("legacyaccount", None, Some(""))),
|
||||
("sas with a leading question mark", with("legacyaccount", None, Some("?sv=1"))),
|
||||
("sas with whitespace", with("legacyaccount", None, Some("sv=1 &sig=a"))),
|
||||
] {
|
||||
assert!(
|
||||
matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("azure", _))),
|
||||
"{label}: {result:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gcs_native_block_requires_a_usable_service_account_key() {
|
||||
let with = |json: &str| {
|
||||
let mut cfg = gcs_native_cfg();
|
||||
cfg.source.gcs = Some(GcsSourceConfig {
|
||||
service_account_json: json.to_string(),
|
||||
});
|
||||
cfg.validate(empty_ctx())
|
||||
};
|
||||
|
||||
with(SERVICE_ACCOUNT_JSON).expect("a service-account key is accepted");
|
||||
for (label, json) in [
|
||||
("empty", ""),
|
||||
("not json", "not json"),
|
||||
("not an object", "[]"),
|
||||
("wrong type", r#"{"type":"authorized_user","client_email":"a@b","private_key":"k"}"#),
|
||||
("no private key", r#"{"type":"service_account","client_email":"a@b"}"#),
|
||||
("empty client email", r#"{"type":"service_account","client_email":"","private_key":"k"}"#),
|
||||
] {
|
||||
let result = with(json);
|
||||
assert!(
|
||||
matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("gcs", _))),
|
||||
"{label}: {result:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_secrets_never_survive_redaction_or_debug() {
|
||||
let mut azure = azure_cfg();
|
||||
azure.source.azure.as_mut().expect("block").sas_token = Some("sv=2021-08-06&sig=top-secret".to_string());
|
||||
azure.source.azure.as_mut().expect("block").account_key = None;
|
||||
let gcs = gcs_native_cfg();
|
||||
|
||||
for rendered in [
|
||||
format!("{:?}", azure.redacted()),
|
||||
format!("{azure:?}"),
|
||||
String::from_utf8(azure.redacted().to_json().expect("json")).expect("utf8"),
|
||||
] {
|
||||
assert!(!rendered.contains("top-secret"), "{rendered}");
|
||||
assert!(rendered.contains("legacyaccount"), "the account name is not a secret: {rendered}");
|
||||
}
|
||||
for rendered in [
|
||||
format!("{:?}", gcs.redacted()),
|
||||
format!("{gcs:?}"),
|
||||
String::from_utf8(gcs.redacted().to_json().expect("json")).expect("utf8"),
|
||||
] {
|
||||
assert!(!rendered.contains("BEGIN PRIVATE KEY"), "{rendered}");
|
||||
assert!(!rendered.contains("gserviceaccount"), "{rendered}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bucket_rules() {
|
||||
let mut cfg = sample();
|
||||
|
||||
@@ -0,0 +1,494 @@
|
||||
// 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.
|
||||
|
||||
//! Native Google Cloud Storage source backend.
|
||||
//!
|
||||
//! The `gcs` provider already reaches GCS through its S3 interoperability API,
|
||||
//! which needs an HMAC key pair. This backend is the other half: it authorizes
|
||||
//! with a service-account key, the credential most GCS projects actually issue,
|
||||
//! by minting OAuth tokens through the shared `google-cloud-auth` credential
|
||||
//! machinery the tier layer already uses.
|
||||
//!
|
||||
//! Two GCS surfaces are involved, each for the half it describes best. The read
|
||||
//! path uses the XML API (`/{bucket}/{object}`), whose responses carry
|
||||
//! `x-goog-meta-*` user metadata and the `x-goog-hash` digest in one round trip.
|
||||
//! Listing uses the JSON API (`objects.list`), whose `pageToken` maps directly
|
||||
//! onto the shared page cursor and whose `prefixes` are the delimiter roll-up.
|
||||
//! Both accept the same bearer token.
|
||||
//!
|
||||
//! `x-goog-hash` carries a base64 MD5 for every non-composite object; it is
|
||||
//! converted to hex and becomes the head's ETag, so a pulled object is checked
|
||||
//! against the digest GCS itself computed. A composite object has no MD5, and
|
||||
//! its ETag is then marked opaque rather than checked.
|
||||
|
||||
use super::native_http::{
|
||||
NativeHeadFields, NativeHttp, base64_md5_to_hex, header, native_source_head, parse_http_timestamp, read_text, response_body,
|
||||
};
|
||||
use super::source_client::{
|
||||
GcsSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
|
||||
SourceTimeouts, range_header_value,
|
||||
};
|
||||
use crate::bucket::remote_s3_client::RemoteS3ClientError;
|
||||
use crate::storage_api_contracts::range::HTTPRangeSpec;
|
||||
use google_cloud_auth::credentials::service_account::{AccessSpecifier, Builder as ServiceAccountBuilder};
|
||||
use google_cloud_auth::credentials::{CacheableResource, Credentials};
|
||||
use http::{HeaderMap, HeaderValue, Method};
|
||||
use serde::Deserialize;
|
||||
use std::collections::HashMap;
|
||||
use url::Url;
|
||||
|
||||
/// Read-only object scope: this backend never writes to the source.
|
||||
const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only";
|
||||
const METADATA_PREFIX: &str = "x-goog-meta-";
|
||||
/// GCS reports its error code in the response body, not a header; the shared
|
||||
/// transport takes a header name, so it is given one that never matches and
|
||||
/// classification falls back to the status.
|
||||
const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code";
|
||||
/// One `objects.list` page is small; refuse an unbounded document.
|
||||
const MAX_JSON_BYTES: usize = 8 * 1024 * 1024;
|
||||
|
||||
pub struct GcsNativeSourceBackend {
|
||||
http: NativeHttp,
|
||||
bucket: String,
|
||||
credentials: Credentials,
|
||||
}
|
||||
|
||||
impl GcsNativeSourceBackend {
|
||||
pub fn new(
|
||||
endpoint: &str,
|
||||
bucket: &str,
|
||||
spec: &GcsSourceSpec,
|
||||
timeouts: SourceTimeouts,
|
||||
skip_tls_verify: bool,
|
||||
ca_cert_pem: Option<&str>,
|
||||
) -> Result<Self, RemoteS3ClientError> {
|
||||
let key: serde_json::Value = serde_json::from_str(&spec.service_account_json)
|
||||
.map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not valid JSON"))?;
|
||||
let credentials = ServiceAccountBuilder::new(key)
|
||||
.with_access_specifier(AccessSpecifier::from_scopes([READ_ONLY_SCOPE]))
|
||||
.build()
|
||||
.map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not usable"))?;
|
||||
Ok(Self {
|
||||
http: NativeHttp::new(endpoint, timeouts, skip_tls_verify, ca_cert_pem)?,
|
||||
bucket: bucket.to_string(),
|
||||
credentials,
|
||||
})
|
||||
}
|
||||
|
||||
/// Authorization headers for one request. A credential failure is reported
|
||||
/// as `AccessDenied` with no message: the renderer of a credential error
|
||||
/// has the key material in scope, and the class is what callers act on.
|
||||
async fn auth_headers(&self) -> Result<HeaderMap, SourceError> {
|
||||
match self.credentials.headers(http::Extensions::new()).await {
|
||||
Ok(CacheableResource::New { data, .. }) => Ok(data),
|
||||
// Only returned when the caller passes an entity tag, which this
|
||||
// backend never does; an empty set is still the honest answer.
|
||||
Ok(CacheableResource::NotModified) => Ok(HeaderMap::new()),
|
||||
Err(_) => Err(SourceError::AccessDenied),
|
||||
}
|
||||
}
|
||||
|
||||
/// XML API URL of one object; `/` in the key stay path separators.
|
||||
fn object_url(&self, key: &str) -> Result<Url, SourceError> {
|
||||
self.http.url(std::iter::once(self.bucket.as_str()).chain(key.split('/')))
|
||||
}
|
||||
|
||||
/// JSON API URL of the bucket's object collection.
|
||||
fn objects_url(&self) -> Result<Url, SourceError> {
|
||||
self.http.url(["storage", "v1", "b", self.bucket.as_str(), "o"])
|
||||
}
|
||||
|
||||
fn bucket_url(&self) -> Result<Url, SourceError> {
|
||||
self.http.url(["storage", "v1", "b", self.bucket.as_str()])
|
||||
}
|
||||
|
||||
async fn request(&self, method: Method, url: Url, mut headers: HeaderMap) -> Result<reqwest::Request, SourceError> {
|
||||
for (name, value) in self.auth_headers().await? {
|
||||
if let Some(name) = name {
|
||||
headers.insert(name, value);
|
||||
}
|
||||
}
|
||||
let mut request = reqwest::Request::new(method, url);
|
||||
*request.headers_mut() = headers;
|
||||
Ok(request)
|
||||
}
|
||||
|
||||
/// Shared mapping for the XML API's HEAD and GET responses.
|
||||
fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
|
||||
if header(headers, "x-goog-encryption-key-sha256").is_some() {
|
||||
return Err(SourceError::Unsupported(
|
||||
"source object uses a customer-supplied encryption key; customer-key sources are not supported".to_string(),
|
||||
));
|
||||
}
|
||||
// `x-goog-hash` lists digests as `name=base64`, comma separated, and may
|
||||
// repeat across header lines. Only the MD5 describes the whole object.
|
||||
let md5 = headers
|
||||
.get_all("x-goog-hash")
|
||||
.iter()
|
||||
.filter_map(|value| value.to_str().ok())
|
||||
.flat_map(|value| value.split(','))
|
||||
.filter_map(|digest| digest.trim().strip_prefix("md5="))
|
||||
.find_map(base64_md5_to_hex);
|
||||
|
||||
let (etag, etag_is_opaque) = match md5 {
|
||||
Some(md5) => (Some(md5), false),
|
||||
// A composite object has no MD5; its ETag describes the composition
|
||||
// rather than the bytes, so it is provenance only.
|
||||
None => (header(headers, "etag").map(str::to_string), true),
|
||||
};
|
||||
native_source_head(
|
||||
headers,
|
||||
METADATA_PREFIX,
|
||||
NativeHeadFields {
|
||||
etag,
|
||||
etag_is_opaque,
|
||||
version_id: header(headers, "x-goog-generation").map(str::to_string),
|
||||
storage_class: header(headers, "x-goog-storage-class").map(str::to_string),
|
||||
},
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl SourceBackend for GcsNativeSourceBackend {
|
||||
async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
|
||||
let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?;
|
||||
let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
|
||||
Self::head_from_response(response.headers())
|
||||
}
|
||||
|
||||
async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
|
||||
let mut headers = HeaderMap::new();
|
||||
if let Some(range) = range.map(range_header_value).transpose()? {
|
||||
headers.insert(
|
||||
http::header::RANGE,
|
||||
HeaderValue::from_str(&range).map_err(|_| SourceError::Other("invalid range header".to_string()))?,
|
||||
);
|
||||
}
|
||||
let request = self.request(Method::GET, self.object_url(key)?, headers).await?;
|
||||
let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
|
||||
let head = Self::head_from_response(response.headers())?;
|
||||
let content_range = header(response.headers(), "content-range").map(str::to_string);
|
||||
Ok(SourceGet {
|
||||
head,
|
||||
body: response_body(response),
|
||||
content_range,
|
||||
})
|
||||
}
|
||||
|
||||
async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
|
||||
// `objects.list` offers `startOffset`, which is inclusive, so it cannot
|
||||
// express "resume after this key" without silently repeating it.
|
||||
if request.start_after.is_some() {
|
||||
return Err(SourceError::Unsupported(
|
||||
"gcs sources cannot resume a listing from a key; use the continuation token".to_string(),
|
||||
));
|
||||
}
|
||||
let mut url = self.objects_url()?;
|
||||
{
|
||||
let mut query = url.query_pairs_mut();
|
||||
if let Some(prefix) = request.prefix.filter(|prefix| !prefix.is_empty()) {
|
||||
query.append_pair("prefix", prefix);
|
||||
}
|
||||
if let Some(delimiter) = request.delimiter.filter(|delimiter| !delimiter.is_empty()) {
|
||||
query.append_pair("delimiter", delimiter);
|
||||
}
|
||||
if let Some(token) = request.continuation_token.filter(|token| !token.is_empty()) {
|
||||
query.append_pair("pageToken", token);
|
||||
}
|
||||
if request.max_keys > 0 {
|
||||
query.append_pair("maxResults", &request.max_keys.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
let request = self.request(Method::GET, url, HeaderMap::new()).await?;
|
||||
let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
|
||||
let body = read_text(response, MAX_JSON_BYTES).await?;
|
||||
parse_objects_list(&body)
|
||||
}
|
||||
|
||||
/// GCS has no object tagging API; user metadata is already carried by the
|
||||
/// head mapping. An empty map keeps `policy.copy_tags` from failing a pull
|
||||
/// over a concept the provider does not have.
|
||||
async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
|
||||
Ok(HashMap::new())
|
||||
}
|
||||
|
||||
async fn probe(&self) -> Result<(), SourceError> {
|
||||
let request = self.request(Method::GET, self.bucket_url()?, HeaderMap::new()).await?;
|
||||
self.http.send(request, NO_ERROR_CODE_HEADER).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ObjectsList {
|
||||
#[serde(default)]
|
||||
items: Vec<ListedObject>,
|
||||
#[serde(default)]
|
||||
prefixes: Vec<String>,
|
||||
#[serde(default)]
|
||||
next_page_token: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ListedObject {
|
||||
name: String,
|
||||
/// GCS renders the size as a decimal string, not a JSON number.
|
||||
#[serde(default)]
|
||||
size: Option<String>,
|
||||
#[serde(default)]
|
||||
updated: Option<String>,
|
||||
#[serde(default)]
|
||||
md5_hash: Option<String>,
|
||||
#[serde(default)]
|
||||
etag: Option<String>,
|
||||
#[serde(default)]
|
||||
storage_class: Option<String>,
|
||||
}
|
||||
|
||||
fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
|
||||
let listing: ObjectsList =
|
||||
serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
|
||||
let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
|
||||
let objects = listing
|
||||
.items
|
||||
.into_iter()
|
||||
.map(|item| {
|
||||
let etag = item
|
||||
.md5_hash
|
||||
.as_deref()
|
||||
.and_then(base64_md5_to_hex)
|
||||
.or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string()))
|
||||
.filter(|etag| !etag.is_empty());
|
||||
SourceObject {
|
||||
key: item.name,
|
||||
etag,
|
||||
size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0),
|
||||
last_modified: item.updated.as_deref().and_then(parse_http_timestamp),
|
||||
storage_class: item.storage_class,
|
||||
// GCS never encodes a part count in a digest or an ETag.
|
||||
is_multipart_etag: false,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(SourcePage {
|
||||
objects,
|
||||
common_prefixes: listing.prefixes,
|
||||
is_truncated: next_continuation_token.is_some(),
|
||||
next_continuation_token,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
|
||||
use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
|
||||
use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder;
|
||||
|
||||
const LIST_PAGE_ONE: &str = r#"{
|
||||
"kind": "storage#objects",
|
||||
"nextPageToken": "cursor-1",
|
||||
"prefixes": ["dir/sub/"],
|
||||
"items": [
|
||||
{
|
||||
"name": "dir/a.txt",
|
||||
"size": "5",
|
||||
"updated": "2015-10-21T07:28:00.000Z",
|
||||
"md5Hash": "XUFAKrxLKna5cZ2REBfFkg==",
|
||||
"etag": "CJizy9Wq0McCEAE=",
|
||||
"storageClass": "STANDARD"
|
||||
}
|
||||
]
|
||||
}"#;
|
||||
|
||||
const LIST_PAGE_TWO: &str = r#"{
|
||||
"kind": "storage#objects",
|
||||
"items": [
|
||||
{
|
||||
"name": "dir/b.txt",
|
||||
"size": "7",
|
||||
"updated": "2015-10-21T07:28:00.000Z",
|
||||
"etag": "\"CJizy9Wq0McCEAI=\""
|
||||
}
|
||||
]
|
||||
}"#;
|
||||
|
||||
fn backend(endpoint: &Url) -> GcsNativeSourceBackend {
|
||||
GcsNativeSourceBackend {
|
||||
http: NativeHttp::for_test(endpoint.clone()),
|
||||
bucket: "legacy".to_string(),
|
||||
// Anonymous credentials add no headers, so the fixture sees exactly
|
||||
// the request this backend builds.
|
||||
credentials: AnonymousBuilder::new().build(),
|
||||
}
|
||||
}
|
||||
|
||||
fn object_headers() -> Vec<(&'static str, String)> {
|
||||
vec![
|
||||
("Content-Type", "text/plain".to_string()),
|
||||
("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
|
||||
("ETag", "\"CJizy9Wq0McCEAE=\"".to_string()),
|
||||
("x-goog-hash", "crc32c=AAAAAA==,md5=XUFAKrxLKna5cZ2REBfFkg==".to_string()),
|
||||
("x-goog-meta-owner", "alice".to_string()),
|
||||
("x-goog-storage-class", "STANDARD".to_string()),
|
||||
("x-goog-generation", "1445412480000000".to_string()),
|
||||
]
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn objects_list_maps_items_prefixes_and_the_page_token() {
|
||||
let page = parse_objects_list(LIST_PAGE_ONE).expect("page should parse");
|
||||
assert_eq!(page.common_prefixes, vec!["dir/sub/"]);
|
||||
assert!(page.is_truncated);
|
||||
assert_eq!(page.next_continuation_token.as_deref(), Some("cursor-1"));
|
||||
assert_eq!(page.objects.len(), 1);
|
||||
assert_eq!(page.objects[0].key, "dir/a.txt");
|
||||
assert_eq!(page.objects[0].size, 5, "the string size is parsed");
|
||||
assert_eq!(
|
||||
page.objects[0].etag.as_deref(),
|
||||
Some("5d41402abc4b2a76b9719d911017c592"),
|
||||
"the base64 md5Hash becomes a hex ETag"
|
||||
);
|
||||
assert_eq!(page.objects[0].storage_class.as_deref(), Some("STANDARD"));
|
||||
assert!(page.objects[0].last_modified.is_some(), "RFC 3339 `updated` is parsed");
|
||||
|
||||
let page = parse_objects_list(LIST_PAGE_TWO).expect("page should parse");
|
||||
assert!(!page.is_truncated);
|
||||
assert!(page.next_continuation_token.is_none());
|
||||
assert_eq!(
|
||||
page.objects[0].etag.as_deref(),
|
||||
Some("CJizy9Wq0McCEAI="),
|
||||
"without md5Hash the raw etag is carried"
|
||||
);
|
||||
|
||||
assert!(parse_objects_list("not json").is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn head_prefers_the_goog_hash_md5_over_the_etag() {
|
||||
let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, object_headers(), String::new())]).await;
|
||||
let head = backend(&endpoint).head("dir/a b.txt").await.expect("HEAD should map");
|
||||
|
||||
let recorded = recorded.lock().expect("recorder lock").clone();
|
||||
assert_eq!(recorded[0].method, "HEAD");
|
||||
assert_eq!(recorded[0].target, "/legacy/dir/a%20b.txt", "the XML API addresses the object by path");
|
||||
assert_eq!(
|
||||
head.etag.as_deref(),
|
||||
Some("5d41402abc4b2a76b9719d911017c592"),
|
||||
"the x-goog-hash md5 is the content digest"
|
||||
);
|
||||
assert!(!head.etag_is_opaque, "a GCS md5 may be checked against the pulled bytes");
|
||||
assert_eq!(head.user_metadata, HashMap::from([("owner".to_string(), "alice".to_string())]));
|
||||
assert_eq!(head.version_id.as_deref(), Some("1445412480000000"));
|
||||
assert_eq!(head.storage_class.as_deref(), Some("STANDARD"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_composite_object_without_an_md5_keeps_an_opaque_etag() {
|
||||
let headers = object_headers()
|
||||
.into_iter()
|
||||
.map(|(name, value)| {
|
||||
if name == "x-goog-hash" {
|
||||
(name, "crc32c=AAAAAA==".to_string())
|
||||
} else {
|
||||
(name, value)
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
|
||||
let head = backend(&endpoint).head("composed").await.expect("HEAD should map");
|
||||
assert_eq!(head.etag.as_deref(), Some("CJizy9Wq0McCEAE="));
|
||||
assert!(head.etag_is_opaque, "a composite ETag describes the composition, not the bytes");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn customer_supplied_key_objects_are_refused() {
|
||||
let mut headers = object_headers();
|
||||
headers.push(("x-goog-encryption-key-sha256", "abc".to_string()));
|
||||
let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
|
||||
let err = backend(&endpoint)
|
||||
.head("a.txt")
|
||||
.await
|
||||
.expect_err("CSEK objects are unsupported");
|
||||
assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_and_probe_address_the_json_api() {
|
||||
let (endpoint, recorded) = scripted_server(vec![
|
||||
ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
|
||||
])
|
||||
.await;
|
||||
let backend = backend(&endpoint);
|
||||
|
||||
backend
|
||||
.list(&SourceListRequest {
|
||||
prefix: Some("dir/"),
|
||||
delimiter: Some("/"),
|
||||
continuation_token: Some("cursor-0"),
|
||||
max_keys: 2,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect("listing should succeed");
|
||||
backend.probe().await.expect("probe should succeed");
|
||||
|
||||
let recorded = recorded.lock().expect("recorder lock").clone();
|
||||
assert!(recorded[0].target.starts_with("/storage/v1/b/legacy/o?"), "{}", recorded[0].target);
|
||||
assert!(recorded[0].target.contains("prefix=dir%2F"), "{}", recorded[0].target);
|
||||
assert!(recorded[0].target.contains("delimiter=%2F"), "{}", recorded[0].target);
|
||||
assert!(recorded[0].target.contains("pageToken=cursor-0"), "{}", recorded[0].target);
|
||||
assert!(recorded[0].target.contains("maxResults=2"), "{}", recorded[0].target);
|
||||
assert_eq!(recorded[1].target, "/storage/v1/b/legacy");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gcs_native_backend_satisfies_the_shared_backend_contract() {
|
||||
let mut ranged = object_headers();
|
||||
ranged.push(("Content-Range", "bytes 1-3/5".to_string()));
|
||||
// A HEAD reports the object size with no body, exactly as GCS does.
|
||||
let mut head_only = object_headers();
|
||||
head_only.push(("Content-Length", "5".to_string()));
|
||||
let (endpoint, _) = scripted_server(vec![
|
||||
ScriptedResponse::new(200, head_only, String::new()),
|
||||
ScriptedResponse::new(200, object_headers(), "hello".to_string()),
|
||||
ScriptedResponse::new(206, ranged, "ell".to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
|
||||
ScriptedResponse::new(200, Vec::new(), LIST_PAGE_TWO.to_string()),
|
||||
// GCS has no tagging call, so the contract's tag step issues no
|
||||
// request; the probe is the next one on the wire.
|
||||
ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
|
||||
ScriptedResponse::new(404, Vec::new(), String::new()),
|
||||
ScriptedResponse::new(403, Vec::new(), String::new()),
|
||||
])
|
||||
.await;
|
||||
|
||||
assert_backend_contract(
|
||||
&backend(&endpoint),
|
||||
BackendCapabilities {
|
||||
etag_is_opaque: false,
|
||||
supports_start_after: false,
|
||||
// GCS objects have no tags; the contract's tag step is skipped.
|
||||
supports_tagging: false,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
@@ -22,12 +22,15 @@
|
||||
//!
|
||||
//! A source is reached through one `SourceBackend`: the S3 dialect for every
|
||||
//! S3-compatible provider, and a native backend for the providers that have no
|
||||
//! S3 API (`azure`).
|
||||
//! S3 API (`azure`, `gcs_native`).
|
||||
|
||||
pub mod azure;
|
||||
#[cfg(test)]
|
||||
mod backend_contract;
|
||||
pub mod backfill;
|
||||
pub mod breaker;
|
||||
pub mod config;
|
||||
pub mod gcs;
|
||||
pub mod list_through;
|
||||
mod native_http;
|
||||
pub mod negative_cache;
|
||||
@@ -35,15 +38,17 @@ pub mod pull;
|
||||
pub mod source_client;
|
||||
pub mod stats;
|
||||
pub mod sys;
|
||||
#[cfg(test)]
|
||||
mod test_http_fixture;
|
||||
|
||||
pub use breaker::{
|
||||
BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker,
|
||||
BreakerState, BreakerTransition, BreakerVerdict,
|
||||
};
|
||||
pub use config::{
|
||||
ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
|
||||
OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
|
||||
SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
|
||||
AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
|
||||
ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider,
|
||||
RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
|
||||
};
|
||||
pub use list_through::{
|
||||
FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
|
||||
@@ -62,5 +67,6 @@ pub use stats::{
|
||||
};
|
||||
pub use sys::{
|
||||
ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError,
|
||||
OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec,
|
||||
OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec,
|
||||
source_client_spec,
|
||||
};
|
||||
|
||||
@@ -190,6 +190,14 @@ pub(super) async fn read_text(response: reqwest::Response, max_bytes: usize) ->
|
||||
String::from_utf8(body).map_err(|_| SourceError::Other("source listing response is not valid UTF-8".to_string()))
|
||||
}
|
||||
|
||||
/// Base64 digest (`Content-MD5`, `md5Hash`, `x-goog-hash`) as lowercase hex.
|
||||
/// `None` when the value is not a 16-byte digest, so a CRC32C never passes as
|
||||
/// an MD5.
|
||||
pub(super) fn base64_md5_to_hex(value: &str) -> Option<String> {
|
||||
let raw = base64_simd::STANDARD.decode_to_vec(value.trim().as_bytes()).ok()?;
|
||||
(raw.len() == 16).then(|| faster_hex::hex_string(&raw))
|
||||
}
|
||||
|
||||
pub(super) fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
|
||||
headers.get(name).and_then(|value| value.to_str().ok()).map(str::trim)
|
||||
}
|
||||
@@ -370,6 +378,17 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn base64_md5_converts_only_sixteen_byte_digests() {
|
||||
assert_eq!(
|
||||
base64_md5_to_hex("1B2M2Y8AsgTpgAmY7PhCfg==").as_deref(),
|
||||
Some("d41d8cd98f00b204e9800998ecf8427e")
|
||||
);
|
||||
assert_eq!(base64_md5_to_hex("not base64!").as_deref(), None);
|
||||
// A CRC32C digest is four bytes: it must not pass as an MD5.
|
||||
assert_eq!(base64_md5_to_hex("AAAAAA==").as_deref(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_http_rejects_endpoints_that_are_not_bare_origins() {
|
||||
for bad in [
|
||||
|
||||
@@ -26,6 +26,7 @@
|
||||
//! forwarded: v1 rejects SSE-C source objects outright.
|
||||
|
||||
use super::azure::AzureSourceBackend;
|
||||
use super::gcs::GcsNativeSourceBackend;
|
||||
use crate::bucket::remote_s3_client::{
|
||||
PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
|
||||
};
|
||||
@@ -716,8 +717,16 @@ impl SourceClient {
|
||||
)?;
|
||||
Ok(Self::from_backend(Box::new(backend), spec))
|
||||
}
|
||||
SourceBackendSpec::Gcs(_) => {
|
||||
Err(RemoteS3ClientError::Credentials("the native gcs source backend is not implemented yet"))
|
||||
SourceBackendSpec::Gcs(gcs) => {
|
||||
let backend = GcsNativeSourceBackend::new(
|
||||
&spec.endpoint,
|
||||
&spec.bucket,
|
||||
gcs,
|
||||
spec.timeouts,
|
||||
spec.skip_tls_verify,
|
||||
spec.ca_cert_pem.as_deref(),
|
||||
)?;
|
||||
Ok(Self::from_backend(Box::new(backend), spec))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -975,6 +984,7 @@ fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract};
|
||||
use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn};
|
||||
use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
|
||||
use aws_smithy_runtime_api::client::result::ConnectorError;
|
||||
@@ -1602,6 +1612,98 @@ mod tests {
|
||||
assert_eq!(SourceProvider::from_label("swift"), None);
|
||||
}
|
||||
|
||||
const CONTRACT_LIST_PAGE_ONE: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<Name>source-bucket</Name>
|
||||
<IsTruncated>true</IsTruncated>
|
||||
<NextContinuationToken>cursor-1</NextContinuationToken>
|
||||
<Contents>
|
||||
<Key>dir/a.txt</Key>
|
||||
<LastModified>2015-10-21T07:28:00.000Z</LastModified>
|
||||
<ETag>"5d41402abc4b2a76b9719d911017c592"</ETag>
|
||||
<Size>5</Size>
|
||||
<StorageClass>STANDARD</StorageClass>
|
||||
</Contents>
|
||||
<CommonPrefixes><Prefix>dir/sub/</Prefix></CommonPrefixes>
|
||||
</ListBucketResult>"#;
|
||||
|
||||
const CONTRACT_LIST_PAGE_TWO: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<Name>source-bucket</Name>
|
||||
<IsTruncated>false</IsTruncated>
|
||||
<Contents>
|
||||
<Key>dir/b.txt</Key>
|
||||
<LastModified>2015-10-21T07:28:00.000Z</LastModified>
|
||||
<ETag>"7d41402abc4b2a76b9719d911017c592"</ETag>
|
||||
<Size>7</Size>
|
||||
</Contents>
|
||||
</ListBucketResult>"#;
|
||||
|
||||
const CONTRACT_TAGGING: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><TagSet>
|
||||
<Tag><Key>env</Key><Value>prod</Value></Tag>
|
||||
</TagSet></Tagging>"#;
|
||||
|
||||
fn contract_object_headers(content_length: u64) -> Vec<(&'static str, String)> {
|
||||
vec![
|
||||
("etag", format!("\"{OBJECT_MD5}\"")),
|
||||
("content-length", content_length.to_string()),
|
||||
("content-type", "text/plain".to_string()),
|
||||
("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
|
||||
("x-amz-meta-owner", "alice".to_string()),
|
||||
("x-amz-storage-class", "STANDARD".to_string()),
|
||||
]
|
||||
}
|
||||
|
||||
/// The S3 backend behind the scripted connector, without the prefix-mapping
|
||||
/// client on top: the contract is a property of the backend itself.
|
||||
async fn scripted_s3_backend(responses: Vec<Scripted>) -> S3SourceBackend {
|
||||
let spec = spec(None);
|
||||
let connector = SharedHttpConnector::new(ScriptedConnector {
|
||||
requests: Arc::new(Mutex::new(Vec::new())),
|
||||
responses: Arc::new(Mutex::new(responses.into_iter().collect())),
|
||||
});
|
||||
let http_client = http_client_fn(move |_settings, _components| connector.clone());
|
||||
let endpoint = spec.endpoint_spec().expect("test spec endpoint should parse");
|
||||
let config = build_remote_s3_config(&endpoint)
|
||||
.await
|
||||
.expect("test spec should build")
|
||||
.http_client(http_client)
|
||||
.interceptor(SourceProxyMarkerInterceptor::new());
|
||||
S3SourceBackend {
|
||||
client: S3Client::from_conf(config.build()),
|
||||
bucket: spec.bucket.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn s3_backend_satisfies_the_shared_backend_contract() {
|
||||
let mut ranged = contract_object_headers(3);
|
||||
ranged.push(("content-range", "bytes 1-3/5".to_string()));
|
||||
let backend = scripted_s3_backend(vec![
|
||||
ok(contract_object_headers(5), ""),
|
||||
ok(contract_object_headers(5), "hello"),
|
||||
ok(ranged, "ell"),
|
||||
ok(Vec::new(), CONTRACT_LIST_PAGE_ONE),
|
||||
ok(Vec::new(), CONTRACT_LIST_PAGE_TWO),
|
||||
ok(Vec::new(), CONTRACT_TAGGING),
|
||||
ok(Vec::new(), ""),
|
||||
status(404, ""),
|
||||
status(403, ACCESS_DENIED_BODY),
|
||||
])
|
||||
.await;
|
||||
|
||||
assert_backend_contract(
|
||||
&backend,
|
||||
BackendCapabilities {
|
||||
etag_is_opaque: false,
|
||||
supports_start_after: true,
|
||||
supports_tagging: true,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
fn prefix_client(prefix: Option<String>) -> SourceClient {
|
||||
SourceClient {
|
||||
backend: Box::new(S3SourceBackend {
|
||||
|
||||
@@ -643,7 +643,7 @@ fn source_provider(provider: Provider) -> SourceProvider {
|
||||
/// falls back to the S3 spec, where the builder reports the missing
|
||||
/// credentials: the config layer already refuses to store that shape, so this
|
||||
/// only covers a config written by an older or hand-edited build.
|
||||
pub(crate) fn source_backend_spec(source: &SourceConfig) -> SourceBackendSpec {
|
||||
pub fn source_backend_spec(source: &SourceConfig) -> SourceBackendSpec {
|
||||
match (source.provider, source.azure.as_ref(), source.gcs.as_ref()) {
|
||||
(Provider::Azure, Some(azure), _) => SourceBackendSpec::Azure(AzureSourceSpec {
|
||||
account: azure.account.clone(),
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
// 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.
|
||||
|
||||
//! Scripted HTTP server for the native source backends' tests.
|
||||
//!
|
||||
//! The S3 backend can be driven through the SDK's own connector; the native
|
||||
//! backends talk to a real socket, so their tests need a server that answers a
|
||||
//! fixed script and records what it was asked. Every response closes its
|
||||
//! connection, which keeps one request on one socket and makes the script order
|
||||
//! exactly the request order.
|
||||
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use url::Url;
|
||||
|
||||
pub(super) struct ScriptedResponse {
|
||||
status: u16,
|
||||
headers: Vec<(&'static str, String)>,
|
||||
body: String,
|
||||
}
|
||||
|
||||
impl ScriptedResponse {
|
||||
pub(super) fn new(status: u16, headers: Vec<(&'static str, String)>, body: String) -> Self {
|
||||
Self { status, headers, body }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub(super) struct RecordedRequest {
|
||||
pub(super) method: String,
|
||||
/// Request target as it appeared on the wire: path plus query.
|
||||
pub(super) target: String,
|
||||
pub(super) headers: Vec<(String, String)>,
|
||||
}
|
||||
|
||||
impl RecordedRequest {
|
||||
pub(super) fn header(&self, name: &str) -> Option<&str> {
|
||||
self.headers
|
||||
.iter()
|
||||
.find(|(key, _)| key.eq_ignore_ascii_case(name))
|
||||
.map(|(_, value)| value.as_str())
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) type Recorder = Arc<Mutex<Vec<RecordedRequest>>>;
|
||||
|
||||
/// Binds a loopback listener that answers `responses` in order and returns its
|
||||
/// origin plus the recorder. The task ends once the script is exhausted.
|
||||
pub(super) async fn scripted_server(responses: Vec<ScriptedResponse>) -> (Url, Recorder) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind");
|
||||
let port = listener.local_addr().expect("fixture address").port();
|
||||
let recorder: Recorder = Arc::new(Mutex::new(Vec::new()));
|
||||
let sink = Arc::clone(&recorder);
|
||||
|
||||
tokio::spawn(async move {
|
||||
for response in responses {
|
||||
let Ok((mut stream, _)) = listener.accept().await else {
|
||||
return;
|
||||
};
|
||||
let mut request = Vec::new();
|
||||
let mut buffer = [0_u8; 2048];
|
||||
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
|
||||
match stream.read(&mut buffer).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(read) => request.extend_from_slice(&buffer[..read]),
|
||||
}
|
||||
}
|
||||
let text = String::from_utf8_lossy(&request).into_owned();
|
||||
let mut lines = text.lines();
|
||||
let start = lines.next().unwrap_or_default().to_string();
|
||||
let mut parts = start.split_whitespace();
|
||||
sink.lock().expect("recorder lock").push(RecordedRequest {
|
||||
method: parts.next().unwrap_or_default().to_string(),
|
||||
target: parts.next().unwrap_or_default().to_string(),
|
||||
headers: lines
|
||||
.take_while(|line| !line.is_empty())
|
||||
.filter_map(|line| line.split_once(':'))
|
||||
.map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
|
||||
.collect(),
|
||||
});
|
||||
|
||||
// A scripted HEAD declares the object size in its own headers while
|
||||
// carrying no body, so an explicit `Content-Length` wins over the
|
||||
// body length.
|
||||
let declares_length = response
|
||||
.headers
|
||||
.iter()
|
||||
.any(|(name, _)| name.eq_ignore_ascii_case("content-length"));
|
||||
let mut rendered = match declares_length {
|
||||
true => format!("HTTP/1.1 {} Scripted\r\nConnection: close\r\n", response.status),
|
||||
false => format!(
|
||||
"HTTP/1.1 {} Scripted\r\nContent-Length: {}\r\nConnection: close\r\n",
|
||||
response.status,
|
||||
response.body.len()
|
||||
),
|
||||
};
|
||||
for (name, value) in response.headers {
|
||||
rendered.push_str(&format!("{name}: {value}\r\n"));
|
||||
}
|
||||
rendered.push_str("\r\n");
|
||||
rendered.push_str(&response.body);
|
||||
let _ = stream.write_all(rendered.as_bytes()).await;
|
||||
let _ = stream.flush().await;
|
||||
}
|
||||
});
|
||||
|
||||
(Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorder)
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}
|
||||
{"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},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}
|
||||
{"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},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}
|
||||
{"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},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}
|
||||
|
||||
@@ -78,10 +78,18 @@ pub struct OnDemandMigrationSource {
|
||||
#[serde(default)]
|
||||
pub path_style: OnDemandMigrationPathStyle,
|
||||
/// `None` means anonymous access to a public source bucket.
|
||||
/// `None` means anonymous access to a public source bucket. The native
|
||||
/// providers carry their credentials in `azure` / `gcs` instead.
|
||||
#[serde(default)]
|
||||
pub credentials: Option<OnDemandMigrationCredentials>,
|
||||
#[serde(default)]
|
||||
pub tls: OnDemandMigrationTls,
|
||||
/// Required for `azure` and rejected for every other provider.
|
||||
#[serde(default)]
|
||||
pub azure: Option<OnDemandMigrationAzure>,
|
||||
/// Required for `gcs_native` and rejected for every other provider.
|
||||
#[serde(default)]
|
||||
pub gcs: Option<OnDemandMigrationGcs>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
@@ -92,7 +100,49 @@ pub enum OnDemandMigrationProvider {
|
||||
Minio,
|
||||
Rustfs,
|
||||
R2,
|
||||
/// GCS XML interoperability API with HMAC keys.
|
||||
Gcs,
|
||||
/// Native Azure Blob service.
|
||||
Azure,
|
||||
/// Native GCS JSON API with a service-account key.
|
||||
#[serde(rename = "gcs_native")]
|
||||
GcsNative,
|
||||
}
|
||||
|
||||
/// Native Azure Blob parameters. The container is `source.bucket`; exactly one
|
||||
/// of `account_key` and `sas_token` is set. Responses carry both as `REDACTED`.
|
||||
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct OnDemandMigrationAzure {
|
||||
pub account: String,
|
||||
#[serde(default)]
|
||||
pub account_key: Option<String>,
|
||||
#[serde(default)]
|
||||
pub sas_token: Option<String>,
|
||||
}
|
||||
|
||||
impl fmt::Debug for OnDemandMigrationAzure {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("OnDemandMigrationAzure")
|
||||
.field("account", &self.account)
|
||||
.field("account_key", &self.account_key.as_ref().map(|_| "REDACTED"))
|
||||
.field("sas_token", &self.sas_token.as_ref().map(|_| "REDACTED"))
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
/// Native GCS parameters. The bucket is `source.bucket`; the key JSON embeds a
|
||||
/// private key, so responses carry it as `REDACTED`.
|
||||
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct OnDemandMigrationGcs {
|
||||
pub service_account_json: String,
|
||||
}
|
||||
|
||||
impl fmt::Debug for OnDemandMigrationGcs {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("OnDemandMigrationGcs")
|
||||
.field("service_account_json", &"REDACTED")
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
|
||||
@@ -806,6 +856,8 @@ mod tests {
|
||||
session_token: None,
|
||||
}),
|
||||
tls: OnDemandMigrationTls::default(),
|
||||
azure: None,
|
||||
gcs: None,
|
||||
});
|
||||
let mut expected: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).expect("fixture");
|
||||
expected.filter.source_prefix = None;
|
||||
@@ -821,6 +873,42 @@ mod tests {
|
||||
assert!(minimal.source.credentials.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn native_provider_documents_round_trip_and_hide_their_secrets() {
|
||||
for (label, json) in [
|
||||
(
|
||||
"azure",
|
||||
r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"},"gcs":null}"#,
|
||||
),
|
||||
(
|
||||
"gcs_native",
|
||||
r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#,
|
||||
),
|
||||
] {
|
||||
let source: OnDemandMigrationSource = serde_json::from_str(json).unwrap_or_else(|err| panic!("{label}: {err}"));
|
||||
assert_eq!(
|
||||
serde_json::to_string(&source).expect("re-encodes"),
|
||||
json,
|
||||
"{label} must reproduce the server wire shape byte for byte"
|
||||
);
|
||||
}
|
||||
|
||||
let azure = OnDemandMigrationAzure {
|
||||
account: "legacyaccount".to_string(),
|
||||
account_key: Some("c2VjcmV0".to_string()),
|
||||
sas_token: Some("sig=topsecret".to_string()),
|
||||
};
|
||||
let rendered = format!("{azure:?}");
|
||||
assert!(rendered.contains("legacyaccount"));
|
||||
assert!(!rendered.contains("c2VjcmV0"), "{rendered}");
|
||||
assert!(!rendered.contains("topsecret"), "{rendered}");
|
||||
|
||||
let gcs = OnDemandMigrationGcs {
|
||||
service_account_json: r#"{"private_key":"-----BEGIN PRIVATE KEY-----"}"#.to_string(),
|
||||
};
|
||||
assert!(!format!("{gcs:?}").contains("PRIVATE KEY"), "{gcs:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn credentials_debug_never_prints_secrets() {
|
||||
let credentials = OnDemandMigrationCredentials {
|
||||
|
||||
@@ -91,14 +91,20 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
|
||||
|---|---|---|---|
|
||||
| `version` | integer | `1` | Must be `1` |
|
||||
| `enabled` | bool | `true` | `false` keeps the config but stops all source traffic |
|
||||
| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` | — (required) | Drives endpoint and addressing defaults |
|
||||
| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required for every provider except `aws`, where it is derived from `region` |
|
||||
| `source.region` | string | — (required) | Non-empty. `auto` is accepted only for `r2`, `minio`, `rustfs` and is signed as `us-east-1` |
|
||||
| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace |
|
||||
| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` \| `azure` \| `gcs_native` | — (required) | Drives endpoint and addressing defaults, and which backend the client builds: every value but `azure` and `gcs_native` speaks S3 |
|
||||
| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required except for `aws` (derived from `region`), `azure` (derived as `https://<account>.blob.core.windows.net`) and `gcs_native` (`https://storage.googleapis.com`). Set it explicitly to point at Azurite or fake-gcs-server |
|
||||
| `source.region` | string | — (required) | Non-empty. `auto` is accepted for `r2`, `minio`, `rustfs` and for the native providers, and is signed as `us-east-1`. `azure` and `gcs_native` never sign with a region, so `auto` is the honest value there |
|
||||
| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace. For `azure` this is the container name, for `gcs_native` the bucket name; the provider block never repeats it |
|
||||
| `source.path_style` | `auto` \| `path` \| `virtual` | `auto` | `auto` resolves to path-style for IP-literal or `localhost` endpoints and for `s3`/`minio`/`rustfs`; virtual-host for `aws`/`gcs`/`r2` |
|
||||
| `source.credentials` | object \| null | `null` | `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present |
|
||||
| `source.tls.skip_verify` | bool | `false` | Disables certificate verification for the source connection |
|
||||
| `source.tls.ca_cert_pem` | string \| null | `null` | Must contain `-----BEGIN CERTIFICATE-----` |
|
||||
| `source.azure` | object \| null | `null` | Required for `provider = "azure"` and rejected for every other provider |
|
||||
| `source.azure.account` | string | — (required) | Storage account name; `[A-Za-z0-9-]` only, because it becomes the first label of the derived host |
|
||||
| `source.azure.account_key` | string \| null | `null` | Base64 storage-account key, signed per request with Shared Key. Mutually exclusive with `sas_token`; exactly one of the two is required |
|
||||
| `source.azure.sas_token` | string \| null | `null` | SAS query string without the leading `?` and without whitespace, appended to every request URL |
|
||||
| `source.gcs` | object \| null | `null` | Required for `provider = "gcs_native"` and rejected for every other provider |
|
||||
| `source.gcs.service_account_json` | string | — (required) | Service-account key JSON; must parse and carry `type: service_account`, `client_email` and `private_key`. Tokens are minted read-only (`devstorage.read_only`) |
|
||||
| `filter.prefix` | string \| null | `null` | Null or non-empty. Only local keys with this prefix consult the source |
|
||||
| `filter.source_prefix` | string \| null | `null` | Null or non-empty. Prepended to the local key to form the source key |
|
||||
| `policy.head` | `proxy` \| `local_only` | `proxy` | `local_only` answers a HEAD miss with 404 and no source traffic |
|
||||
@@ -107,7 +113,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
|
||||
| `policy.list_through` | bool | `false` | Merges the source listing into `ListObjectsV2` so clients see the whole namespace during the migration. Off by default: it puts the source in the path of every listing |
|
||||
| `policy.respect_local_delete_marker` | bool | `true` | A local delete marker is the final answer; only a versioned bucket can produce one |
|
||||
| `policy.preserve_etag` | bool | `true` | Keeps the source ETag on the stored object unless the bucket encrypts by default |
|
||||
| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull |
|
||||
| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull. `azure` reads blob tags instead; `gcs_native` has no tags and always finds none |
|
||||
| `policy.emit_events` | bool | `true` | Whether a write-back emits `ObjectCreated` notifications |
|
||||
| `policy.negative_cache_ttl_secs` | integer | `30` | `0..=3600`; `0` disables the negative cache |
|
||||
| `policy.inline_max_bytes` | integer | `16777216` (16 MiB) | `0..=268435456` (256 MiB). At or below this size a GET miss is teed inline; above it the response streams through and a background pull stores the object |
|
||||
@@ -133,8 +139,14 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket
|
||||
| `rustfs` | Required | Path-style | `auto` allowed | A RustFS source answers the migration request locally thanks to the anti-loop marker | `real_source_test.rs` in the `e2e-nightly` lane |
|
||||
| `r2` | `https://<account-id>.r2.cloudflarestorage.com` | Virtual-host | `auto` allowed (signed as `us-east-1`) | | `cloud-source (r2)`, only while `ODM_INTEROP_R2_*` are configured; no difference recorded yet |
|
||||
| `gcs` | `https://storage.googleapis.com` | Virtual-host | Real region required | Uses the GCS XML interoperability API with an HMAC key pair, not a service-account JSON key | `cloud-source (gcs)`, only while `ODM_INTEROP_GCS_HMAC_*` are configured; no difference recorded yet |
|
||||
| `azure` | Optional; derived as `https://<account>.blob.core.windows.net` | Native Blob REST, not S3 | Unused; write `auto` | Needs `source.azure`; the container is `source.bucket`. Reads need `Read` on the blob and `List` on the container, plus `Tags` when `policy.copy_tags` is on | None yet: no interop job covers Azure |
|
||||
| `gcs_native` | Optional; derived as `https://storage.googleapis.com` | Native GCS API, not S3 | Unused; write `auto` | Needs `source.gcs`. Reads use the XML API for objects and `objects.list` for listings, both with an OAuth token minted from the service-account key; the key needs `storage.objects.get` and `storage.objects.list` | None yet: no interop job covers native GCS |
|
||||
|
||||
Azure Blob has no preset; a native provider is deferred (rustfs/backlog#2166).
|
||||
Every backend answers the same trait contract, pinned by `backend_contract.rs` in `crates/ecstore/src/bucket/on_demand_migration/`, and the three differences that contract allows are the ones documented here.
|
||||
|
||||
`azure` differs in two of them. Its ETag is a concurrency token rather than a digest of the bytes, so it is stored as `odm-source-etag` provenance and never used as the expected MD5 of a pulled object — the write-back integrity check falls back to the local digest. And its listing paginates only with an opaque marker: there is no "start after this key" form, so a caller that asks for one gets `Unsupported` instead of a listing that silently starts over.
|
||||
|
||||
`gcs_native` differs in the other two. Its listing also has no exclusive "start after" form (`startOffset` is inclusive), so it refuses one the same way. And GCS has no object tagging at all: `policy.copy_tags` finds no tags rather than failing the pull, because GCS custom metadata is already carried by the head mapping. Its ETag is normally usable: the `x-goog-hash` MD5 is converted to hex and checked against the pulled bytes, except on a composite object, which has no MD5 and whose ETag is then treated as opaque.
|
||||
|
||||
The "Interop evidence" column names the job in `.github/workflows/on-demand-migration-interop.yml` (rustfs/backlog#2167) that last exercised the preset against a real implementation, and is where a provider difference belongs once the lane finds one. That lane is report-only and scheduled: it runs `crates/e2e_test/src/on_demand_migration/interop_test.rs` — the same case bodies as the merge-gate suite, with the source injected through `RUSTFS_ODM_INTEROP_*` — against a pinned MinIO container, and against each cloud provider whose repository secrets are configured. A provider without secrets is skipped with a note in the run summary rather than failing, so "no difference recorded yet" means exactly that and not "verified clean"; see [ci-gates.md](../testing/ci-gates.md) for the row.
|
||||
|
||||
|
||||
@@ -46,6 +46,7 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{
|
||||
};
|
||||
use crate::admin::storage_api::bucket::on_demand_migration::{
|
||||
OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext,
|
||||
source_backend_spec,
|
||||
};
|
||||
use crate::admin::storage_api::bucket::remote_s3_client::{
|
||||
PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
|
||||
@@ -585,6 +586,8 @@ fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider {
|
||||
Provider::Rustfs => SourceProvider::Rustfs,
|
||||
Provider::R2 => SourceProvider::R2,
|
||||
Provider::Gcs => SourceProvider::Gcs,
|
||||
Provider::Azure => SourceProvider::Azure,
|
||||
Provider::GcsNative => SourceProvider::GcsNative,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -621,6 +624,9 @@ pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClie
|
||||
// a flapping source behind a success and triple the probe's cost.
|
||||
retry: RemoteS3RetryPolicy::Disabled,
|
||||
bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
|
||||
// One mapping serves the probe and the runtime, so an admin probe
|
||||
// always exercises the backend the runtime will build.
|
||||
backend: source_backend_spec(source),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -292,6 +292,7 @@ pub(crate) mod on_demand_migration {
|
||||
pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle;
|
||||
pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider;
|
||||
pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>;
|
||||
pub(crate) use super::ecstore_bucket::on_demand_migration::source_backend_spec;
|
||||
|
||||
pub(crate) mod backfill {
|
||||
pub(crate) type BackfillCheckpoint = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillCheckpoint;
|
||||
|
||||
@@ -4821,6 +4821,8 @@ mod on_demand_migration_tests {
|
||||
session_token: None,
|
||||
}),
|
||||
tls: TlsConfig::default(),
|
||||
azure: None,
|
||||
gcs: None,
|
||||
},
|
||||
filter: FilterConfig {
|
||||
prefix: None,
|
||||
|
||||
@@ -665,6 +665,8 @@ mod tests {
|
||||
session_token: None,
|
||||
}),
|
||||
tls: TlsConfig::default(),
|
||||
azure: None,
|
||||
gcs: None,
|
||||
},
|
||||
filter: FilterConfig {
|
||||
prefix: None,
|
||||
@@ -743,6 +745,7 @@ mod tests {
|
||||
},
|
||||
),
|
||||
is_multipart_etag: true,
|
||||
etag_is_opaque: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -115,6 +115,12 @@ pub(super) fn expected_md5_hex(head: &SourceHead) -> Option<String> {
|
||||
if head.sse.is_some() {
|
||||
return None;
|
||||
}
|
||||
// Azure stamps an opaque concurrency token in the ETag slot. It is
|
||||
// recorded as provenance, but reading it as a digest would compare the
|
||||
// pulled bytes against a value that never described them.
|
||||
if head.etag_is_opaque {
|
||||
return None;
|
||||
}
|
||||
let etag = head.etag.as_deref()?;
|
||||
if etag.len() != 32 || is_multipart_etag(etag) || !etag.bytes().all(|byte| byte.is_ascii_hexdigit()) {
|
||||
return None;
|
||||
@@ -874,6 +880,12 @@ mod tests {
|
||||
head.sse = None;
|
||||
head.etag = None;
|
||||
assert_eq!(expected_md5_hex(&head), None);
|
||||
|
||||
// An Azure ETag can be any string the service chooses; even one that
|
||||
// happens to look like an MD5 must not be checked against the bytes.
|
||||
let mut head = source_head(b"abc");
|
||||
head.etag_is_opaque = true;
|
||||
assert_eq!(expected_md5_hex(&head), None, "opaque provider ETag");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user