This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 052e50502 feat(layers/hotpath): add HotpathLayer integration (#7083)
052e50502 is described below
commit 052e50502d23bf5c370713eb991d48aa69409b4a
Author: Chojan Shang <[email protected]>
AuthorDate: Mon Dec 22 19:41:27 2025 +0800
feat(layers/hotpath): add HotpathLayer integration (#7083)
* feat(layers/hotpath): add HotpathLayer integration
Signed-off-by: Chojan Shang <[email protected]>
* feat(layers/hotpath): instrument HttpClient via HotpathHttpFetcher
Signed-off-by: Chojan Shang <[email protected]>
* refactor(layers/hotpath): resolve comments
Signed-off-by: Chojan Shang <[email protected]>
---------
Signed-off-by: Chojan Shang <[email protected]>
---
core/Cargo.lock | 27 +++++
core/Cargo.toml | 2 +
core/layers/hotpath/Cargo.toml | 41 +++++++
core/layers/hotpath/src/lib.rs | 238 +++++++++++++++++++++++++++++++++++++++++
core/src/lib.rs | 2 +
5 files changed, 310 insertions(+)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 74a986bac..9f9e82226 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -3851,6 +3851,21 @@ dependencies = [
"windows-sys 0.59.0",
]
+[[package]]
+name = "hotpath"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "daa7e73e5b9dd467b2cbe7cc50ba224cd43bfcc0a618fd75f4adc32cf3a95d66"
+dependencies = [
+ "hotpath-macros",
+]
+
+[[package]]
+name = "hotpath-macros"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "4aca1659ca1e70edc8f96de48530a104bec11a267f663625098137bfd072d601"
+
[[package]]
name = "html5ever"
version = "0.35.0"
@@ -5542,6 +5557,7 @@ dependencies = [
"opendal-layer-dtrace",
"opendal-layer-fastmetrics",
"opendal-layer-fastrace",
+ "opendal-layer-hotpath",
"opendal-layer-immutable-index",
"opendal-layer-logging",
"opendal-layer-metrics",
@@ -5804,6 +5820,17 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "opendal-layer-hotpath"
+version = "0.55.0"
+dependencies = [
+ "futures",
+ "hotpath",
+ "http 1.4.0",
+ "opendal-core",
+ "tokio",
+]
+
[[package]]
name = "opendal-layer-immutable-index"
version = "0.55.0"
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 6906c60c1..3a301d73d 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -99,6 +99,7 @@ layers-concurrent-limit =
["dep:opendal-layer-concurrent-limit"]
layers-dtrace = ["dep:opendal-layer-dtrace"]
layers-fastmetrics = ["dep:opendal-layer-fastmetrics"]
layers-fastrace = ["dep:opendal-layer-fastrace"]
+layers-hotpath = ["dep:opendal-layer-hotpath"]
layers-immutable-index = ["dep:opendal-layer-immutable-index"]
layers-logging = ["dep:opendal-layer-logging"]
layers-metrics = ["dep:opendal-layer-metrics"]
@@ -206,6 +207,7 @@ opendal-layer-concurrent-limit = { path =
"layers/concurrent-limit", version = "
opendal-layer-dtrace = { path = "layers/dtrace", version = "0.55.0", optional
= true, default-features = false }
opendal-layer-fastmetrics = { path = "layers/fastmetrics", version = "0.55.0",
optional = true, default-features = false }
opendal-layer-fastrace = { path = "layers/fastrace", version = "0.55.0",
optional = true, default-features = false }
+opendal-layer-hotpath = { path = "layers/hotpath", version = "0.55.0",
optional = true, default-features = false }
opendal-layer-immutable-index = { path = "layers/immutable-index", version =
"0.55.0", optional = true, default-features = false }
opendal-layer-logging = { path = "layers/logging", version = "0.55.0",
optional = true, default-features = false }
opendal-layer-metrics = { path = "layers/metrics", version = "0.55.0",
optional = true, default-features = false }
diff --git a/core/layers/hotpath/Cargo.toml b/core/layers/hotpath/Cargo.toml
new file mode 100644
index 000000000..3071a3725
--- /dev/null
+++ b/core/layers/hotpath/Cargo.toml
@@ -0,0 +1,41 @@
+# 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.
+
+[package]
+description = "Apache OpenDAL hotpath layer"
+name = "opendal-layer-hotpath"
+
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+version = { workspace = true }
+
+[package.metadata.docs.rs]
+all-features = true
+
+[dependencies]
+futures = { workspace = true }
+hotpath = "0.9.1"
+http = { workspace = true }
+opendal-core = { path = "../../core", version = "0.55.0", default-features =
false }
+
+[dev-dependencies]
+opendal-core = { path = "../../core", version = "0.55.0" }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/core/layers/hotpath/src/lib.rs b/core/layers/hotpath/src/lib.rs
new file mode 100644
index 000000000..6e29b8e2b
--- /dev/null
+++ b/core/layers/hotpath/src/lib.rs
@@ -0,0 +1,238 @@
+// 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.
+
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+use futures::Stream;
+use futures::StreamExt;
+use hotpath::MeasurementGuard;
+use opendal_core::raw::*;
+use opendal_core::*;
+
+const LABEL_CREATE_DIR: &str = "opendal.create_dir";
+const LABEL_READ: &str = "opendal.read";
+const LABEL_WRITE: &str = "opendal.write";
+const LABEL_COPY: &str = "opendal.copy";
+const LABEL_RENAME: &str = "opendal.rename";
+const LABEL_STAT: &str = "opendal.stat";
+const LABEL_DELETE: &str = "opendal.delete";
+const LABEL_LIST: &str = "opendal.list";
+const LABEL_PRESIGN: &str = "opendal.presign";
+
+const LABEL_READER_READ: &str = "opendal.reader.read";
+const LABEL_WRITER_WRITE: &str = "opendal.writer.write";
+const LABEL_WRITER_CLOSE: &str = "opendal.writer.close";
+const LABEL_WRITER_ABORT: &str = "opendal.writer.abort";
+const LABEL_LISTER_NEXT: &str = "opendal.lister.next";
+const LABEL_DELETER_DELETE: &str = "opendal.deleter.delete";
+const LABEL_DELETER_CLOSE: &str = "opendal.deleter.close";
+const LABEL_HTTP_FETCH: &str = "opendal.http.fetch";
+const LABEL_HTTP_BODY_POLL: &str = "opendal.http.body.poll";
+
+/// Add [hotpath](https://docs.rs/hotpath/) profiling for every operation.
+///
+/// # Notes
+///
+/// When `hotpath` profiling is enabled, initialize a guard via
+/// [`hotpath::FunctionsGuardBuilder`] or `#[hotpath::main]` before running
+/// operations. Otherwise, hotpath will panic on the first measurement.
+///
+/// # Examples
+///
+/// ```no_run
+/// # use opendal_core::services;
+/// # use opendal_core::Operator;
+/// # use opendal_core::Result;
+/// # use opendal_layer_hotpath::HotpathLayer;
+/// #
+/// # #[tokio::main]
+/// # async fn main() -> Result<()> {
+/// let _guard = hotpath::FunctionsGuardBuilder::new("opendal").build();
+/// let op = Operator::new(services::Memory::default())?
+/// .layer(HotpathLayer)
+/// .finish();
+/// op.write("test", "hello").await?;
+/// # Ok(())
+/// # }
+/// ```
+pub struct HotpathLayer;
+
+impl<A: Access> Layer<A> for HotpathLayer {
+ type LayeredAccess = HotpathAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccess {
+ let info = inner.info();
+ info.update_http_client(|client| {
+ HttpClient::with(HotpathHttpFetcher {
+ inner: client.into_inner(),
+ })
+ });
+
+ HotpathAccessor { inner }
+ }
+}
+
+#[derive(Debug)]
+pub struct HotpathAccessor<A> {
+ inner: A,
+}
+
+impl<A: Access> LayeredAccess for HotpathAccessor<A> {
+ type Inner = A;
+ type Reader = HotpathWrapper<A::Reader>;
+ type Writer = HotpathWrapper<A::Writer>;
+ type Lister = HotpathWrapper<A::Lister>;
+ type Deleter = HotpathWrapper<A::Deleter>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ async fn create_dir(&self, path: &str, args: OpCreateDir) ->
Result<RpCreateDir> {
+ let _guard = MeasurementGuard::build(LABEL_CREATE_DIR, false, true);
+ self.inner.create_dir(path, args).await
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead,
Self::Reader)> {
+ let _guard = MeasurementGuard::build(LABEL_READ, false, true);
+ let (rp, reader) = self.inner.read(path, args).await?;
+ Ok((rp, HotpathWrapper::new(reader)))
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite,
Self::Writer)> {
+ let _guard = MeasurementGuard::build(LABEL_WRITE, false, true);
+ let (rp, writer) = self.inner.write(path, args).await?;
+ Ok((rp, HotpathWrapper::new(writer)))
+ }
+
+ async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy>
{
+ let _guard = MeasurementGuard::build(LABEL_COPY, false, true);
+ self.inner.copy(from, to, args).await
+ }
+
+ async fn rename(&self, from: &str, to: &str, args: OpRename) ->
Result<RpRename> {
+ let _guard = MeasurementGuard::build(LABEL_RENAME, false, true);
+ self.inner.rename(from, to, args).await
+ }
+
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let _guard = MeasurementGuard::build(LABEL_STAT, false, true);
+ self.inner.stat(path, args).await
+ }
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ let _guard = MeasurementGuard::build(LABEL_DELETE, false, true);
+ let (rp, deleter) = self.inner.delete().await?;
+ Ok((rp, HotpathWrapper::new(deleter)))
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList,
Self::Lister)> {
+ let _guard = MeasurementGuard::build(LABEL_LIST, false, true);
+ let (rp, lister) = self.inner.list(path, args).await?;
+ Ok((rp, HotpathWrapper::new(lister)))
+ }
+
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ let _guard = MeasurementGuard::build(LABEL_PRESIGN, false, true);
+ self.inner.presign(path, args).await
+ }
+}
+
+pub struct HotpathWrapper<R> {
+ inner: R,
+}
+
+impl<R> HotpathWrapper<R> {
+ fn new(inner: R) -> Self {
+ Self { inner }
+ }
+}
+
+impl<R: oio::Read> oio::Read for HotpathWrapper<R> {
+ async fn read(&mut self) -> Result<Buffer> {
+ let _guard = MeasurementGuard::build(LABEL_READER_READ, false, true);
+ self.inner.read().await
+ }
+}
+
+impl<R: oio::Write> oio::Write for HotpathWrapper<R> {
+ async fn write(&mut self, bs: Buffer) -> Result<()> {
+ let _guard = MeasurementGuard::build(LABEL_WRITER_WRITE, false, true);
+ self.inner.write(bs).await
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let _guard = MeasurementGuard::build(LABEL_WRITER_CLOSE, false, true);
+ self.inner.close().await
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ let _guard = MeasurementGuard::build(LABEL_WRITER_ABORT, false, true);
+ self.inner.abort().await
+ }
+}
+
+impl<R: oio::List> oio::List for HotpathWrapper<R> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ let _guard = MeasurementGuard::build(LABEL_LISTER_NEXT, false, true);
+ self.inner.next().await
+ }
+}
+
+impl<R: oio::Delete> oio::Delete for HotpathWrapper<R> {
+ async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let _guard = MeasurementGuard::build(LABEL_DELETER_DELETE, false,
true);
+ self.inner.delete(path, args).await
+ }
+
+ async fn close(&mut self) -> Result<()> {
+ let _guard = MeasurementGuard::build(LABEL_DELETER_CLOSE, false, true);
+ self.inner.close().await
+ }
+}
+
+pub struct HotpathHttpFetcher {
+ inner: HttpFetcher,
+}
+
+impl HttpFetch for HotpathHttpFetcher {
+ async fn fetch(&self, req: http::Request<Buffer>) ->
Result<http::Response<HttpBody>> {
+ let _guard = MeasurementGuard::build(LABEL_HTTP_FETCH, false, true);
+ let resp = self.inner.fetch(req).await?;
+ let (parts, body) = resp.into_parts();
+ let body = body.map_inner(|stream| Box::new(HotpathStream { inner:
stream }));
+ Ok(http::Response::from_parts(parts, body))
+ }
+}
+
+pub struct HotpathStream<S> {
+ inner: S,
+}
+
+impl<S> Stream for HotpathStream<S>
+where
+ S: Stream<Item = Result<Buffer>> + Unpin + 'static,
+{
+ type Item = Result<Buffer>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) ->
Poll<Option<Self::Item>> {
+ let _guard = MeasurementGuard::build(LABEL_HTTP_BODY_POLL, false,
true);
+ self.inner.poll_next_unpin(cx)
+ }
+}
diff --git a/core/src/lib.rs b/core/src/lib.rs
index c58e39325..502185981 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -169,6 +169,8 @@ pub mod layers {
pub use opendal_layer_fastmetrics::*;
#[cfg(feature = "layers-fastrace")]
pub use opendal_layer_fastrace::*;
+ #[cfg(feature = "layers-hotpath")]
+ pub use opendal_layer_hotpath::*;
#[cfg(feature = "layers-immutable-index")]
pub use opendal_layer_immutable_index::*;
#[cfg(feature = "layers-logging")]
