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
163    fn info(&self) -> ServiceInfo {
164        self.inner.info()
165    }
166
167    fn capability(&self) -> Capability {
168        self.inner.capability()
169    }
170
171    async fn create_dir(
172        &self,
173        ctx: &OperationContext,
174        path: &str,
175        args: OpCreateDir,
176    ) -> Result<RpCreateDir> {
177        let c_path = CString::new(path).unwrap();
178        probe_lazy!(opendal, create_dir_start, c_path.as_ptr());
179        let result = self.inner.create_dir(ctx, path, args).await;
180        probe_lazy!(opendal, create_dir_end, c_path.as_ptr());
181        result
182    }
183
184    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
185        let c_path = CString::new(path).unwrap();
186        probe_lazy!(opendal, read_start, c_path.as_ptr());
187        let result = self
188            .inner
189            .read(ctx, path, args)
190            .map(|r| DtraceLayerWrapper::new(r, path));
191        probe_lazy!(opendal, read_end, c_path.as_ptr());
192        result
193    }
194
195    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
196        let c_path = CString::new(path).unwrap();
197        probe_lazy!(opendal, write_start, c_path.as_ptr());
198        let result = self
199            .inner
200            .write(ctx, path, args)
201            .map(|r| DtraceLayerWrapper::new(r, path));
202
203        probe_lazy!(opendal, write_end, c_path.as_ptr());
204        result
205    }
206
207    fn copy(
208        &self,
209        ctx: &OperationContext,
210        from: &str,
211        to: &str,
212        args: OpCopy,
213        opts: OpCopier,
214    ) -> Result<Self::Copier> {
215        let c_from = CString::new(from).unwrap();
216        probe_lazy!(opendal, copy_start, c_from.as_ptr());
217        let result = self.inner.copy(ctx, from, to, args, opts);
218        probe_lazy!(opendal, copy_end, c_from.as_ptr());
219        result
220    }
221
222    async fn rename(
223        &self,
224        ctx: &OperationContext,
225        from: &str,
226        to: &str,
227        args: OpRename,
228    ) -> Result<RpRename> {
229        self.inner.rename(ctx, from, to, args).await
230    }
231
232    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
233        let c_path = CString::new(path).unwrap();
234        probe_lazy!(opendal, stat_start, c_path.as_ptr());
235        let result = self.inner.stat(ctx, path, args).await;
236        probe_lazy!(opendal, stat_end, c_path.as_ptr());
237        result
238    }
239
240    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
241        self.inner.delete(ctx)
242    }
243
244    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
245        let c_path = CString::new(path).unwrap();
246        probe_lazy!(opendal, list_start, c_path.as_ptr());
247        let result = self.inner.list(ctx, path, args);
248        probe_lazy!(opendal, list_end, c_path.as_ptr());
249        result
250    }
251
252    async fn presign(
253        &self,
254        ctx: &OperationContext,
255        path: &str,
256        args: OpPresign,
257    ) -> Result<RpPresign> {
258        let c_path = CString::new(path).unwrap();
259        probe_lazy!(opendal, presign_start, c_path.as_ptr());
260        let result = self.inner.presign(ctx, path, args).await;
261        probe_lazy!(opendal, presign_end, c_path.as_ptr());
262        result
263    }
264}
265
266#[doc(hidden)]
267pub struct DtraceLayerWrapper<R> {
268    inner: R,
269    path: String,
270    range: Option<BytesRange>,
271}
272
273impl<R> DtraceLayerWrapper<R> {
274    fn new(inner: R, path: &str) -> Self {
275        Self::with_range(inner, path, None)
276    }
277
278    fn with_range(inner: R, path: &str, range: Option<BytesRange>) -> Self {
279        Self {
280            inner,
281            path: path.to_string(),
282            range,
283        }
284    }
285
286    fn range_label(&self) -> String {
287        self.range
288            .map(|range| range.to_string())
289            .unwrap_or_default()
290    }
291}
292
293impl<R: oio::ReadStream> oio::ReadStream for DtraceLayerWrapper<R> {
294    async fn read(&mut self) -> Result<Buffer> {
295        let c_path = CString::new(self.path.clone()).unwrap();
296        let c_range = CString::new(self.range_label()).unwrap();
297        probe_lazy!(
298            opendal,
299            reader_read_start,
300            c_path.as_ptr(),
301            c_range.as_ptr()
302        );
303        match self.inner.read().await {
304            Ok(bs) => {
305                probe_lazy!(
306                    opendal,
307                    reader_read_ok,
308                    c_path.as_ptr(),
309                    c_range.as_ptr(),
310                    bs.remaining()
311                );
312                Ok(bs)
313            }
314            Err(e) => {
315                probe_lazy!(
316                    opendal,
317                    reader_read_error,
318                    c_path.as_ptr(),
319                    c_range.as_ptr()
320                );
321                Err(e)
322            }
323        }
324    }
325}
326
327impl<R: oio::Read> oio::Read for DtraceLayerWrapper<R> {
328    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
329        let c_path = CString::new(self.path.clone()).unwrap();
330        let c_range = CString::new(range.to_string()).unwrap();
331        probe_lazy!(
332            opendal,
333            reader_read_start,
334            c_path.as_ptr(),
335            c_range.as_ptr()
336        );
337        match self.inner.open(range).await {
338            Ok((rp, stream)) => {
339                probe_lazy!(
340                    opendal,
341                    reader_read_ok,
342                    c_path.as_ptr(),
343                    c_range.as_ptr(),
344                    0
345                );
346                Ok((
347                    rp,
348                    Box::new(DtraceLayerWrapper::with_range(
349                        stream,
350                        &self.path,
351                        Some(range),
352                    )) as Box<dyn oio::ReadStreamDyn>,
353                ))
354            }
355            Err(e) => {
356                probe_lazy!(
357                    opendal,
358                    reader_read_error,
359                    c_path.as_ptr(),
360                    c_range.as_ptr()
361                );
362                Err(e)
363            }
364        }
365    }
366
367    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
368        let c_path = CString::new(self.path.clone()).unwrap();
369        let c_range = CString::new(range.to_string()).unwrap();
370        probe_lazy!(
371            opendal,
372            reader_read_start,
373            c_path.as_ptr(),
374            c_range.as_ptr()
375        );
376        match self.inner.read(range).await {
377            Ok((rp, buffer)) => {
378                probe_lazy!(
379                    opendal,
380                    reader_read_ok,
381                    c_path.as_ptr(),
382                    c_range.as_ptr(),
383                    buffer.len()
384                );
385                Ok((rp, buffer))
386            }
387            Err(e) => {
388                probe_lazy!(
389                    opendal,
390                    reader_read_error,
391                    c_path.as_ptr(),
392                    c_range.as_ptr()
393                );
394                Err(e)
395            }
396        }
397    }
398}
399
400impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
401    async fn write(&mut self, bs: Buffer) -> Result<()> {
402        let c_path = CString::new(self.path.clone()).unwrap();
403        probe_lazy!(opendal, writer_write_start, c_path.as_ptr());
404        self.inner
405            .write(bs)
406            .await
407            .map(|_| {
408                probe_lazy!(opendal, writer_write_ok, c_path.as_ptr());
409            })
410            .inspect_err(|_| {
411                probe_lazy!(opendal, writer_write_error, c_path.as_ptr());
412            })
413    }
414
415    async fn abort(&mut self) -> Result<()> {
416        let c_path = CString::new(self.path.clone()).unwrap();
417        probe_lazy!(opendal, writer_poll_abort_start, c_path.as_ptr());
418        self.inner
419            .abort()
420            .await
421            .map(|_| {
422                probe_lazy!(opendal, writer_poll_abort_ok, c_path.as_ptr());
423            })
424            .inspect_err(|_| {
425                probe_lazy!(opendal, writer_poll_abort_error, c_path.as_ptr());
426            })
427    }
428
429    async fn close(&mut self) -> Result<Metadata> {
430        let c_path = CString::new(self.path.clone()).unwrap();
431        probe_lazy!(opendal, writer_close_start, c_path.as_ptr());
432        self.inner
433            .close()
434            .await
435            .inspect(|_| {
436                probe_lazy!(opendal, writer_close_ok, c_path.as_ptr());
437            })
438            .inspect_err(|_| {
439                probe_lazy!(opendal, writer_close_error, c_path.as_ptr());
440            })
441    }
442}
opendal_layer_dtrace/lib.rs

opendal_layer_dtrace/
lib.rs
