reth_tracing_otlp/
file.rs

1//! Local length-delimited OTLP trace batch exporter.
2
3use crate::trace_file_path;
4use eyre::WrapErr;
5use opentelemetry_proto::{
6    tonic::collector::trace::v1::ExportTraceServiceRequest,
7    transform::{
8        common::tonic::ResourceAttributesWithSchema,
9        trace::tonic::group_spans_by_resource_and_scope,
10    },
11};
12use opentelemetry_sdk::{
13    error::{OTelSdkError, OTelSdkResult},
14    trace::{SpanData, SpanExporter},
15    Resource,
16};
17use prost::Message;
18use reth_fs_util::FsPathError;
19use std::{
20    fs::{File, OpenOptions},
21    io::{BufReader, Read, Seek, SeekFrom, Write},
22    sync::Mutex,
23};
24use url::Url;
25
26#[cfg(unix)]
27use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
28
29#[derive(Debug)]
30pub(crate) struct FileSpanExporter {
31    file: Mutex<Option<File>>,
32    resource: ResourceAttributesWithSchema,
33}
34
35impl FileSpanExporter {
36    pub(crate) fn new(endpoint: &Url) -> eyre::Result<Self> {
37        let path = trace_file_path(endpoint)?;
38        let mut options = OpenOptions::new();
39        options.create(true).read(true).append(true);
40        #[cfg(unix)]
41        options.mode(0o600);
42        let file = options.open(&path).map_err(|err| FsPathError::open(err, &path))?;
43        // Readers can acquire advisory locks too, so reject shared existing files.
44        #[cfg(unix)]
45        eyre::ensure!(
46            file.metadata()?.permissions().mode() & 0o077 == 0,
47            "Trace file {path:?} must be owner-only; restrict its permissions to 0600"
48        );
49        // Keep the lock for the exporter's lifetime so partial writes cannot interleave.
50        file.try_lock()
51            .wrap_err_with(|| format!("Could not exclusively lock trace file {path:?}"))?;
52        // Never append after an incomplete record left by an interrupted writer.
53        // Inspect framing without allocating or decoding entire trace batches.
54        let length = file.metadata()?.len();
55        let mut reader = BufReader::new(&file);
56        while reader.stream_position()? < length {
57            let mut delimiter = [0u8; 10];
58            let mut used = 0;
59            for byte in &mut delimiter {
60                reader
61                    .read_exact(std::slice::from_mut(byte))
62                    .wrap_err("Incomplete OTLP file length delimiter")?;
63                used += 1;
64                if *byte & 0x80 == 0 {
65                    break;
66                }
67            }
68            let payload = prost::decode_length_delimiter(&delimiter[..used])?;
69            let end = reader
70                .stream_position()?
71                .checked_add(u64::try_from(payload)?)
72                .filter(|end| *end <= length)
73                .ok_or_else(|| eyre::eyre!("Incomplete OTLP file batch"))?;
74            reader.seek(SeekFrom::Start(end))?;
75        }
76        Ok(Self {
77            file: Mutex::new(Some(file)),
78            resource: (&Resource::builder_empty().build()).into(),
79        })
80    }
81}
82
83impl SpanExporter for FileSpanExporter {
84    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
85        let request = ExportTraceServiceRequest {
86            resource_spans: group_spans_by_resource_and_scope(batch, &self.resource),
87        };
88        let bytes = request.encode_length_delimited_to_vec();
89        let mut file =
90            self.file.lock().map_err(|err| OTelSdkError::InternalFailure(err.to_string()))?;
91        let writer = file.as_mut().ok_or_else(|| {
92            OTelSdkError::InternalFailure("Trace file disabled after a write failure".into())
93        })?;
94        let result = writer.write_all(&bytes).and_then(|()| writer.flush());
95        if result.is_err() {
96            // A failed write may have committed a partial record. Do not append
97            // another batch after it, including on a later SDK retry.
98            file.take();
99        }
100        result.map_err(|err| OTelSdkError::InternalFailure(err.to_string()))
101    }
102
103    fn set_resource(&mut self, resource: &Resource) {
104        self.resource = resource.into();
105    }
106}
reth_tracing_otlp/file.rs

reth_tracing_otlp/
file.rs
