You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ho...@apache.org on 2025/01/09 09:50:26 UTC
(opendal) branch oss-list-with-deleted updated (2840db79e -> 8d142b0f8)
This is an automated email from the ASF dual-hosted git repository.
hoslo pushed a change to branch oss-list-with-deleted
in repository https://gitbox.apache.org/repos/asf/opendal.git
discard 2840db79e feat(core): Implement list with deleted and versions for oss
add 90c64a19b Bump to version 0.51.1 Round 1 (#5523)
new 8d142b0f8 feat(core): Implement list with deleted and versions for oss
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (2840db79e)
\
N -- N -- N refs/heads/oss-list-with-deleted (8d142b0f8)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
(opendal) 01/01: feat(core): Implement list with deleted and versions for oss
Posted by ho...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hoslo pushed a commit to branch oss-list-with-deleted
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 8d142b0f8f48678945d857d2097f6d64be025bd5
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Jan 8 21:28:12 2025 +0800
feat(core): Implement list with deleted and versions for oss
---
core/src/services/oss/backend.rs | 45 +++++++++----
core/src/services/oss/config.rs | 3 +
core/src/services/oss/core.rs | 84 +++++++++++++++++++++++
core/src/services/oss/lister.rs | 139 +++++++++++++++++++++++++++++++++++++++
4 files changed, 260 insertions(+), 11 deletions(-)
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index 0ec56cdc2..ad035cba2 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -30,7 +30,7 @@ use reqsign::AliyunOssSigner;
use super::core::*;
use super::delete::OssDeleter;
use super::error::parse_error;
-use super::lister::OssLister;
+use super::lister::{OssLister, OssListers, OssObjectVersionsLister};
use super::writer::OssWriter;
use super::writer::OssWriters;
use crate::raw::*;
@@ -97,6 +97,13 @@ impl OssBuilder {
self
}
+ /// Set bucket versioning status for this backend
+ pub fn enable_versioning(mut self, enabled: bool) -> Self {
+ self.config.enable_versioning = enabled;
+
+ self
+ }
+
/// Set an endpoint for generating presigned urls.
///
/// You can offer a public endpoint like <https://oss-cn-beijing.aliyuncs.com> to return a presinged url for
@@ -408,6 +415,7 @@ impl Builder for OssBuilder {
host,
presign_endpoint,
allow_anonymous: self.config.allow_anonymous,
+ enable_versioning: self.config.enable_versioning,
signer,
loader,
client,
@@ -428,7 +436,7 @@ pub struct OssBackend {
impl Access for OssBackend {
type Reader = HttpBody;
type Writer = OssWriters;
- type Lister = oio::PageLister<OssLister>;
+ type Lister = OssListers;
type Deleter = oio::BatchDeleter<OssDeleter>;
type BlockingReader = ();
type BlockingWriter = ();
@@ -449,16 +457,19 @@ impl Access for OssBackend {
stat_has_content_type: true,
stat_has_content_encoding: true,
stat_has_content_range: true,
+ stat_with_version: self.core.enable_versioning,
stat_has_etag: true,
stat_has_content_md5: true,
stat_has_last_modified: true,
stat_has_content_disposition: true,
stat_has_user_metadata: true,
+ stat_has_version: true,
read: true,
read_with_if_match: true,
read_with_if_none_match: true,
+ read_with_version: self.core.enable_versioning,
write: true,
write_can_empty: true,
@@ -468,7 +479,7 @@ impl Access for OssBackend {
write_with_content_type: true,
write_with_content_disposition: true,
// TODO: set this to false while version has been enabled.
- write_with_if_not_exists: true,
+ write_with_if_not_exists: !self.core.enable_versioning,
// The min multipart size of OSS is 100 KiB.
//
@@ -485,6 +496,7 @@ impl Access for OssBackend {
write_with_user_metadata: true,
delete: true,
+ delete_with_version: self.core.enable_versioning,
delete_max_size: Some(self.core.delete_max_size),
copy: true,
@@ -495,6 +507,8 @@ impl Access for OssBackend {
list_with_recursive: true,
list_has_etag: true,
list_has_content_md5: true,
+ list_with_versions: self.core.enable_versioning,
+ list_with_deleted: self.core.enable_versioning,
list_has_content_length: true,
list_has_last_modified: true,
@@ -572,14 +586,23 @@ impl Access for OssBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let l = OssLister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- );
- Ok((RpList::default(), oio::PageLister::new(l)))
+ let l = if args.versions() || args.deleted() {
+ TwoWays::Two(oio::PageLister::new(OssObjectVersionsLister::new(
+ self.core.clone(),
+ path,
+ args,
+ )))
+ } else {
+ TwoWays::One(oio::PageLister::new(OssLister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ )))
+ };
+
+ Ok((RpList::default(), l))
}
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
diff --git a/core/src/services/oss/config.rs b/core/src/services/oss/config.rs
index 73259f3e3..1e44888d2 100644
--- a/core/src/services/oss/config.rs
+++ b/core/src/services/oss/config.rs
@@ -36,6 +36,9 @@ pub struct OssConfig {
/// Bucket for oss.
pub bucket: String,
+ /// is bucket versioning enabled for this bucket
+ pub enable_versioning: bool,
+
// OSS features
/// Server side encryption for oss.
pub server_side_encryption: Option<String>,
diff --git a/core/src/services/oss/core.rs b/core/src/services/oss/core.rs
index fef9a8604..942bed2a5 100644
--- a/core/src/services/oss/core.rs
+++ b/core/src/services/oss/core.rs
@@ -68,6 +68,7 @@ pub struct OssCore {
pub endpoint: String,
pub presign_endpoint: String,
pub allow_anonymous: bool,
+ pub enable_versioning: bool,
pub server_side_encryption: Option<HeaderValue>,
pub server_side_encryption_key_id: Option<HeaderValue>,
@@ -504,6 +505,50 @@ impl OssCore {
self.send(req).await
}
+ pub async fn oss_list_object_versions(
+ &self,
+ prefix: &str,
+ delimiter: &str,
+ limit: Option<usize>,
+ key_marker: &str,
+ version_id_marker: &str,
+ ) -> Result<Response<Buffer>> {
+ let p = build_abs_path(&self.root, prefix);
+
+ let mut url = format!("{}?versions", self.endpoint);
+ if !p.is_empty() {
+ write!(url, "&prefix={}", percent_encode_path(p.as_str()))
+ .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 !key_marker.is_empty() {
+ write!(url, "&key-marker={}", percent_encode_path(key_marker))
+ .expect("write into string must succeed");
+ }
+ if !version_id_marker.is_empty() {
+ write!(
+ url,
+ "&version-id-marker={}",
+ percent_encode_path(version_id_marker)
+ )
+ .expect("write into string must succeed");
+ }
+
+ let mut req = Request::get(&url)
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+
+ self.sign(&mut req).await?;
+
+ self.send(req).await
+ }
+
pub async fn oss_delete_object(&self, path: &str, args: &OpDelete) -> Result<Response<Buffer>> {
let mut req = self.oss_delete_object_request(path, args)?;
self.sign(&mut req).await?;
@@ -752,6 +797,45 @@ pub struct CommonPrefix {
pub prefix: String,
}
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct OutputCommonPrefix {
+ pub prefix: String,
+}
+
+/// Output of ListObjectVersions
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutput {
+ pub is_truncated: Option<bool>,
+ pub next_key_marker: Option<String>,
+ pub next_version_id_marker: Option<String>,
+ pub common_prefixes: Vec<OutputCommonPrefix>,
+ pub version: Vec<ListObjectVersionsOutputVersion>,
+ pub delete_marker: Vec<ListObjectVersionsOutputDeleteMarker>,
+}
+
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputVersion {
+ pub key: String,
+ pub version_id: String,
+ pub is_latest: bool,
+ pub size: u64,
+ pub last_modified: String,
+ #[serde(rename = "ETag")]
+ pub etag: Option<String>,
+}
+
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputDeleteMarker {
+ pub key: String,
+ pub version_id: String,
+ pub is_latest: bool,
+ pub last_modified: String,
+}
+
#[cfg(test)]
mod tests {
use bytes::Buf;
diff --git a/core/src/services/oss/lister.rs b/core/src/services/oss/lister.rs
index e69096e50..286147a00 100644
--- a/core/src/services/oss/lister.rs
+++ b/core/src/services/oss/lister.rs
@@ -22,9 +22,12 @@ use quick_xml::de;
use super::core::*;
use super::error::parse_error;
+use crate::raw::oio::PageContext;
use crate::raw::*;
use crate::*;
+pub type OssListers = TwoWays<oio::PageLister<OssLister>, oio::PageLister<OssObjectVersionsLister>>;
+
pub struct OssLister {
core: Arc<OssCore>,
@@ -115,3 +118,139 @@ impl oio::PageList for OssLister {
Ok(())
}
}
+
+/// refer: https://help.aliyun.com/zh/oss/developer-reference/listobjectversions?spm=a2c4g.11186623.help-menu-31815.d_3_1_1_5_5_2.53f67237GJlMPw&scm=20140722.H_112467._.OR_help-T_cn~zh-V_1
+pub struct OssObjectVersionsLister {
+ core: Arc<OssCore>,
+
+ prefix: String,
+ args: OpList,
+
+ delimiter: &'static str,
+ abs_start_after: Option<String>,
+}
+
+impl OssObjectVersionsLister {
+ pub fn new(core: Arc<OssCore>, path: &str, args: OpList) -> Self {
+ let delimiter = if args.recursive() { "" } else { "/" };
+ let abs_start_after = args
+ .start_after()
+ .map(|start_after| build_abs_path(&core.root, start_after));
+
+ Self {
+ core,
+ prefix: path.to_string(),
+ args,
+ delimiter,
+ abs_start_after,
+ }
+ }
+}
+
+impl oio::PageList for OssObjectVersionsLister {
+ async fn next_page(&self, ctx: &mut PageContext) -> Result<()> {
+ let markers = ctx.token.rsplit_once(" ");
+ let (key_marker, version_id_marker) = if let Some(data) = markers {
+ data
+ } else if let Some(start_after) = &self.abs_start_after {
+ (start_after.as_str(), "")
+ } else {
+ ("", "")
+ };
+
+ let resp = self
+ .core
+ .oss_list_object_versions(
+ &self.prefix,
+ self.delimiter,
+ self.args.limit(),
+ key_marker,
+ version_id_marker,
+ )
+ .await?;
+ if resp.status() != http::StatusCode::OK {
+ return Err(parse_error(resp));
+ }
+
+ let body = resp.into_body();
+ let output: ListObjectVersionsOutput = de::from_reader(body.reader())
+ .map_err(new_xml_deserialize_error)
+ // Allow Cos list to retry on XML deserialization errors.
+ //
+ // This is because the Cos 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)?;
+
+ ctx.done = if let Some(is_truncated) = output.is_truncated {
+ !is_truncated
+ } else {
+ false
+ };
+ ctx.token = format!(
+ "{} {}",
+ output.next_key_marker.unwrap_or_default(),
+ output.next_version_id_marker.unwrap_or_default()
+ );
+
+ 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 version_object in output.version {
+ // `list` must be additive, so we need to include the latest version object
+ // even if `versions` is not enabled.
+ //
+ // Here we skip all non-latest version objects if `versions` is not enabled.
+ if !(self.args.versions() || version_object.is_latest) {
+ continue;
+ }
+
+ let mut path = build_rel_path(&self.core.root, &version_object.key);
+ if path.is_empty() {
+ path = "/".to_owned();
+ }
+
+ let mut meta = Metadata::new(EntryMode::from_path(&path));
+ meta.set_version(&version_object.version_id);
+ meta.set_is_current(version_object.is_latest);
+ meta.set_content_length(version_object.size);
+ meta.set_last_modified(parse_datetime_from_rfc3339(
+ version_object.last_modified.as_str(),
+ )?);
+ if let Some(etag) = version_object.etag {
+ meta.set_etag(&etag);
+ meta.set_content_md5(etag.trim_matches('"'));
+ }
+
+ let entry = oio::Entry::new(&path, meta);
+ ctx.entries.push_back(entry);
+ }
+
+ if self.args.deleted() {
+ for delete_marker in output.delete_marker {
+ let mut path = build_rel_path(&self.core.root, &delete_marker.key);
+ if path.is_empty() {
+ path = "/".to_owned();
+ }
+
+ let mut meta = Metadata::new(EntryMode::FILE);
+ meta.set_version(&delete_marker.version_id);
+ meta.set_is_deleted(true);
+ meta.set_is_current(delete_marker.is_latest);
+ meta.set_last_modified(parse_datetime_from_rfc3339(
+ delete_marker.last_modified.as_str(),
+ )?);
+
+ let entry = oio::Entry::new(&path, meta);
+ ctx.entries.push_back(entry);
+ }
+ }
+
+ Ok(())
+ }
+}