opendal_layer_dtrace/
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#![cfg(target_os = "linux")]
19#![doc = include_str!("../README.md")]
20#![cfg_attr(docsrs, feature(doc_cfg))]
21#![cfg_attr(docsrs, doc(auto_cfg))]
22#![deny(missing_docs)]
23use std::ffi::CString;
24use std::sync::Arc;
25
26use bytes::Buf;
27use opendal_core::raw::*;
28use opendal_core::*;
29use probe::probe_lazy;
30
31/// `DtraceLayer` supports User Statically-Defined Tracing (USDT) on Linux.
32///
33/// Enable this experimental layer with `features = ["layers-dtrace"]` in
34/// `Cargo.toml`.
35///
36/// For now we have following probes:
37///
38/// ### For Service
39///
40/// 1. ${operation}_start, arguments: path
41///     1. create_dir
42///     2. read
43///     3. write
44///     4. stat
45///     5. delete
46///     6. list
47///     7. presign
48///
49/// 2. ${operation}_end, arguments: path
50///     1. create_dir
51///     2. read
52///     3. write
53///     4. stat
54///     5. delete
55///     6. list
56///     7. presign
57///
58/// ### For Reader
59///
60/// 1. reader_read_start, arguments: path, range
61/// 2. reader_read_ok, arguments: path, range, length
62/// 3. reader_read_error, arguments: path, range
63///
64/// ### For Writer
65///
66/// 1. writer_write_start, arguments: path
67/// 2. writer_write_ok, arguments: path, length
68/// 3. writer_write_error, arguments: path
69/// 4. writer_abort_start, arguments: path
70/// 5. writer_abort_ok, arguments: path
71/// 6. writer_abort_error, arguments: path
72/// 7. writer_close_start, arguments: path
73/// 8. writer_close_ok, arguments: path
74/// 9. writer_close_error, arguments: path
75///
76/// Example:
77///
78/// ```no_run
79/// # use opendal_core::services;
80/// # use opendal_core::Operator;
81/// # use opendal_core::Result;
82/// # use opendal_layer_dtrace::DtraceLayer;
83/// #
84/// # #[tokio::main]
85/// # async fn main() -> Result<()> {
86/// // `Service` provides the low level APIs, we will use `Operator` normally.
87/// let op: Operator = Operator::new(services::Memory::default().root("/tmp"))?
88///     .layer(DtraceLayer::new());
89///
90/// let path = "/tmp/test.txt";
91/// for _ in 1..100000 {
92///     let bs = vec![0; 64 * 1024 * 1024];
93///     op.write(path, bs).await?;
94///     op.read(path).await?;
95/// }
96/// # Ok(())
97/// # }
98/// ```
99///
100/// Then you can use `readelf -n target/debug/examples/dtrace` to see the probes:
101///
102/// ```text
103/// Displaying notes found in: .note.stapsdt
104///   Owner                Data size        Description
105///   stapsdt              0x00000039       NT_STAPSDT (SystemTap probe descriptors)
106///     Provider: opendal
107///     Name: create_dir_start
108///     Location: 0x00000000000f8f05, Base: 0x0000000000000000, Semaphore: 0x00000000003649f8
109///     Arguments: -8@%rax
110///   stapsdt              0x00000037       NT_STAPSDT (SystemTap probe descriptors)
111///     Provider: opendal
112///     Name: create_dir_end
113///     Location: 0x00000000000f9284, Base: 0x0000000000000000, Semaphore: 0x00000000003649fa
114///     Arguments: -8@%rax
115///   stapsdt              0x0000003c       NT_STAPSDT (SystemTap probe descriptors)
116///     Provider: opendal
117///     Name: blocking_list_start
118///     Location: 0x00000000000f9487, Base: 0x0000000000000000, Semaphore: 0x0000000000364a28
119///     Arguments: -8@%rax
120///   stapsdt              0x0000003a       NT_STAPSDT (SystemTap probe descriptors)
121///     Provider: opendal
122///     Name: blocking_list_end
123///     Location: 0x00000000000f9546, Base: 0x0000000000000000, Semaphore: 0x0000000000364a2a
124///     Arguments: -8@%rax
125///   stapsdt              0x0000003c       NT_STAPSDT (SystemTap probe descriptors)
126/// ```
127#[derive(Clone, Debug, Default)]
128#[non_exhaustive]
129pub struct DtraceLayer {}
130
131impl DtraceLayer {
132    /// Create a new [`DtraceLayer`].
133    pub fn new() -> Self {
134        Self::default()
135    }
136}
137
138impl Layer for DtraceLayer {
139    fn apply_service(&self, inner: Servicer) -> Servicer {
140        Arc::new(self.layer(inner))
141    }
142}
143
144impl DtraceLayer {
145    fn layer(&self, inner: Servicer) -> DTraceService {
146        DTraceService { inner }
147    }
148}
149
150#[doc(hidden)]
151#[derive(Debug)]
152pub struct DTraceService {
153    inner: Servicer,
154}
155
156impl Service for DTraceService {
157    type Reader = DtraceLayerWrapper<oio::Reader>;
158    type Writer = DtraceLayerWrapper<oio::Writer>;
159    type Lister = oio::Lister;
160    type Deleter = oio::Deleter;
161    type Copier = oio::Copier;
162    type Composer = oio::Composer;
163
164    fn info(&self) -> ServiceInfo {
165        self.inner.info()
166    }
167
168    fn capability(&self) -> Capability {
169        self.inner.capability()
170    }
171
172    fn compose(&self, ctx: &OperationContext, to: &str, args: OpCompose) -> Result<Self::Composer> {
173        self.inner.compose(ctx, to, args)
174    }
175
176    async fn create_dir(
177        &self,
178        ctx: &OperationContext,
179        path: &str,
180        args: OpCreateDir,
181    ) -> Result<RpCreateDir> {
182        let c_path = CString::new(path).unwrap();
183        probe_lazy!(opendal, create_dir_start, c_path.as_ptr());
184        let result = self.inner.create_dir(ctx, path, args).await;
185        probe_lazy!(opendal, create_dir_end, c_path.as_ptr());
186        result
187    }
188
189    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
190        let c_path = CString::new(path).unwrap();
191        probe_lazy!(opendal, read_start, c_path.as_ptr());
192        let result = self
193            .inner
194            .read(ctx, path, args)
195            .map(|r| DtraceLayerWrapper::new(r, path));
196        probe_lazy!(opendal, read_end, c_path.as_ptr());
197        result
198    }
199
200    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
201        let c_path = CString::new(path).unwrap();
202        probe_lazy!(opendal, write_start, c_path.as_ptr());
203        let result = self
204            .inner
205            .write(ctx, path, args)
206            .map(|r| DtraceLayerWrapper::new(r, path));
207
208        probe_lazy!(opendal, write_end, c_path.as_ptr());
209        result
210    }
211
212    fn copy(
213        &self,
214        ctx: &OperationContext,
215        from: &str,
216        to: &str,
217        args: OpCopy,
218    ) -> Result<Self::Copier> {
219        let c_from = CString::new(from).unwrap();
220        probe_lazy!(opendal, copy_start, c_from.as_ptr());
221        let result = self.inner.copy(ctx, from, to, args);
222        probe_lazy!(opendal, copy_end, c_from.as_ptr());
223        result
224    }
225
226    async fn rename(
227        &self,
228        ctx: &OperationContext,
229        from: &str,
230        to: &str,
231        args: OpRename,
232    ) -> Result<RpRename> {
233        self.inner.rename(ctx, from, to, args).await
234    }
235
236    async fn restore(
237        &self,
238        ctx: &OperationContext,
239        path: &str,
240        args: OpRestore,
241    ) -> Result<RpRestore> {
242        self.inner.restore(ctx, path, args).await
243    }
244
245    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
246        let c_path = CString::new(path).unwrap();
247        probe_lazy!(opendal, stat_start, c_path.as_ptr());
248        let result = self.inner.stat(ctx, path, args).await;
249        probe_lazy!(opendal, stat_end, c_path.as_ptr());
250        result
251    }
252
253    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
254        self.inner.delete(ctx)
255    }
256
257    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
258        let c_path = CString::new(path).unwrap();
259        probe_lazy!(opendal, list_start, c_path.as_ptr());
260        let result = self.inner.list(ctx, path, args);
261        probe_lazy!(opendal, list_end, c_path.as_ptr());
262        result
263    }
264
265    async fn presign(
266        &self,
267        ctx: &OperationContext,
268        path: &str,
269        args: OpPresign,
270    ) -> Result<RpPresign> {
271        let c_path = CString::new(path).unwrap();
272        probe_lazy!(opendal, presign_start, c_path.as_ptr());
273        let result = self.inner.presign(ctx, path, args).await;
274        probe_lazy!(opendal, presign_end, c_path.as_ptr());
275        result
276    }
277}
278
279#[doc(hidden)]
280pub struct DtraceLayerWrapper<R> {
281    inner: R,
282    path: String,
283    range: Option<BytesRange>,
284}
285
286impl<R> DtraceLayerWrapper<R> {
287    fn new(inner: R, path: &str) -> Self {
288        Self::with_range(inner, path, None)
289    }
290
291    fn with_range(inner: R, path: &str, range: Option<BytesRange>) -> Self {
292        Self {
293            inner,
294            path: path.to_string(),
295            range,
296        }
297    }
298
299    fn range_label(&self) -> String {
300        self.range
301            .map(|range| range.to_string())
302            .unwrap_or_default()
303    }
304}
305
306impl<R: oio::ReadStream> oio::ReadStream for DtraceLayerWrapper<R> {
307    async fn read(&mut self) -> Result<Buffer> {
308        let c_path = CString::new(self.path.clone()).unwrap();
309        let c_range = CString::new(self.range_label()).unwrap();
310        probe_lazy!(
311            opendal,
312            reader_read_start,
313            c_path.as_ptr(),
314            c_range.as_ptr()
315        );
316        match self.inner.read().await {
317            Ok(bs) => {
318                probe_lazy!(
319                    opendal,
320                    reader_read_ok,
321                    c_path.as_ptr(),
322                    c_range.as_ptr(),
323                    bs.remaining()
324                );
325                Ok(bs)
326            }
327            Err(e) => {
328                probe_lazy!(
329                    opendal,
330                    reader_read_error,
331                    c_path.as_ptr(),
332                    c_range.as_ptr()
333                );
334                Err(e)
335            }
336        }
337    }
338}
339
340impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
341    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
342        let c_path = CString::new(self.path.clone()).unwrap();
343        let c_range = CString::new(range.to_string()).unwrap();
344        probe_lazy!(
345            opendal,
346            reader_read_start,
347            c_path.as_ptr(),
348            c_range.as_ptr()
349        );
350        match self.inner.open(range).await {
351            Ok((rp, stream)) => {
352                probe_lazy!(
353                    opendal,
354                    reader_read_ok,
355                    c_path.as_ptr(),
356                    c_range.as_ptr(),
357                    0
358                );
359                Ok((
360                    rp,
361                    Box::new(DtraceLayerWrapper::with_range(
362                        stream,
363                        &self.path,
364                        Some(range),
365                    )) as Box<dyn oio::ReadStreamDyn>,
366                ))
367            }
368            Err(e) => {
369                probe_lazy!(
370                    opendal,
371                    reader_read_error,
372                    c_path.as_ptr(),
373                    c_range.as_ptr()
374                );
375                Err(e)
376            }
377        }
378    }
379
380    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
381        let c_path = CString::new(self.path.clone()).unwrap();
382        let c_range = CString::new(range.to_string()).unwrap();
383        probe_lazy!(
384            opendal,
385            reader_read_start,
386            c_path.as_ptr(),
387            c_range.as_ptr()
388        );
389        match self.inner.read(range).await {
390            Ok((rp, buffer)) => {
391                probe_lazy!(
392                    opendal,
393                    reader_read_ok,
394                    c_path.as_ptr(),
395                    c_range.as_ptr(),
396                    buffer.len()
397                );
398                Ok((rp, buffer))
399            }
400            Err(e) => {
401                probe_lazy!(
402                    opendal,
403                    reader_read_error,
404                    c_path.as_ptr(),
405                    c_range.as_ptr()
406                );
407                Err(e)
408            }
409        }
410    }
411}
412
413impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
414    async fn write(&mut self, bs: Buffer) -> Result<()> {
415        let c_path = CString::new(self.path.clone()).unwrap();
416        probe_lazy!(opendal, writer_write_start, c_path.as_ptr());
417        self.inner
418            .write(bs)
419            .await
420            .map(|_| {
421                probe_lazy!(opendal, writer_write_ok, c_path.as_ptr());
422            })
423            .inspect_err(|_| {
424                probe_lazy!(opendal, writer_write_error, c_path.as_ptr());
425            })
426    }
427
428    async fn copy_from(&mut self, path: &str, args: OpRead, range: BytesRange) -> Result<()> {
429        let c_path = CString::new(self.path.clone()).unwrap();
430        probe_lazy!(opendal, writer_write_start, c_path.as_ptr());
431        self.inner
432            .copy_from(path, args, range)
433            .await
434            .map(|_| {
435                probe_lazy!(opendal, writer_write_ok, c_path.as_ptr());
436            })
437            .inspect_err(|_| {
438                probe_lazy!(opendal, writer_write_error, c_path.as_ptr());
439            })
440    }
441
442    async fn abort(&mut self) -> Result<()> {
443        let c_path = CString::new(self.path.clone()).unwrap();
444        probe_lazy!(opendal, writer_poll_abort_start, c_path.as_ptr());
445        self.inner
446            .abort()
447            .await
448            .map(|_| {
449                probe_lazy!(opendal, writer_poll_abort_ok, c_path.as_ptr());
450            })
451            .inspect_err(|_| {
452                probe_lazy!(opendal, writer_poll_abort_error, c_path.as_ptr());
453            })
454    }
455
456    async fn close(&mut self) -> Result<Metadata> {
457        let c_path = CString::new(self.path.clone()).unwrap();
458        probe_lazy!(opendal, writer_close_start, c_path.as_ptr());
459        self.inner
460            .close()
461            .await
462            .inspect(|_| {
463                probe_lazy!(opendal, writer_close_ok, c_path.as_ptr());
464            })
465            .inspect_err(|_| {
466                probe_lazy!(opendal, writer_close_error, c_path.as_ptr());
467            })
468    }
469}
opendal_layer_dtrace/lib.rs

opendal_layer_dtrace/
lib.rs
