This is an automated email from the ASF dual-hosted git repository. xuanwo pushed a commit to branch add-list-objects-v1-support in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 72894f529f498307d3d632d59ce5a9cf6e68a6ec Author: Xuanwo <[email protected]> AuthorDate: Mon Apr 7 17:07:02 2025 +0800 Implement list objects v1 Signed-off-by: Xuanwo <[email protected]> --- .../s3/aws_s3_with_list_objects_v1/action.yml | 39 ++++++ .../s3/minio_s3_with_list_objects_v1/action.yml | 47 ++++++++ core/src/raw/enum_utils.rs | 10 ++ core/src/services/s3/backend.rs | 13 +- core/src/services/s3/config.rs | 5 + core/src/services/s3/core.rs | 119 ++++++++++++++++++- core/src/services/s3/lister.rs | 132 ++++++++++++++++++++- 7 files changed, 358 insertions(+), 7 deletions(-) diff --git a/.github/services/s3/aws_s3_with_list_objects_v1/action.yml b/.github/services/s3/aws_s3_with_list_objects_v1/action.yml new file mode 100644 index 000000000..bce43e1aa --- /dev/null +++ b/.github/services/s3/aws_s3_with_list_objects_v1/action.yml @@ -0,0 +1,39 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +name: aws_s3 +description: "Behavior test for AWS S3. This service is sponsored by @datafuse_labs." + +runs: + using: "composite" + steps: + - name: Setup + uses: 1password/load-secrets-action@v1 + with: + export-env: true + env: + OPENDAL_S3_ROOT: op://services/s3/root + OPENDAL_S3_BUCKET: op://services/s3/bucket + OPENDAL_S3_ENDPOINT: op://services/s3/endpoint + OPENDAL_S3_ACCESS_KEY_ID: op://services/s3/access_key_id + OPENDAL_S3_SECRET_ACCESS_KEY: op://services/s3/secret_access_key + OPENDAL_S3_REGION: op://services/s3/region + + - name: Add extra settings + shell: bash + run: | + echo "OPENDAL_S3_DISABLE_LIST_OBJECTS_V2=true" >> $GITHUB_ENV diff --git a/.github/services/s3/minio_s3_with_list_objects_v1/action.yml b/.github/services/s3/minio_s3_with_list_objects_v1/action.yml new file mode 100644 index 000000000..46904d7aa --- /dev/null +++ b/.github/services/s3/minio_s3_with_list_objects_v1/action.yml @@ -0,0 +1,47 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +name: minio_s3_with_versioning +description: "Behavior test for Minio S3 with bucket versioning enabled" + +runs: + using: "composite" + steps: + - name: Setup MinIO Server + shell: bash + working-directory: fixtures/s3 + run: docker compose -f docker-compose-minio.yml up -d --wait + - name: Setup test bucket + shell: bash + env: + AWS_ACCESS_KEY_ID: "minioadmin" + AWS_SECRET_ACCESS_KEY: "minioadmin" + AWS_EC2_METADATA_DISABLED: "true" + run: | + aws --endpoint-url http://127.0.0.1:9000/ s3 mb s3://test + + - name: Setup + shell: bash + run: | + cat << EOF >> $GITHUB_ENV + OPENDAL_S3_BUCKET=test + OPENDAL_S3_ENDPOINT=http://127.0.0.1:9000 + OPENDAL_S3_ACCESS_KEY_ID=minioadmin + OPENDAL_S3_SECRET_ACCESS_KEY=minioadmin + OPENDAL_S3_REGION=us-east-1 + OPENDAL_S3_DISABLE_LIST_OBJECTS_V2=true + EOF diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs index e95059975..9a539943b 100644 --- a/core/src/raw/enum_utils.rs +++ b/core/src/raw/enum_utils.rs @@ -163,6 +163,16 @@ impl<ONE: oio::Write, TWO: oio::Write, THREE: oio::Write> oio::Write } } +impl<ONE: oio::List, TWO: oio::List, THREE: oio::List> oio::List for ThreeWays<ONE, TWO, THREE> { + async fn next(&mut self) -> Result<Option<oio::Entry>> { + match self { + Self::One(v) => v.next().await, + Self::Two(v) => v.next().await, + Self::Three(v) => v.next().await, + } + } +} + /// FourWays is used to implement traits that based on four ways. /// /// Users can wrap four different trait types together. diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs index b696793fb..0a46ebf03 100644 --- a/core/src/services/s3/backend.rs +++ b/core/src/services/s3/backend.rs @@ -43,7 +43,7 @@ use std::sync::LazyLock; use super::core::*; use super::delete::S3Deleter; use super::error::parse_error; -use super::lister::{S3ListerV2, S3Listers, S3ObjectVersionsLister}; +use super::lister::{S3ListerV1, S3ListerV2, S3Listers, S3ObjectVersionsLister}; use super::writer::S3Writer; use super::writer::S3Writers; use crate::raw::oio::PageLister; @@ -994,6 +994,7 @@ impl Builder for S3Builder { server_side_encryption_customer_key_md5, default_storage_class, allow_anonymous: self.config.allow_anonymous, + disable_list_objects_v2: self.config.disable_list_objects_v2, signer, loader, credential_loaded: AtomicBool::new(false), @@ -1089,13 +1090,19 @@ impl Access for S3Backend { async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> { let l = if args.versions() || args.deleted() { - TwoWays::Two(PageLister::new(S3ObjectVersionsLister::new( + ThreeWays::Three(PageLister::new(S3ObjectVersionsLister::new( + self.core.clone(), + path, + args, + ))) + } else if self.core.disable_list_objects_v2 { + ThreeWays::One(PageLister::new(S3ListerV1::new( self.core.clone(), path, args, ))) } else { - TwoWays::One(PageLister::new(S3ListerV2::new( + ThreeWays::Two(PageLister::new(S3ListerV2::new( self.core.clone(), path, args, diff --git a/core/src/services/s3/config.rs b/core/src/services/s3/config.rs index 74b78a03c..de2edc94c 100644 --- a/core/src/services/s3/config.rs +++ b/core/src/services/s3/config.rs @@ -188,6 +188,11 @@ pub struct S3Config { /// Enable write with append so that opendal will send write request with append headers. pub enable_write_with_append: bool, + + /// OpenDAL uses List Objects V2 by default to list objects. + /// However, some legacy services do not yet support V2. + /// This option allows users to switch back to the older List Objects V1. + pub disable_list_objects_v2: bool, } impl Debug for S3Config { diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs index 60a2db6b4..9314e0fc4 100644 --- a/core/src/services/s3/core.rs +++ b/core/src/services/s3/core.rs @@ -97,6 +97,7 @@ pub struct S3Core { pub server_side_encryption_customer_key_md5: Option<HeaderValue>, pub default_storage_class: Option<HeaderValue>, pub allow_anonymous: bool, + pub disable_list_objects_v2: bool, pub signer: AwsV4Signer, pub loader: Box<dyn AwsCredentialLoad>, @@ -676,6 +677,42 @@ impl S3Core { self.send(req).await } + pub async fn s3_list_objects_v1( + &self, + path: &str, + marker: &str, + delimiter: &str, + limit: Option<usize>, + ) -> Result<Response<Buffer>> { + let p = build_abs_path(&self.root, path); + + let mut url = self.endpoint.to_string(); + if !p.is_empty() { + write!(url, "&prefix={}", percent_encode_path(&p)) + .expect("write into string must succeed"); + } + if !delimiter.is_empty() { + write!(url, "&delimiter={delimiter}").expect("write into string must succeed"); + } + if let Some(limit) = limit { + write!(url, "&max-keys={limit}").expect("write into string must succeed"); + } + if !marker.is_empty() { + write!(url, "&marker={}", percent_encode_path(marker)) + .expect("write into string must succeed"); + } + + let mut req = Request::get(&url) + // Inject operation to the request. + .extension(Operation::List) + .body(Buffer::new()) + .map_err(new_request_build_error)?; + + self.sign(&mut req).await?; + + self.send(req).await + } + pub async fn s3_list_objects_v2( &self, path: &str, @@ -1070,6 +1107,25 @@ pub struct DeleteObjectsResultError { pub version_id: Option<String>, } +/// Output of ListBucket/ListObjects (a.k.a ListObjectsV1). +#[derive(Default, Debug, Deserialize)] +#[serde(default, rename_all = "PascalCase")] +pub struct ListObjectsOutputV1 { + pub is_truncated: Option<bool>, + /// ## Notes + /// + /// `next_marker` is returned only if we have the delimiter request parameter + /// specified. If the response does not include the NextMarker element and it + /// is truncated, we should use the value of the last Key element in the + /// response as the marker parameter in the subsequent request to get the + /// next set of object keys. + /// + /// If the contents is empty, we should find common_prefixes instead. + pub next_marker: Option<String>, + pub common_prefixes: Vec<OutputCommonPrefix>, + pub contents: Vec<ListObjectsOutputContent>, +} + /// Output of ListBucketV2/ListObjectsV2. /// /// ## Note @@ -1368,8 +1424,69 @@ mod tests { assert_eq!(out.error.len(), 0); } + /// This example is from https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjects.html#API_ListObjects_Examples + #[test] + fn test_parse_list_output_v1() { + let bs = bytes::Bytes::from( + r#"<?xml version="1.0" encoding="UTF-8"?> + <ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/";> + <Name>bucket</Name> + <Prefix/> + <Marker/> + <MaxKeys>1000</MaxKeys> + <IsTruncated>false</IsTruncated> + <Contents> + <Key>my-image.jpg</Key> + <LastModified>2009-10-12T17:50:30.000Z</LastModified> + <ETag>"fba9dede5f27731c9771645a39863328"</ETag> + <Size>434234</Size> + <StorageClass>STANDARD</StorageClass> + <Owner> + <ID>75aa57f09aa0c8caeab4f8c24e99d10f8e7faeebf76c078efc7c6caea54ba06a</ID> + <DisplayName>[email protected]</DisplayName> + </Owner> + </Contents> + <Contents> + <Key>my-third-image.jpg</Key> + <LastModified>2009-10-12T17:50:30.000Z</LastModified> + <ETag>"1b2cf535f27731c974343645a3985328"</ETag> + <Size>64994</Size> + <StorageClass>STANDARD_IA</StorageClass> + <Owner> + <ID>75aa57f09aa0c8caeab4f8c24e99d10f8e7faeebf76c078efc7c6caea54ba06a</ID> + <DisplayName>[email protected]</DisplayName> + </Owner> + </Contents> + </ListBucketResult>"#, + ); + + let out: ListObjectsOutputV1 = + quick_xml::de::from_reader(bs.reader()).expect("must success"); + + assert!(!out.is_truncated.unwrap()); + assert!(out.next_marker.is_none()); + assert!(out.common_prefixes.is_empty()); + assert_eq!( + out.contents, + vec![ + ListObjectsOutputContent { + key: "my-image.jpg".to_string(), + size: 434234, + etag: Some("\"fba9dede5f27731c9771645a39863328\"".to_string()), + last_modified: "2009-10-12T17:50:30.000Z".to_string(), + }, + ListObjectsOutputContent { + key: "my-third-image.jpg".to_string(), + size: 64994, + last_modified: "2009-10-12T17:50:30.000Z".to_string(), + etag: Some("\"1b2cf535f27731c974343645a3985328\"".to_string()), + }, + ] + ) + } + #[test] - fn test_parse_list_output() { + fn test_parse_list_output_v2() { let bs = bytes::Bytes::from( r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/";> <Name>example-bucket</Name> diff --git a/core/src/services/s3/lister.rs b/core/src/services/s3/lister.rs index f1bf4b5ef..b5d3703f4 100644 --- a/core/src/services/s3/lister.rs +++ b/core/src/services/s3/lister.rs @@ -17,8 +17,7 @@ use std::sync::Arc; -use super::core::S3Core; -use super::core::{ListObjectVersionsOutput, ListObjectsOutputV2}; +use super::core::*; use super::error::parse_error; use crate::raw::oio::PageContext; use crate::raw::*; @@ -29,7 +28,134 @@ use crate::Result; use bytes::Buf; use quick_xml::de; -pub type S3Listers = TwoWays<oio::PageLister<S3ListerV2>, oio::PageLister<S3ObjectVersionsLister>>; +pub type S3Listers = ThreeWays< + oio::PageLister<S3ListerV1>, + oio::PageLister<S3ListerV2>, + oio::PageLister<S3ObjectVersionsLister>, +>; + +/// S3ListerV1 implements ListObjectV1 for s3 backend. +pub struct S3ListerV1 { + core: Arc<S3Core>, + + path: String, + args: OpList, + + delimiter: &'static str, + /// marker can also be used as `start-after` for list objects v1. + /// We will use it as `start-after` for the first page and then ignore + /// it in the following pages. + first_marker: String, +} + +impl S3ListerV1 { + pub fn new(core: Arc<S3Core>, path: &str, args: OpList) -> Self { + let delimiter = if args.recursive() { "" } else { "/" }; + let first_marker = args + .start_after() + .map(|start_after| build_abs_path(&core.root, start_after)) + .unwrap_or_default(); + + Self { + core, + + path: path.to_string(), + args, + delimiter, + first_marker, + } + } +} + +impl oio::PageList for S3ListerV1 { + async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> { + let resp = self + .core + .s3_list_objects_v1( + &self.path, + // `marker` is used as `start-after` for the first page. + if !ctx.token.is_empty() { + &ctx.token + } else { + &self.first_marker + }, + self.delimiter, + self.args.limit(), + ) + .await?; + + if resp.status() != http::StatusCode::OK { + return Err(parse_error(resp)); + } + let bs = resp.into_body(); + + let output: ListObjectsOutputV1 = de::from_reader(bs.reader()) + .map_err(new_xml_deserialize_error) + // Allow S3 list to retry on XML deserialization errors. + // + // This is because the S3 list API may return incomplete XML data under high load. + // We are confident that our XML decoding logic is correct. When this error occurs, + // we allow retries to obtain the correct data. + .map_err(Error::set_temporary)?; + + // Try our best to check whether this list is done. + // + // - Check `is_truncated` + // - Check the length of `common_prefixes` and `contents` (very rare case) + ctx.done = if let Some(is_truncated) = output.is_truncated { + !is_truncated + } else { + output.common_prefixes.is_empty() && output.contents.is_empty() + }; + // Try out best to find the next marker. + // + // - Check `next-marker` + // - Check the last object key + // - Check the last common prefix + ctx.token = if let Some(next_marker) = &output.next_marker { + next_marker.clone() + } else if let Some(content) = output.contents.last() { + content.key.clone() + } else if let Some(prefix) = output.common_prefixes.last() { + prefix.prefix.clone() + } else { + "".to_string() + }; + + for prefix in output.common_prefixes { + let de = oio::Entry::new( + &build_rel_path(&self.core.root, &prefix.prefix), + Metadata::new(EntryMode::DIR), + ); + + ctx.entries.push_back(de); + } + + for object in output.contents { + let mut path = build_rel_path(&self.core.root, &object.key); + if path.is_empty() { + path = "/".to_string(); + } + + let mut meta = Metadata::new(EntryMode::from_path(&path)); + meta.set_is_current(true); + if let Some(etag) = &object.etag { + meta.set_etag(etag); + meta.set_content_md5(etag.trim_matches('"')); + } + meta.set_content_length(object.size); + + // object.last_modified provides more precise time that contains + // nanosecond, let's trim them. + meta.set_last_modified(parse_datetime_from_rfc3339(object.last_modified.as_str())?); + + let de = oio::Entry::with(path, meta); + ctx.entries.push_back(de); + } + + Ok(()) + } +} /// S3ListerV2 implements ListObjectV2 for s3 backend. pub struct S3ListerV2 {
