rustev_otel/
config.rs

1//! Sink configuration (spec 028, 3.2.4, 3.6.3, 3.7, 3.8 and 3.9.4): every
2//! bound has a stated range, and a configuration outside it is refused at
3//! construction.
4
5use std::collections::BTreeSet;
6use std::fmt;
7use std::ops::RangeInclusive;
8
9use rustev_contract::redaction::{DigestKey, RedactionPolicy, RedactorError};
10
11use crate::project::Bindings;
12
13/// What a receipt establishes (3.8.2).
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
15pub enum AckMode {
16    /// The tree is in the bounded export queue; a later export failure is
17    /// counted loss.
18    #[default]
19    Accepted,
20    /// The exporter reported success for the batch holding the tree.
21    Exported,
22}
23
24/// Whether a registered valid parent's sampled flag decides (3.9.4).
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
26pub enum ParentSampling {
27    #[default]
28    Honor,
29    Ignore,
30}
31
32/// The rule for a decision no parent decides (3.9.4).
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
34pub enum RootSampling {
35    #[default]
36    Always,
37    Never,
38    /// Decided from the trace id as OpenTelemetry's trace-id-ratio sampler
39    /// does, so it agrees across services.
40    Ratio {
41        per_million: u32,
42    },
43}
44
45/// Outcomes sampled whatever the parent or root rule said (3.9.4).
46#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
47pub enum AlwaysSampleOn {
48    /// The outcome is unresolved, or any step is.
49    Unresolved,
50    /// The outcome is `escalate`.
51    Refusal,
52    /// Any runtime fallback transition.
53    Fallback,
54    /// The decision was cancelled.
55    Cancelled,
56}
57
58/// Sampling (3.9.4). Metrics are never sampled (3.6.2).
59#[derive(Debug, Clone, Default, PartialEq, Eq)]
60pub struct Sampling {
61    pub parent: ParentSampling,
62    pub root: RootSampling,
63    pub always_sample_on: BTreeSet<AlwaysSampleOn>,
64}
65
66/// The sink's configuration. [`SinkConfig::default`] holds every default
67/// the spec states.
68#[derive(Debug, Clone)]
69pub struct SinkConfig {
70    /// The span schema version to emit (3.2.4); only `1` exists.
71    pub schema: u32,
72    pub ack: AckMode,
73    /// The redaction policy (3.4); the default policy when the host gives
74    /// none.
75    pub policy: RedactionPolicy,
76    /// The digest key, needed when any effective action is `digest`.
77    pub key: Option<DigestKey>,
78    pub bindings: Bindings,
79    pub sampling: Sampling,
80    /// Log records for unresolved, escalated, cancelled and failed work
81    /// (3.6.3).
82    pub logs: bool,
83    pub max_queued_spans: usize,
84    /// Spans per export batch; a single larger tree is exported alone.
85    pub batch_size: usize,
86    pub export_timeout_ms: u64,
87    /// The per-attribute value bound in bytes (3.8.5).
88    pub value_bound: usize,
89    pub max_pending_contexts: usize,
90    pub context_ttl_ms: u64,
91    /// Exchange records waiting to be joined (3.7.3).
92    pub join_buffer: usize,
93}
94
95impl Default for SinkConfig {
96    fn default() -> Self {
97        SinkConfig {
98            schema: 1,
99            ack: AckMode::Accepted,
100            policy: RedactionPolicy::default(),
101            key: None,
102            bindings: Bindings::default(),
103            sampling: Sampling::default(),
104            logs: false,
105            max_queued_spans: 65_536,
106            batch_size: 512,
107            export_timeout_ms: 10_000,
108            value_bound: 4096,
109            max_pending_contexts: 4096,
110            context_ttl_ms: 600_000,
111            join_buffer: 4096,
112        }
113    }
114}
115
116/// The stated ranges (3.8.5, 3.7.1, 3.7.3).
117pub mod ranges {
118    use std::ops::RangeInclusive;
119    pub const MAX_QUEUED_SPANS: RangeInclusive<usize> = 1..=1_048_576;
120    pub const BATCH_SIZE: RangeInclusive<usize> = 1..=8192;
121    pub const EXPORT_TIMEOUT_MS: RangeInclusive<u64> = 1..=600_000;
122    pub const VALUE_BOUND: RangeInclusive<usize> = 16..=4096;
123    pub const MAX_PENDING_CONTEXTS: RangeInclusive<usize> = 1..=65_536;
124    pub const CONTEXT_TTL_MS: RangeInclusive<u64> = 1..=86_400_000;
125    pub const JOIN_BUFFER: RangeInclusive<usize> = 1..=65_536;
126    pub const RATIO_PER_MILLION: RangeInclusive<u32> = 0..=1_000_000;
127}
128
129/// Why a sink was not constructed.
130#[derive(Debug, Clone, PartialEq, Eq)]
131pub enum ConfigError {
132    /// A setting outside its stated range.
133    OutOfRange { setting: &'static str, value: u64 },
134    /// A span schema version this build cannot emit.
135    UnknownSchema { schema: u32 },
136    /// The policy failed validation, or digests with no key.
137    Redaction(RedactorError),
138    /// Logs are on but the exporter takes no log records.
139    LogsUnsupported,
140    /// Construction needs a Tokio runtime for the export worker.
141    NoRuntime,
142}
143
144impl fmt::Display for ConfigError {
145    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
146        match self {
147            ConfigError::OutOfRange { setting, value } => {
148                write!(f, "{setting} = {value} is outside its stated range")
149            }
150            ConfigError::UnknownSchema { schema } => {
151                write!(
152                    f,
153                    "span schema rustev.otel/{schema} is not emitted by this build"
154                )
155            }
156            ConfigError::Redaction(e) => write!(f, "{e}"),
157            ConfigError::LogsUnsupported => f.write_str("logs are on and the exporter takes none"),
158            ConfigError::NoRuntime => f.write_str("the sink needs a Tokio runtime"),
159        }
160    }
161}
162
163impl std::error::Error for ConfigError {}
164
165fn check<T: PartialOrd + Copy + Into<u64>>(
166    setting: &'static str,
167    v: T,
168    r: RangeInclusive<T>,
169) -> Result<(), ConfigError> {
170    if r.contains(&v) {
171        Ok(())
172    } else {
173        Err(ConfigError::OutOfRange {
174            setting,
175            value: v.into(),
176        })
177    }
178}
179
180fn check_usize(
181    setting: &'static str,
182    v: usize,
183    r: RangeInclusive<usize>,
184) -> Result<(), ConfigError> {
185    if r.contains(&v) {
186        Ok(())
187    } else {
188        Err(ConfigError::OutOfRange {
189            setting,
190            value: v as u64,
191        })
192    }
193}
194
195impl SinkConfig {
196    /// Every range of this module, and the schema version.
197    pub fn validate(&self) -> Result<(), ConfigError> {
198        if self.schema != 1 {
199            return Err(ConfigError::UnknownSchema {
200                schema: self.schema,
201            });
202        }
203        check_usize(
204            "max_queued_spans",
205            self.max_queued_spans,
206            ranges::MAX_QUEUED_SPANS,
207        )?;
208        check_usize("batch_size", self.batch_size, ranges::BATCH_SIZE)?;
209        check(
210            "export_timeout_ms",
211            self.export_timeout_ms,
212            ranges::EXPORT_TIMEOUT_MS,
213        )?;
214        check_usize("value_bound", self.value_bound, ranges::VALUE_BOUND)?;
215        check_usize(
216            "max_pending_contexts",
217            self.max_pending_contexts,
218            ranges::MAX_PENDING_CONTEXTS,
219        )?;
220        check(
221            "context_ttl_ms",
222            self.context_ttl_ms,
223            ranges::CONTEXT_TTL_MS,
224        )?;
225        check_usize("join_buffer", self.join_buffer, ranges::JOIN_BUFFER)?;
226        if let RootSampling::Ratio { per_million } = self.sampling.root {
227            check("ratio.per_million", per_million, ranges::RATIO_PER_MILLION)?;
228        }
229        Ok(())
230    }
231}
rustev_otel/config.rs

rustev_otel/
config.rs
