You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/03/27 03:55:31 UTC
(opendal) branch refactor-operations created (now 05d4e6db9)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch refactor-operations
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 05d4e6db9 refactor(core)!: Merge operations to build more clean metrics
This branch includes the following new commits:
new 05d4e6db9 refactor(core)!: Merge operations to build more clean metrics
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: refactor(core)!: Merge operations to build more clean metrics
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-operations
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 05d4e6db92dc7fb2e7e3ad0869ab3a6f7a8e6ca8
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 27 11:54:59 2025 +0800
refactor(core)!: Merge operations to build more clean metrics
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/await_tree.rs | 20 ++---
core/src/layers/capability_check.rs | 16 ++--
core/src/layers/correctness_check.rs | 31 +++----
core/src/layers/error_context.rs | 42 ++++-----
core/src/layers/fastrace.rs | 36 ++++----
core/src/layers/logging.rs | 163 ++++++++++++++--------------------
core/src/layers/observe/metrics.rs | 20 ++---
core/src/layers/timeout.rs | 20 ++---
core/src/raw/operation.rs | 71 +++------------
core/src/services/ftp/backend.rs | 6 +-
core/src/services/ftp/delete.rs | 2 +-
core/src/services/huggingface/core.rs | 4 +-
core/src/services/lakefs/core.rs | 12 +--
core/src/services/s3/core.rs | 22 ++---
14 files changed, 194 insertions(+), 271 deletions(-)
diff --git a/core/src/layers/await_tree.rs b/core/src/layers/await_tree.rs
index d04866c3d..b6ffa3198 100644
--- a/core/src/layers/await_tree.rs
+++ b/core/src/layers/await_tree.rs
@@ -86,7 +86,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.inner
.read(path, args)
- .instrument_await(format!("opendal::{}", Operation::ReaderStart))
+ .instrument_await(format!("opendal::{}", Operation::Read))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -94,7 +94,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.inner
.write(path, args)
- .instrument_await(format!("opendal::{}", Operation::WriterStart))
+ .instrument_await(format!("opendal::{}", Operation::Write))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -123,7 +123,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner
.delete()
- .instrument_await(format!("opendal::{}", Operation::DeleterStart))
+ .instrument_await(format!("opendal::{}", Operation::Delete))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -131,7 +131,7 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner
.list(path, args)
- .instrument_await(format!("opendal::{}", Operation::ListerStart))
+ .instrument_await(format!("opendal::{}", Operation::List))
.await
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
@@ -182,7 +182,7 @@ impl<R: oio::Read> oio::Read for AwaitTreeWrapper<R> {
async fn read(&mut self) -> Result<Buffer> {
self.inner
.read()
- .instrument_await(format!("opendal::{}", Operation::ReaderRead))
+ .instrument_await(format!("opendal::{}", Operation::Read))
.await
}
}
@@ -197,19 +197,19 @@ impl<R: oio::Write> oio::Write for AwaitTreeWrapper<R> {
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner
.write(bs)
- .instrument_await(format!("opendal::{}", Operation::WriterWrite.into_static()))
+ .instrument_await(format!("opendal::{}", Operation::Write.into_static()))
}
fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner
.abort()
- .instrument_await(format!("opendal::{}", Operation::WriterAbort.into_static()))
+ .instrument_await(format!("opendal::{}", Operation::Write.into_static()))
}
fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
self.inner
.close()
- .instrument_await(format!("opendal::{}", Operation::WriterClose.into_static()))
+ .instrument_await(format!("opendal::{}", Operation::Write.into_static()))
}
}
@@ -227,7 +227,7 @@ impl<R: oio::List> oio::List for AwaitTreeWrapper<R> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
self.inner
.next()
- .instrument_await(format!("opendal::{}", Operation::ListerNext))
+ .instrument_await(format!("opendal::{}", Operation::List))
.await
}
}
@@ -246,7 +246,7 @@ impl<R: oio::Delete> oio::Delete for AwaitTreeWrapper<R> {
async fn flush(&mut self) -> Result<usize> {
self.inner
.flush()
- .instrument_await(format!("opendal::{}", Operation::DeleterFlush))
+ .instrument_await(format!("opendal::{}", Operation::Delete))
.await
}
}
diff --git a/core/src/layers/capability_check.rs b/core/src/layers/capability_check.rs
index 6d412c61d..a380eb844 100644
--- a/core/src/layers/capability_check.rs
+++ b/core/src/layers/capability_check.rs
@@ -102,21 +102,21 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.write_with_content_type && args.content_type().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"content_type",
));
}
if !capability.write_with_cache_control && args.cache_control().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"cache_control",
));
}
if !capability.write_with_content_disposition && args.content_disposition().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"content_disposition",
));
}
@@ -133,7 +133,7 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.list_with_versions && args.versions() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ListerStart,
+ Operation::List,
"version",
));
}
@@ -158,21 +158,21 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.write_with_content_type && args.content_type().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"content_type",
));
}
if !capability.write_with_cache_control && args.cache_control().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"cache_control",
));
}
if !capability.write_with_content_disposition && args.content_disposition().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"content_disposition",
));
}
@@ -193,7 +193,7 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
if !capability.list_with_versions && args.versions() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ListerStart,
+ Operation::List,
"version",
));
}
diff --git a/core/src/layers/correctness_check.rs b/core/src/layers/correctness_check.rs
index e15976358..93ca6e431 100644
--- a/core/src/layers/correctness_check.rs
+++ b/core/src/layers/correctness_check.rs
@@ -98,35 +98,35 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if !capability.read_with_version && args.version().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"version",
));
}
if !capability.read_with_if_match && args.if_match().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"if_match",
));
}
if !capability.read_with_if_none_match && args.if_none_match().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"if_none_match",
));
}
if !capability.read_with_if_modified_since && args.if_modified_since().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"if_modified_since",
));
}
if !capability.read_with_if_unmodified_since && args.if_unmodified_since().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"if_unmodified_since",
));
}
@@ -139,24 +139,21 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if args.append() && !capability.write_can_append {
return Err(new_unsupported_error(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
"append",
));
}
if args.if_not_exists() && !capability.write_with_if_not_exists {
return Err(new_unsupported_error(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
"if_not_exists",
));
}
if let Some(if_none_match) = args.if_none_match() {
if !capability.write_with_if_none_match {
- let mut err = new_unsupported_error(
- self.info.as_ref(),
- Operation::WriterStart,
- "if_none_match",
- );
+ let mut err =
+ new_unsupported_error(self.info.as_ref(), Operation::Write, "if_none_match");
if if_none_match == "*" && capability.write_with_if_not_exists {
err = err.with_context("hint", "use if_not_exists instead");
}
@@ -225,7 +222,7 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if !capability.read_with_version && args.version().is_some() {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::ReaderStart,
+ Operation::Read,
"version",
));
}
@@ -238,21 +235,21 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
if args.append() && !capability.write_can_append {
return Err(new_unsupported_error(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
"append",
));
}
if args.if_not_exists() && !capability.write_with_if_not_exists {
return Err(new_unsupported_error(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
"if_not_exists",
));
}
if args.if_none_match().is_some() && !capability.write_with_if_none_match {
return Err(new_unsupported_error(
self.info.as_ref(),
- Operation::WriterStart,
+ Operation::Write,
"if_none_match",
));
}
@@ -299,7 +296,7 @@ impl<T> CheckWrapper<T> {
if args.version().is_some() && !self.info.full_capability().delete_with_version {
return Err(new_unsupported_error(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
"version",
));
}
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 326409fd9..a820bd07a 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -99,7 +99,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::ReaderStart)
+ err.with_operation(Operation::Read)
.with_context("service", self.info.scheme())
.with_context("path", path)
.with_context("range", range.to_string())
@@ -117,7 +117,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::WriterStart)
+ err.with_operation(Operation::Write)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -160,7 +160,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::DeleterStart)
+ err.with_operation(Operation::Delete)
.with_context("service", self.info.scheme())
})
}
@@ -176,7 +176,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::ListerStart)
+ err.with_operation(Operation::List)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -210,7 +210,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::ReaderStart)
+ err.with_operation(Operation::Read)
.with_context("service", self.info.scheme())
.with_context("path", path)
.with_context("range", range.to_string())
@@ -227,7 +227,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::WriterStart)
+ err.with_operation(Operation::Write)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -269,7 +269,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::DeleterStart)
+ err.with_operation(Operation::Delete)
.with_context("service", self.info.scheme())
})
}
@@ -284,7 +284,7 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
)
})
.map_err(|err| {
- err.with_operation(Operation::ListerStart)
+ err.with_operation(Operation::List)
.with_context("service", self.info.scheme())
.with_context("path", path)
})
@@ -325,7 +325,7 @@ impl<T: oio::Read> oio::Read for ErrorContextWrapper<T> {
self.processed += bs.len() as u64;
})
.map_err(|err| {
- err.with_operation(Operation::ReaderRead)
+ err.with_operation(Operation::Read)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("range", self.range.to_string())
@@ -342,7 +342,7 @@ impl<T: oio::BlockingRead> oio::BlockingRead for ErrorContextWrapper<T> {
self.processed += bs.len() as u64;
})
.map_err(|err| {
- err.with_operation(Operation::ReaderRead)
+ err.with_operation(Operation::Read)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("range", self.range.to_string())
@@ -361,7 +361,7 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
self.processed += size as u64;
})
.map_err(|err| {
- err.with_operation(Operation::WriterWrite)
+ err.with_operation(Operation::Write)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("size", size.to_string())
@@ -371,7 +371,7 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
async fn close(&mut self) -> Result<Metadata> {
self.inner.close().await.map_err(|err| {
- err.with_operation(Operation::WriterClose)
+ err.with_operation(Operation::Write)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("written", self.processed.to_string())
@@ -380,7 +380,7 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
async fn abort(&mut self) -> Result<()> {
self.inner.abort().await.map_err(|err| {
- err.with_operation(Operation::WriterAbort)
+ err.with_operation(Operation::Write)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("processed", self.processed.to_string())
@@ -397,7 +397,7 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
self.processed += size as u64;
})
.map_err(|err| {
- err.with_operation(Operation::WriterWrite)
+ err.with_operation(Operation::Write)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("size", size.to_string())
@@ -407,7 +407,7 @@ impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
fn close(&mut self) -> Result<Metadata> {
self.inner.close().map_err(|err| {
- err.with_operation(Operation::WriterClose)
+ err.with_operation(Operation::Write)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("written", self.processed.to_string())
@@ -424,7 +424,7 @@ impl<T: oio::List> oio::List for ErrorContextWrapper<T> {
self.processed += bs.is_some() as u64;
})
.map_err(|err| {
- err.with_operation(Operation::ListerNext)
+ err.with_operation(Operation::List)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("listed", self.processed.to_string())
@@ -440,7 +440,7 @@ impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
self.processed += bs.is_some() as u64;
})
.map_err(|err| {
- err.with_operation(Operation::ListerNext)
+ err.with_operation(Operation::List)
.with_context("service", self.scheme)
.with_context("path", &self.path)
.with_context("listed", self.processed.to_string())
@@ -451,7 +451,7 @@ impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
self.inner.delete(path, args).map_err(|err| {
- err.with_operation(Operation::DeleterDelete)
+ err.with_operation(Operation::Delete)
.with_context("service", self.scheme)
.with_context("path", path)
.with_context("deleted", self.processed.to_string())
@@ -466,7 +466,7 @@ impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
self.processed += n as u64;
})
.map_err(|err| {
- err.with_operation(Operation::DeleterFlush)
+ err.with_operation(Operation::Delete)
.with_context("service", self.scheme)
.with_context("deleted", self.processed.to_string())
})
@@ -476,7 +476,7 @@ impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
impl<T: oio::BlockingDelete> oio::BlockingDelete for ErrorContextWrapper<T> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
self.inner.delete(path, args).map_err(|err| {
- err.with_operation(Operation::DeleterDelete)
+ err.with_operation(Operation::Delete)
.with_context("service", self.scheme)
.with_context("path", path)
.with_context("deleted", self.processed.to_string())
@@ -490,7 +490,7 @@ impl<T: oio::BlockingDelete> oio::BlockingDelete for ErrorContextWrapper<T> {
self.processed += n as u64;
})
.map_err(|err| {
- err.with_operation(Operation::DeleterFlush)
+ err.with_operation(Operation::Delete)
.with_context("service", self.scheme)
.with_context("deleted", self.processed.to_string())
})
diff --git a/core/src/layers/fastrace.rs b/core/src/layers/fastrace.rs
index fabef47db..f01f4d2bc 100644
--- a/core/src/layers/fastrace.rs
+++ b/core/src/layers/fastrace.rs
@@ -149,7 +149,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::ReaderStart.into_static()),
+ Span::enter_with_local_parent(Operation::Read.into_static()),
r,
),
)
@@ -162,7 +162,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::WriterStart.into_static()),
+ Span::enter_with_local_parent(Operation::Write.into_static()),
r,
),
)
@@ -190,7 +190,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::DeleterStart.into_static()),
+ Span::enter_with_local_parent(Operation::Delete.into_static()),
r,
),
)
@@ -203,7 +203,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::ListerStart.into_static()),
+ Span::enter_with_local_parent(Operation::List.into_static()),
s,
),
)
@@ -226,7 +226,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::ReaderStart.into_static()),
+ Span::enter_with_local_parent(Operation::Read.into_static()),
r,
),
)
@@ -239,7 +239,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::WriterStart.into_static()),
+ Span::enter_with_local_parent(Operation::Write.into_static()),
r,
),
)
@@ -267,7 +267,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::DeleterStart.into_static()),
+ Span::enter_with_local_parent(Operation::Delete.into_static()),
r,
),
)
@@ -280,7 +280,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
(
rp,
FastraceWrapper::new(
- Span::enter_with_local_parent(Operation::ListerStart.into_static()),
+ Span::enter_with_local_parent(Operation::List.into_static()),
it,
),
)
@@ -309,7 +309,7 @@ impl<R: oio::Read> oio::Read for FastraceWrapper<R> {
impl<R: oio::BlockingRead> oio::BlockingRead for FastraceWrapper<R> {
fn read(&mut self) -> Result<Buffer> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::ReaderRead.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Read.into_static());
self.inner.read()
}
}
@@ -317,19 +317,19 @@ impl<R: oio::BlockingRead> oio::BlockingRead for FastraceWrapper<R> {
impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
fn write(&mut self, bs: Buffer) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::WriterWrite.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
self.inner.write(bs)
}
fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::WriterAbort.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
self.inner.abort()
}
fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::WriterClose.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
self.inner.close()
}
}
@@ -337,13 +337,13 @@ impl<R: oio::Write> oio::Write for FastraceWrapper<R> {
impl<R: oio::BlockingWrite> oio::BlockingWrite for FastraceWrapper<R> {
fn write(&mut self, bs: Buffer) -> Result<()> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::WriterWrite.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
self.inner.write(bs)
}
fn close(&mut self) -> Result<Metadata> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::WriterClose.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Write.into_static());
self.inner.close()
}
}
@@ -358,7 +358,7 @@ impl<R: oio::List> oio::List for FastraceWrapper<R> {
impl<R: oio::BlockingList> oio::BlockingList for FastraceWrapper<R> {
fn next(&mut self) -> Result<Option<oio::Entry>> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::ListerNext.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::List.into_static());
self.inner.next()
}
}
@@ -366,7 +366,7 @@ impl<R: oio::BlockingList> oio::BlockingList for FastraceWrapper<R> {
impl<R: oio::Delete> oio::Delete for FastraceWrapper<R> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::DeleterDelete.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Delete.into_static());
self.inner.delete(path, args)
}
@@ -379,13 +379,13 @@ impl<R: oio::Delete> oio::Delete for FastraceWrapper<R> {
impl<R: oio::BlockingDelete> oio::BlockingDelete for FastraceWrapper<R> {
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::DeleterDelete.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Delete.into_static());
self.inner.delete(path, args)
}
fn flush(&mut self) -> Result<usize> {
let _g = self.span.set_local_parent();
- let _span = LocalSpan::enter_with_local_parent(Operation::DeleterFlush.into_static());
+ let _span = LocalSpan::enter_with_local_parent(Operation::Delete.into_static());
self.inner.flush()
}
}
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index d1b2c49fb..c679b2e01 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -210,16 +210,9 @@ impl LoggingInterceptor for DefaultLoggingInterceptor {
);
}
- // Print debug message if operation is oneshot, otherwise in trace.
- let lvl = if operation.is_oneshot() {
- Level::Debug
- } else {
- Level::Trace
- };
-
log!(
target: LOGGING_TARGET,
- lvl,
+ Level::Debug,
"service={} name={}{}: {operation} {message}",
info.scheme(),
info.name(),
@@ -265,15 +258,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
}
fn info(&self) -> Arc<AccessorInfo> {
- self.logger
- .log(&self.info, Operation::Info, &[], "started", None);
-
- let info = self.info.clone();
-
- self.logger
- .log(&self.info, Operation::Info, &[], "finished", None);
-
- info
+ self.info.clone()
}
async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
@@ -311,7 +296,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"started",
None,
@@ -323,7 +308,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, r)| {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"created reader",
None,
@@ -336,7 +321,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"failed",
Some(err),
@@ -347,7 +332,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"started",
None,
@@ -359,7 +344,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, w)| {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"created writer",
None,
@@ -370,7 +355,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"failed",
Some(err),
@@ -476,32 +461,27 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.logger
- .log(&self.info, Operation::DeleterStart, &[], "started", None);
+ .log(&self.info, Operation::Delete, &[], "started", None);
self.inner
.delete()
.await
.map(|(rp, d)| {
self.logger
- .log(&self.info, Operation::DeleterStart, &[], "finished", None);
+ .log(&self.info, Operation::Delete, &[], "finished", None);
let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
(rp, d)
})
.inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::DeleterStart,
- &[],
- "failed",
- Some(err),
- );
+ self.logger
+ .log(&self.info, Operation::Delete, &[], "failed", Some(err));
})
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"started",
None,
@@ -513,7 +493,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, v)| {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"created lister",
None,
@@ -524,7 +504,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"failed",
Some(err),
@@ -598,7 +578,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"started",
None,
@@ -609,7 +589,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, r)| {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"created reader",
None,
@@ -620,7 +600,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::ReaderStart,
+ Operation::Read,
&[("path", path)],
"failed",
Some(err),
@@ -631,7 +611,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"started",
None,
@@ -642,7 +622,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, w)| {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"created writer",
None,
@@ -653,7 +633,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::WriterStart,
+ Operation::Write,
&[("path", path)],
"failed",
Some(err),
@@ -756,31 +736,26 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
self.logger
- .log(&self.info, Operation::DeleterStart, &[], "started", None);
+ .log(&self.info, Operation::Delete, &[], "started", None);
self.inner
.blocking_delete()
.map(|(rp, d)| {
self.logger
- .log(&self.info, Operation::DeleterStart, &[], "finished", None);
+ .log(&self.info, Operation::Delete, &[], "finished", None);
let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
(rp, d)
})
.inspect_err(|err| {
- self.logger.log(
- &self.info,
- Operation::DeleterStart,
- &[],
- "failed",
- Some(err),
- );
+ self.logger
+ .log(&self.info, Operation::Delete, &[], "failed", Some(err));
})
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"started",
None,
@@ -791,7 +766,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.map(|(rp, v)| {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"created lister",
None,
@@ -802,7 +777,7 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
.inspect_err(|err| {
self.logger.log(
&self.info,
- Operation::ListerStart,
+ Operation::List,
&[("path", path)],
"",
Some(err),
@@ -838,7 +813,7 @@ impl<R: oio::Read, I: LoggingInterceptor> oio::Read for LoggingReader<R, I> {
async fn read(&mut self) -> Result<Buffer> {
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[("path", &self.path), ("read", &self.read.to_string())],
"started",
None,
@@ -849,7 +824,7 @@ impl<R: oio::Read, I: LoggingInterceptor> oio::Read for LoggingReader<R, I> {
self.read += bs.len() as u64;
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[
("path", &self.path),
("read", &self.read.to_string()),
@@ -867,7 +842,7 @@ impl<R: oio::Read, I: LoggingInterceptor> oio::Read for LoggingReader<R, I> {
Err(err) => {
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[("path", &self.path), ("read", &self.read.to_string())],
"failed",
Some(&err),
@@ -882,7 +857,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
fn read(&mut self) -> Result<Buffer> {
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[("path", &self.path), ("read", &self.read.to_string())],
"started",
None,
@@ -893,7 +868,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
self.read += bs.len() as u64;
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[
("path", &self.path),
("read", &self.read.to_string()),
@@ -911,7 +886,7 @@ impl<R: oio::BlockingRead, I: LoggingInterceptor> oio::BlockingRead for LoggingR
Err(err) => {
self.logger.log(
&self.info,
- Operation::ReaderRead,
+ Operation::Read,
&[("path", &self.path), ("read", &self.read.to_string())],
"failed",
Some(&err),
@@ -950,7 +925,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -965,7 +940,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
self.written += size as u64;
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -979,7 +954,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
Err(err) => {
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -996,7 +971,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
async fn abort(&mut self) -> Result<()> {
self.logger.log(
&self.info,
- Operation::WriterAbort,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"started",
None,
@@ -1006,7 +981,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
Ok(_) => {
self.logger.log(
&self.info,
- Operation::WriterAbort,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"succeeded",
None,
@@ -1016,7 +991,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
Err(err) => {
self.logger.log(
&self.info,
- Operation::WriterAbort,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"failed",
Some(&err),
@@ -1029,7 +1004,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
async fn close(&mut self) -> Result<Metadata> {
self.logger.log(
&self.info,
- Operation::WriterClose,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"started",
None,
@@ -1039,7 +1014,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
Ok(meta) => {
self.logger.log(
&self.info,
- Operation::WriterClose,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"succeeded",
None,
@@ -1049,7 +1024,7 @@ impl<W: oio::Write, I: LoggingInterceptor> oio::Write for LoggingWriter<W, I> {
Err(err) => {
self.logger.log(
&self.info,
- Operation::WriterClose,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"failed",
Some(&err),
@@ -1066,7 +1041,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1080,7 +1055,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Ok(_) => {
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1094,7 +1069,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Err(err) => {
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[
("path", &self.path),
("written", &self.written.to_string()),
@@ -1111,7 +1086,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
fn close(&mut self) -> Result<Metadata> {
self.logger.log(
&self.info,
- Operation::WriterClose,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"started",
None,
@@ -1121,7 +1096,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Ok(meta) => {
self.logger.log(
&self.info,
- Operation::WriterWrite,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"succeeded",
None,
@@ -1131,7 +1106,7 @@ impl<W: oio::BlockingWrite, I: LoggingInterceptor> oio::BlockingWrite for Loggin
Err(err) => {
self.logger.log(
&self.info,
- Operation::WriterClose,
+ Operation::Write,
&[("path", &self.path), ("written", &self.written.to_string())],
"failed",
Some(&err),
@@ -1168,7 +1143,7 @@ impl<P: oio::List, I: LoggingInterceptor> oio::List for LoggingLister<P, I> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"started",
None,
@@ -1181,7 +1156,7 @@ impl<P: oio::List, I: LoggingInterceptor> oio::List for LoggingLister<P, I> {
self.listed += 1;
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[
("path", &self.path),
("listed", &self.listed.to_string()),
@@ -1194,7 +1169,7 @@ impl<P: oio::List, I: LoggingInterceptor> oio::List for LoggingLister<P, I> {
Ok(None) => {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"finished",
None,
@@ -1203,7 +1178,7 @@ impl<P: oio::List, I: LoggingInterceptor> oio::List for LoggingLister<P, I> {
Err(err) => {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"failed",
Some(err),
@@ -1219,7 +1194,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
fn next(&mut self) -> Result<Option<oio::Entry>> {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"started",
None,
@@ -1231,7 +1206,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
self.listed += 1;
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[
("path", &self.path),
("listed", &self.listed.to_string()),
@@ -1244,7 +1219,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
Ok(None) => {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"finished",
None,
@@ -1253,7 +1228,7 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
Err(err) => {
self.logger.log(
&self.info,
- Operation::ListerNext,
+ Operation::List,
&[("path", &self.path), ("listed", &self.listed.to_string())],
"failed",
Some(err),
@@ -1296,7 +1271,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[("path", path), ("version", &version)],
"started",
None,
@@ -1309,7 +1284,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
self.queued += 1;
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[
("path", path),
("version", &version),
@@ -1323,7 +1298,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
Err(err) => {
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[
("path", path),
("version", &version),
@@ -1342,7 +1317,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
async fn flush(&mut self) -> Result<usize> {
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1359,7 +1334,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
self.deleted += flushed;
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1371,7 +1346,7 @@ impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I>
Err(err) => {
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1395,7 +1370,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[("path", path), ("version", &version)],
"started",
None,
@@ -1408,7 +1383,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.queued += 1;
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[
("path", path),
("version", &version),
@@ -1422,7 +1397,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
Err(err) => {
self.logger.log(
&self.info,
- Operation::DeleterDelete,
+ Operation::Delete,
&[
("path", path),
("version", &version),
@@ -1441,7 +1416,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
fn flush(&mut self) -> Result<usize> {
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1458,7 +1433,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
self.deleted += flushed;
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
@@ -1470,7 +1445,7 @@ impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for Logg
Err(err) => {
self.logger.log(
&self.info,
- Operation::DeleterFlush,
+ Operation::Delete,
&[
("queued", &self.queued.to_string()),
("deleted", &self.deleted.to_string()),
diff --git a/core/src/layers/observe/metrics.rs b/core/src/layers/observe/metrics.rs
index d32b4d609..e8d165713 100644
--- a/core/src/layers/observe/metrics.rs
+++ b/core/src/layers/observe/metrics.rs
@@ -668,7 +668,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::ReaderRead.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Read.into_static());
let start = Instant::now();
@@ -699,7 +699,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::WriterStart.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Write.into_static());
let start = Instant::now();
@@ -810,7 +810,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::DeleterStart.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Delete.into_static());
let start = Instant::now();
@@ -841,7 +841,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::ListerNext.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::List.into_static());
let start = Instant::now();
@@ -929,7 +929,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::ReaderStart.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Read.into_static());
let start = Instant::now();
@@ -959,7 +959,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::WriterWrite.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Write.into_static());
let start = Instant::now();
@@ -1063,7 +1063,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::DeleterStart.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::Delete.into_static());
let start = Instant::now();
@@ -1093,7 +1093,7 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
- let labels = MetricLabels::new(self.info.clone(), Operation::ListerNext.into_static());
+ let labels = MetricLabels::new(self.info.clone(), Operation::List.into_static());
let start = Instant::now();
@@ -1137,8 +1137,8 @@ impl<R, I: MetricsIntercept> Drop for MetricsWrapper<R, I> {
let size = self.size;
let duration = self.start.elapsed();
- if self.labels.operation == Operation::ReaderRead.into_static()
- || self.labels.operation == Operation::WriterWrite.into_static()
+ if self.labels.operation == Operation::Read.into_static()
+ || self.labels.operation == Operation::Write.into_static()
{
self.interceptor
.observe(self.labels.clone(), MetricValue::OperationBytes(self.size));
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index e322671db..b89e13912 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -238,13 +238,13 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- self.io_timeout(Operation::ReaderStart, self.inner.read(path, args))
+ self.io_timeout(Operation::Read, self.inner.read(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- self.io_timeout(Operation::WriterStart, self.inner.write(path, args))
+ self.io_timeout(Operation::Write, self.inner.write(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
@@ -265,13 +265,13 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
}
async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
- self.timeout(Operation::DeleterStart, self.inner.delete())
+ self.timeout(Operation::Delete, self.inner.delete())
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- self.io_timeout(Operation::ListerStart, self.inner.list(path, args))
+ self.io_timeout(Operation::List, self.inner.list(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
@@ -348,31 +348,31 @@ impl<R> TimeoutWrapper<R> {
impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
async fn read(&mut self) -> Result<Buffer> {
let fut = self.inner.read();
- Self::io_timeout(self.timeout, Operation::ReaderRead.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::Read.into_static(), fut).await
}
}
impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
async fn write(&mut self, bs: Buffer) -> Result<()> {
let fut = self.inner.write(bs);
- Self::io_timeout(self.timeout, Operation::WriterWrite.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
}
async fn close(&mut self) -> Result<Metadata> {
let fut = self.inner.close();
- Self::io_timeout(self.timeout, Operation::WriterClose.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
}
async fn abort(&mut self) -> Result<()> {
let fut = self.inner.abort();
- Self::io_timeout(self.timeout, Operation::WriterAbort.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
}
}
impl<R: oio::List> oio::List for TimeoutWrapper<R> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
let fut = self.inner.next();
- Self::io_timeout(self.timeout, Operation::ListerNext.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::List.into_static(), fut).await
}
}
@@ -383,7 +383,7 @@ impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> {
async fn flush(&mut self) -> Result<usize> {
let fut = self.inner.flush();
- Self::io_timeout(self.timeout, Operation::DeleterFlush.into_static(), fut).await
+ Self::io_timeout(self.timeout, Operation::Delete.into_static(), fut).await
}
}
diff --git a/core/src/raw/operation.rs b/core/src/raw/operation.rs
index 633b823e4..ae9e3b47c 100644
--- a/core/src/raw/operation.rs
+++ b/core/src/raw/operation.rs
@@ -33,47 +33,20 @@ pub enum Operation {
Info,
/// Operation to create a directory.
CreateDir,
- /// Operation to start reading from a file.
- ///
- /// This operation only start to read from the file, it does not read any data.
- ///
- /// For example:
- ///
- /// - On fs alike storage, this operation is used to open a file.
- /// - On s3 alike storage, this operation is used to send a request but not consume it's body.
- ReaderStart,
- /// Operation to perform a single read I/O operation on a file.
- ///
- /// Reader might have to call this operation multiple times to read the whole file.
- ReaderRead,
- /// Operation to start writing to a file.
- ///
- /// This operation only start to write into the file, it does not write any data.
- WriterStart,
- /// Operation to perform a single write I/O operation on a file.
- ///
- /// Writer might have to call this operation multiple times to write the whole file.
- WriterWrite,
- /// Operation to close the writer and finish writing.
- WriterClose,
- /// Operation to abort the writer and discard the data.
- WriterAbort,
+ /// Operation to read a file.
+ Read,
+ /// Operation to write to a file.
+ Write,
/// Operation to copy a file.
Copy,
/// Operation to rename a file.
Rename,
/// Operation to stat a file or a directory.
Stat,
- /// Operation to start deleting files.
- DeleterStart,
- /// Operation to push a delete operation into the queue.
- DeleterDelete,
- /// Operation to flush the deleter queue and perform real delete operations.
- DeleterFlush,
- /// Operation to start listing files.
- ListerStart,
+ /// Operation to delete files.
+ Delete,
/// Operation to get the next file from the list.
- ListerNext,
+ List,
/// Operation to generate a presigned URL.
Presign,
}
@@ -83,21 +56,6 @@ impl Operation {
pub fn into_static(self) -> &'static str {
self.into()
}
-
- /// Check if given operation is oneshot or not.
- ///
- /// For example, `Stat` is oneshot but `ReaderRead` could happen multiple times.
- ///
- /// This function can be used to decide take actions based on operations like logging.
- pub fn is_oneshot(&self) -> bool {
- !matches!(
- self,
- Operation::ReaderRead
- | Operation::WriterWrite
- | Operation::ListerNext
- | Operation::DeleterDelete
- )
- }
}
impl Display for Operation {
@@ -111,20 +69,13 @@ impl From<Operation> for &'static str {
match v {
Operation::Info => "info",
Operation::CreateDir => "create_dir",
- Operation::ReaderStart => "reader::start",
- Operation::ReaderRead => "reader::read",
- Operation::WriterStart => "writer::start",
- Operation::WriterWrite => "writer::write",
- Operation::WriterClose => "writer::close",
- Operation::WriterAbort => "writer::abort",
+ Operation::Read => "read",
+ Operation::Write => "write",
Operation::Copy => "copy",
Operation::Rename => "rename",
Operation::Stat => "stat",
- Operation::DeleterStart => "deleter::start",
- Operation::DeleterDelete => "deleter::delete",
- Operation::DeleterFlush => "deleter::flush",
- Operation::ListerStart => "lister::start",
- Operation::ListerNext => "lister::next",
+ Operation::Delete => "delete",
+ Operation::List => "list",
Operation::Presign => "presign",
}
}
diff --git a/core/src/services/ftp/backend.rs b/core/src/services/ftp/backend.rs
index 5294824d4..1a4f83a1c 100644
--- a/core/src/services/ftp/backend.rs
+++ b/core/src/services/ftp/backend.rs
@@ -344,7 +344,7 @@ impl Access for FtpBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let ftp_stream = self.ftp_connect(Operation::ReaderStart).await?;
+ let ftp_stream = self.ftp_connect(Operation::Read).await?;
let reader = FtpReader::new(ftp_stream, path.to_string(), args).await?;
Ok((RpRead::new(), reader))
@@ -356,7 +356,7 @@ impl Access for FtpBackend {
let paths: Vec<&str> = parent.split('/').collect();
// TODO: we can optimize this by checking dir existence first.
- let mut ftp_stream = self.ftp_connect(Operation::WriterStart).await?;
+ let mut ftp_stream = self.ftp_connect(Operation::Write).await?;
let mut curr_path = String::new();
for path in paths {
@@ -398,7 +398,7 @@ impl Access for FtpBackend {
}
async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
- let mut ftp_stream = self.ftp_connect(Operation::ListerStart).await?;
+ let mut ftp_stream = self.ftp_connect(Operation::List).await?;
let pathname = if path == "/" { None } else { Some(path) };
let files = ftp_stream.list(pathname).await.map_err(parse_error)?;
diff --git a/core/src/services/ftp/delete.rs b/core/src/services/ftp/delete.rs
index bc9bd2942..12ebed770 100644
--- a/core/src/services/ftp/delete.rs
+++ b/core/src/services/ftp/delete.rs
@@ -36,7 +36,7 @@ impl FtpDeleter {
impl oio::OneShotDelete for FtpDeleter {
async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let mut ftp_stream = self.core.ftp_connect(Operation::DeleterStart).await?;
+ let mut ftp_stream = self.core.ftp_connect(Operation::Delete).await?;
let result = if path.ends_with('/') {
ftp_stream.rmdir(&path).await
diff --git a/core/src/services/huggingface/core.rs b/core/src/services/huggingface/core.rs
index 1e563f980..8190e0130 100644
--- a/core/src/services/huggingface/core.rs
+++ b/core/src/services/huggingface/core.rs
@@ -110,7 +110,7 @@ impl HuggingfaceCore {
let mut req = Request::get(&url);
// Inject operation to the request.
- req = req.extension(Operation::ListerStart);
+ req = req.extension(Operation::List);
if let Some(token) = &self.token {
let auth_header_content = format_authorization_by_bearer(token)?;
req = req.header(header::AUTHORIZATION, auth_header_content);
@@ -157,7 +157,7 @@ impl HuggingfaceCore {
req = req.header(header::RANGE, range.to_header());
}
// Inject operation to the request.
- let req = req.extension(Operation::ReaderStart);
+ let req = req.extension(Operation::Read);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.info.http_client().fetch(req).await
diff --git a/core/src/services/lakefs/core.rs b/core/src/services/lakefs/core.rs
index 52fe9df87..3e3e190b5 100644
--- a/core/src/services/lakefs/core.rs
+++ b/core/src/services/lakefs/core.rs
@@ -67,7 +67,7 @@ impl LakefsCore {
let auth_header_content = format_authorization_by_basic(&self.username, &self.password)?;
req = req.header(header::AUTHORIZATION, auth_header_content);
// Inject operation to the request.
- let req = req.extension(Operation::ReaderStart);
+ let req = req.extension(Operation::Read);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.info.http_client().send(req).await
@@ -100,7 +100,7 @@ impl LakefsCore {
req = req.header(header::RANGE, range.to_header());
}
// Inject operation to the request.
- let req = req.extension(Operation::ReaderStart);
+ let req = req.extension(Operation::Read);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.info.http_client().fetch(req).await
@@ -141,7 +141,7 @@ impl LakefsCore {
let auth_header_content = format_authorization_by_basic(&self.username, &self.password)?;
req = req.header(header::AUTHORIZATION, auth_header_content);
// Inject operation to the request.
- let req = req.extension(Operation::ReaderStart);
+ let req = req.extension(Operation::Read);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.info.http_client().send(req).await
@@ -170,7 +170,7 @@ impl LakefsCore {
let auth_header_content = format_authorization_by_basic(&self.username, &self.password)?;
req = req.header(header::AUTHORIZATION, auth_header_content);
// Inject operation to the request.
- let req = req.extension(Operation::WriterWrite);
+ let req = req.extension(Operation::Write);
let req = req.body(body).map_err(new_request_build_error)?;
self.info.http_client().send(req).await
@@ -194,7 +194,7 @@ impl LakefsCore {
let auth_header_content = format_authorization_by_basic(&self.username, &self.password)?;
req = req.header(header::AUTHORIZATION, auth_header_content);
// Inject operation to the request.
- let req = req.extension(Operation::DeleterFlush);
+ let req = req.extension(Operation::Delete);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.info.http_client().send(req).await
@@ -226,7 +226,7 @@ impl LakefsCore {
let req = req
// Inject operation to the request.
- .extension(Operation::DeleterFlush)
+ .extension(Operation::Delete)
.body(serde_json::to_vec(&map).unwrap().into())
.map_err(new_request_build_error)?;
self.info.http_client().send(req).await
diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs
index e3e4cb930..ba7d530f8 100644
--- a/core/src/services/s3/core.rs
+++ b/core/src/services/s3/core.rs
@@ -491,7 +491,7 @@ impl S3Core {
req = self.insert_sse_headers(req, false);
// Inject operation to the request.
- let req = req.extension(Operation::ReaderStart);
+ let req = req.extension(Operation::Read);
let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
@@ -536,7 +536,7 @@ impl S3Core {
}
// Inject operation to the request.
- let req = req.extension(Operation::WriterWrite);
+ let req = req.extension(Operation::Write);
// Set body
let req = req.body(body).map_err(new_request_build_error)?;
@@ -570,7 +570,7 @@ impl S3Core {
req = self.insert_sse_headers(req, true);
// Inject operation to the request.
- let req = req.extension(Operation::WriterWrite);
+ let req = req.extension(Operation::Write);
// Set body
let req = req.body(body).map_err(new_request_build_error)?;
@@ -607,7 +607,7 @@ impl S3Core {
let mut req = Request::delete(&url)
// Inject operation to the request.
- .extension(Operation::DeleterFlush)
+ .extension(Operation::Delete)
.body(Buffer::new())
.map_err(new_request_build_error)?;
@@ -716,7 +716,7 @@ impl S3Core {
let mut req = Request::get(&url)
// Inject operation to the request.
- .extension(Operation::ListerNext)
+ .extension(Operation::List)
.body(Buffer::new())
.map_err(new_request_build_error)?;
@@ -767,7 +767,7 @@ impl S3Core {
let req = self.insert_checksum_type_header(req);
// Inject operation to the request.
- let req = req.extension(Operation::WriterStart);
+ let req = req.extension(Operation::Write);
let mut req = req.body(Buffer::new()).map_err(new_request_build_error)?;
@@ -808,7 +808,7 @@ impl S3Core {
}
// Inject operation to the request.
- let req = req.extension(Operation::WriterWrite);
+ let req = req.extension(Operation::Write);
// Set body
let req = req.body(body).map_err(new_request_build_error)?;
@@ -844,7 +844,7 @@ impl S3Core {
let req = req.header(CONTENT_TYPE, "application/xml");
// Inject operation to the request.
- let req = req.extension(Operation::WriterClose);
+ let req = req.extension(Operation::Write);
let mut req = req
.body(Buffer::from(Bytes::from(content)))
@@ -872,7 +872,7 @@ impl S3Core {
let mut req = Request::delete(&url)
// Inject operation to the request.
- .extension(Operation::WriterAbort)
+ .extension(Operation::Write)
.body(Buffer::new())
.map_err(new_request_build_error)?;
self.sign(&mut req).await?;
@@ -906,7 +906,7 @@ impl S3Core {
let req = req.header("CONTENT-MD5", format_content_md5(content.as_bytes()));
// Inject operation to the request.
- let req = req.extension(Operation::DeleterFlush);
+ let req = req.extension(Operation::Delete);
let mut req = req
.body(Buffer::from(Bytes::from(content)))
@@ -954,7 +954,7 @@ impl S3Core {
let mut req = Request::get(&url)
// Inject operation to the request.
- .extension(Operation::ListerNext)
+ .extension(Operation::List)
.body(Buffer::new())
.map_err(new_request_build_error)?;