opendal_layer_oteltrace/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use std::sync::Arc;
23
24use opendal_core::raw::*;
25use opendal_core::*;
26use opentelemetry::Context as TraceContext;
27use opentelemetry::KeyValue;
28use opentelemetry::global;
29use opentelemetry::trace::FutureExt as TraceFutureExt;
30use opentelemetry::trace::Span;
31use opentelemetry::trace::TraceContextExt;
32use opentelemetry::trace::Tracer;
33
34/// `OtelTraceLayer` traces OpenDAL operations with
35/// [OpenTelemetry](https://docs.rs/opentelemetry/latest/opentelemetry/trace/index.html).
36///
37/// The layer obtains the `opendal` tracer from OpenTelemetry's global tracer
38/// provider. Applications must install and configure that provider before
39/// issuing operations.
40///
41/// The layer creates spans for service metadata, create, read, write, copy,
42/// rename, stat, list, and presign calls. It carries read, write, and list
43/// contexts into their stateful I/O bodies. Delete calls currently pass through
44/// without creating a span.
45///
46/// # Examples
47///
48/// ## Basic Setup
49///
50/// ```no_run
51/// # use opendal_core::services;
52/// # use opendal_core::Operator;
53/// # use opendal_core::Result;
54/// # use opendal_layer_oteltrace::OtelTraceLayer;
55/// #
56/// # fn main() -> Result<()> {
57/// let _ = Operator::new(services::Memory::default())?
58///     .layer(OtelTraceLayer::new());
59/// # Ok(())
60/// # }
61/// ```
62#[derive(Clone, Debug, Default)]
63#[non_exhaustive]
64pub struct OtelTraceLayer {}
65
66impl OtelTraceLayer {
67    /// Create a new [`OtelTraceLayer`].
68    pub fn new() -> Self {
69        Self::default()
70    }
71}
72
73impl Layer for OtelTraceLayer {
74    fn apply_service(&self, inner: Servicer) -> Servicer {
75        Arc::new(self.layer(inner))
76    }
77}
78
79impl OtelTraceLayer {
80    fn layer(&self, inner: Servicer) -> OtelTraceService {
81        OtelTraceService { inner }
82    }
83}
84
85#[doc(hidden)]
86#[derive(Debug)]
87pub struct OtelTraceService {
88    inner: Servicer,
89}
90
91impl Service for OtelTraceService {
92    type Reader = OtelTraceWrapper<oio::Reader>;
93    type Writer = OtelTraceWrapper<oio::Writer>;
94    type Lister = OtelTraceWrapper<oio::Lister>;
95    type Deleter = oio::Deleter;
96    type Copier = oio::Copier;
97    type Composer = oio::Composer;
98
99    fn info(&self) -> ServiceInfo {
100        let tracer = global::tracer("opendal");
101        tracer.in_span("info", |_cx| self.inner.info())
102    }
103
104    fn capability(&self) -> Capability {
105        self.inner.capability()
106    }
107
108    fn compose(&self, ctx: &OperationContext, to: &str, args: OpCompose) -> Result<Self::Composer> {
109        self.inner.compose(ctx, to, args)
110    }
111
112    async fn create_dir(
113        &self,
114        ctx: &OperationContext,
115        path: &str,
116        args: OpCreateDir,
117    ) -> Result<RpCreateDir> {
118        let tracer = global::tracer("opendal");
119        let mut span = tracer.start("create");
120        span.set_attribute(KeyValue::new("path", path.to_string()));
121        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
122        let cx = TraceContext::current_with_span(span);
123        self.inner
124            .create_dir(ctx, path, args)
125            .with_context(cx)
126            .await
127    }
128
129    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
130        let tracer = global::tracer("opendal");
131        let mut span = tracer.start("read");
132        span.set_attribute(KeyValue::new("path", path.to_string()));
133        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
134        let cx = TraceContext::current_with_span(span);
135        self.inner
136            .read(ctx, path, args)
137            .map(|r| OtelTraceWrapper::new(cx, r))
138    }
139
140    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
141        let tracer = global::tracer("opendal");
142        let mut span = tracer.start("write");
143        span.set_attribute(KeyValue::new("path", path.to_string()));
144        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
145        let cx = TraceContext::current_with_span(span);
146        self.inner
147            .write(ctx, path, args)
148            .map(|r| OtelTraceWrapper::new(cx, r))
149    }
150
151    fn copy(
152        &self,
153        ctx: &OperationContext,
154        from: &str,
155        to: &str,
156        args: OpCopy,
157    ) -> Result<Self::Copier> {
158        let tracer = global::tracer("opendal");
159        let mut span = tracer.start("copy");
160        span.set_attribute(KeyValue::new("from", from.to_string()));
161        span.set_attribute(KeyValue::new("to", to.to_string()));
162        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
163        let cx = TraceContext::current_with_span(span);
164        let _guard = cx.attach();
165        self.inner.copy(ctx, from, to, args)
166    }
167
168    async fn rename(
169        &self,
170        ctx: &OperationContext,
171        from: &str,
172        to: &str,
173        args: OpRename,
174    ) -> Result<RpRename> {
175        let tracer = global::tracer("opendal");
176        let mut span = tracer.start("rename");
177        span.set_attribute(KeyValue::new("from", from.to_string()));
178        span.set_attribute(KeyValue::new("to", to.to_string()));
179        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
180        let cx = TraceContext::current_with_span(span);
181        self.inner
182            .rename(ctx, from, to, args)
183            .with_context(cx)
184            .await
185    }
186
187    async fn restore(
188        &self,
189        ctx: &OperationContext,
190        path: &str,
191        args: OpRestore,
192    ) -> Result<RpRestore> {
193        let tracer = global::tracer("opendal");
194        let mut span = tracer.start("restore");
195        span.set_attribute(KeyValue::new("path", path.to_string()));
196        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
197        let cx = TraceContext::current_with_span(span);
198        self.inner.restore(ctx, path, args).with_context(cx).await
199    }
200
201    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
202        let tracer = global::tracer("opendal");
203        let mut span = tracer.start("stat");
204        span.set_attribute(KeyValue::new("path", path.to_string()));
205        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
206        let cx = TraceContext::current_with_span(span);
207        self.inner.stat(ctx, path, args).with_context(cx).await
208    }
209
210    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
211        self.inner.delete(ctx)
212    }
213
214    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
215        let tracer = global::tracer("opendal");
216        let mut span = tracer.start("list");
217        span.set_attribute(KeyValue::new("path", path.to_string()));
218        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
219        let cx = TraceContext::current_with_span(span);
220        self.inner
221            .list(ctx, path, args)
222            .map(|s| OtelTraceWrapper::new(cx, s))
223    }
224
225    async fn presign(
226        &self,
227        ctx: &OperationContext,
228        path: &str,
229        args: OpPresign,
230    ) -> Result<RpPresign> {
231        let tracer = global::tracer("opendal");
232        let mut span = tracer.start("presign");
233        span.set_attribute(KeyValue::new("path", path.to_string()));
234        span.set_attribute(KeyValue::new("args", format!("{args:?}")));
235        let cx = TraceContext::current_with_span(span);
236        self.inner.presign(ctx, path, args).with_context(cx).await
237    }
238}
239
240#[doc(hidden)]
241pub struct OtelTraceWrapper<R> {
242    cx: TraceContext,
243    inner: R,
244}
245
246impl<R> OtelTraceWrapper<R> {
247    fn new(cx: TraceContext, inner: R) -> Self {
248        Self { cx, inner }
249    }
250
251    fn child_context(&self, name: &'static str, range: BytesRange) -> TraceContext {
252        let tracer = global::tracer("opendal");
253        let mut span = tracer.start_with_context(name, &self.cx);
254        span.set_attribute(KeyValue::new("range", range.to_string()));
255        self.cx.with_span(span)
256    }
257}
258
259impl<R: oio::ReadStream> oio::ReadStream for OtelTraceWrapper<R> {
260    async fn read(&mut self) -> Result<Buffer> {
261        self.inner.read().with_context(self.cx.clone()).await
262    }
263}
264
265impl<R: oio::Read> oio::Read for OtelTraceWrapper<R> {
266    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
267        let cx = self.child_context("reader.open", range);
268        let (rp, stream) = self.inner.open(range).with_context(cx.clone()).await?;
269        Ok((
270            rp,
271            Box::new(OtelTraceWrapper::new(cx, stream)) as Box<dyn oio::ReadStreamDyn>,
272        ))
273    }
274
275    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
276        let cx = self.child_context("reader.read", range);
277        self.inner.read(range).with_context(cx).await
278    }
279}
280
281impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
282    async fn write(&mut self, bs: Buffer) -> Result<()> {
283        self.inner.write(bs).with_context(self.cx.clone()).await
284    }
285
286    async fn copy_from(&mut self, path: &str, args: OpRead, range: BytesRange) -> Result<()> {
287        self.inner
288            .copy_from(path, args, range)
289            .with_context(self.cx.clone())
290            .await
291    }
292
293    async fn abort(&mut self) -> Result<()> {
294        self.inner.abort().with_context(self.cx.clone()).await
295    }
296
297    async fn close(&mut self) -> Result<Metadata> {
298        self.inner.close().with_context(self.cx.clone()).await
299    }
300}
301
302impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
303    async fn next(&mut self) -> Result<Option<oio::Entry>> {
304        self.inner.next().with_context(self.cx.clone()).await
305    }
306}
opendal_layer_oteltrace/lib.rs

opendal_layer_oteltrace/
lib.rs
