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/06 12:27:36 UTC
(opendal) branch cos-list-with-deleted created (now d71042db5)
This is an automated email from the ASF dual-hosted git repository.
hoslo pushed a change to branch cos-list-with-deleted
in repository https://gitbox.apache.org/repos/asf/opendal.git
at d71042db5 feat: cos list-with-deleted
This branch includes the following new commits:
new d71042db5 feat: cos list-with-deleted
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.
(opendal) 01/01: feat: cos list-with-deleted
Posted by ho...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hoslo pushed a commit to branch cos-list-with-deleted
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit d71042db5fd66dffe35884a40fb4758462fba77a
Author: yangshuai.hoslo <ya...@bytedance.com>
AuthorDate: Mon Jan 6 20:25:53 2025 +0800
feat: cos list-with-deleted
---
core/src/services/cos/backend.rs | 14 +++++
core/src/services/cos/config.rs | 2 +
core/src/services/cos/core.rs | 133 +++++++++++++++++++++++++++++++++++++--
core/src/services/cos/delete.rs | 4 +-
core/src/services/cos/lister.rs | 132 ++++++++++++++++++++++++++++++++++++++
5 files changed, 279 insertions(+), 6 deletions(-)
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
index 86d7e342c..f1ee663d6 100644
--- a/core/src/services/cos/backend.rs
+++ b/core/src/services/cos/backend.rs
@@ -123,6 +123,13 @@ impl CosBuilder {
self
}
+ /// Set bucket versioning status for this backend
+ pub fn enable_versioning(mut self, enabled: bool) -> Self {
+ self.config.enable_versioning = enabled;
+
+ self
+ }
+
/// Disable config load so that opendal will not load config from
/// environment.
///
@@ -215,6 +222,7 @@ impl Builder for CosBuilder {
bucket: bucket.clone(),
root,
endpoint: format!("{}://{}.{}", &scheme, &bucket, &endpoint),
+ enable_versioning: self.config.enable_versioning,
signer,
loader: cred_loader,
client,
@@ -253,15 +261,18 @@ impl Access for CosBackend {
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_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,
@@ -286,10 +297,13 @@ impl Access for CosBackend {
},
delete: true,
+ delete_with_version: self.core.enable_versioning,
copy: true,
list: true,
list_with_recursive: true,
+ list_with_versions: self.core.enable_versioning,
+ list_with_deleted: self.core.enable_versioning,
list_has_content_length: true,
presign: true,
diff --git a/core/src/services/cos/config.rs b/core/src/services/cos/config.rs
index 8c73f00ca..87436a84b 100644
--- a/core/src/services/cos/config.rs
+++ b/core/src/services/cos/config.rs
@@ -35,6 +35,8 @@ pub struct CosConfig {
pub secret_key: Option<String>,
/// Bucket of this backend.
pub bucket: Option<String>,
+ /// is bucket versioning enabled for this bucket
+ pub enable_versioning: bool,
/// Disable config load so that opendal will not load config from
pub disable_config_load: bool,
}
diff --git a/core/src/services/cos/core.rs b/core/src/services/cos/core.rs
index 4d28b2688..020b57bab 100644
--- a/core/src/services/cos/core.rs
+++ b/core/src/services/cos/core.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::fmt::Write;
use std::time::Duration;
use bytes::Bytes;
@@ -37,10 +38,15 @@ use serde::Serialize;
use crate::raw::*;
use crate::*;
+pub mod constants {
+ pub const COS_QUERY_VERSION_ID: &str = "versionId";
+}
+
pub struct CosCore {
pub bucket: String,
pub root: String,
pub endpoint: String,
+ pub enable_versioning: bool,
pub signer: TencentCosSigner,
pub loader: TencentCosCredentialLoader,
@@ -125,7 +131,19 @@ impl CosCore {
) -> Result<Request<Buffer>> {
let p = build_abs_path(&self.root, path);
- let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+ let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+ let mut query_args = Vec::new();
+ if let Some(version) = args.version() {
+ query_args.push(format!(
+ "{}={}",
+ constants::COS_QUERY_VERSION_ID,
+ percent_decode_path(version)
+ ))
+ }
+ if !query_args.is_empty() {
+ url.push_str(&format!("?{}", query_args.join("&")));
+ }
let mut req = Request::get(&url);
@@ -200,7 +218,19 @@ impl CosCore {
pub fn cos_head_object_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
let p = build_abs_path(&self.root, path);
- let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+ let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+ let mut query_args = Vec::new();
+ if let Some(version) = args.version() {
+ query_args.push(format!(
+ "{}={}",
+ constants::COS_QUERY_VERSION_ID,
+ percent_decode_path(version)
+ ))
+ }
+ if !query_args.is_empty() {
+ url.push_str(&format!("?{}", query_args.join("&")));
+ }
let mut req = Request::head(&url);
@@ -217,10 +247,22 @@ impl CosCore {
Ok(req)
}
- pub async fn cos_delete_object(&self, path: &str) -> Result<Response<Buffer>> {
+ pub async fn cos_delete_object(&self, path: &str, args: &OpDelete) -> Result<Response<Buffer>> {
let p = build_abs_path(&self.root, path);
- let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+ let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+ let mut query_args = Vec::new();
+ if let Some(version) = args.version() {
+ query_args.push(format!(
+ "{}={}",
+ constants::COS_QUERY_VERSION_ID,
+ percent_decode_path(version)
+ ))
+ }
+ if !query_args.is_empty() {
+ url.push_str(&format!("?{}", query_args.join("&")));
+ }
let req = Request::delete(&url);
@@ -434,6 +476,50 @@ impl CosCore {
self.sign(&mut req).await?;
self.send(req).await
}
+
+ pub async fn cos_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
+ }
}
/// Result of CreateMultipartUpload
@@ -511,6 +597,45 @@ pub struct ListObjectsOutputContent {
pub size: u64,
}
+#[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/cos/delete.rs b/core/src/services/cos/delete.rs
index 17319ba91..bd29f8521 100644
--- a/core/src/services/cos/delete.rs
+++ b/core/src/services/cos/delete.rs
@@ -33,8 +33,8 @@ impl CosDeleter {
}
impl oio::OneShotDelete for CosDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let resp = self.core.cos_delete_object(&path).await?;
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ let resp = self.core.cos_delete_object(&path, &args).await?;
let status = resp.status();
diff --git a/core/src/services/cos/lister.rs b/core/src/services/cos/lister.rs
index 880544425..5edc0e77f 100644
--- a/core/src/services/cos/lister.rs
+++ b/core/src/services/cos/lister.rs
@@ -23,9 +23,11 @@ use quick_xml::de;
use super::core::*;
use super::error::parse_error;
use crate::raw::*;
+use crate::raw::oio::PageContext;
use crate::EntryMode;
use crate::Metadata;
use crate::Result;
+use crate::Error;
pub struct CosLister {
core: Arc<CosCore>,
@@ -95,3 +97,133 @@ impl oio::PageList for CosLister {
Ok(())
}
}
+
+pub struct CosObjectVersionsLister {
+ core: Arc<CosCore>,
+
+ prefix: String,
+ args: OpList,
+
+ delimiter: &'static str,
+ abs_start_after: Option<String>,
+}
+
+impl CosObjectVersionsLister {
+ pub fn new(core: Arc<CosCore>, 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 CosObjectVersionsLister {
+ 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
+ .cos_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);
+ }
+
+ if self.args.versions() {
+ for version_object in output.version {
+ 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(())
+ }
+}
+