veecle_telemetry/
id.rs

1// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
2// Copyright 2025 Veecle GmbH.
3//
4// This file has been modified from the original TiKV implementation.
5
6//! Unique identifiers for traces and spans.
7//!
8//! This module provides the core identifier types used throughout the telemetry system
9//! to uniquely identify traces and spans in distributed tracing scenarios.
10//!
11//! # Core Types
12//!
13//! - [`SpanId`]: An identifier that uniquely identifies a span within a process.
14//! - [`SpanContext`]: A combination of process id and span id that uniquely identifies a span globally.
15
16use core::fmt;
17use core::num::NonZeroU64;
18use core::str::FromStr;
19
20/// A globally-unique id identifying a process.
21///
22/// The primary purpose of this id is to provide a globally-unique context within which
23/// [`ThreadId`]s and [`SpanContext`]s are guaranteed to be unique. On a normal operating system
24/// that is the process, on other systems it should be whatever is the closest equivalent, e.g. for
25/// most embedded setups it should be unique for each time the system is restarted.
26#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Default)]
27pub struct ProcessId(u128);
28
29impl ProcessId {
30    /// Uses a random number generator to generate the [`ProcessId`].
31    pub fn random(rng: &mut impl rand::Rng) -> Self {
32        Self(rng.random())
33    }
34
35    /// Creates a [`ProcessId`] from a raw value
36    ///
37    /// Extra care needs to be taken that this is not a constant value or re-used in any way.
38    ///
39    /// When possible prefer using [`ProcessId::random`].
40    pub const fn from_raw(raw: u128) -> Self {
41        Self(raw)
42    }
43
44    /// Returns the raw value of this id.
45    pub fn to_raw(self) -> u128 {
46        self.0
47    }
48}
49
50impl fmt::Display for ProcessId {
51    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52        write!(f, "{:032x}", self.0)
53    }
54}
55
56impl FromStr for ProcessId {
57    type Err = core::num::ParseIntError;
58
59    fn from_str(s: &str) -> Result<Self, Self::Err> {
60        u128::from_str_radix(s, 16).map(ProcessId)
61    }
62}
63
64impl serde::Serialize for ProcessId {
65    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
66    where
67        S: serde::Serializer,
68    {
69        let mut hex_bytes = [0u8; size_of::<u128>() * 2];
70        hex::encode_to_slice(self.0.to_le_bytes(), &mut hex_bytes).unwrap();
71
72        serializer.serialize_str(str::from_utf8(&hex_bytes).unwrap())
73    }
74}
75
76impl<'de> serde::Deserialize<'de> for ProcessId {
77    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
78    where
79        D: serde::Deserializer<'de>,
80    {
81        let bytes: [u8; size_of::<u128>()] = hex::serde::deserialize(deserializer)?;
82
83        Ok(ProcessId(u128::from_le_bytes(bytes)))
84    }
85}
86
87/// A globally-unique id identifying a thread within a specific process.
88///
89/// The primary purpose of this id is to allow the consumer of telemetry messages to associate
90/// spans with the callstack they came from to reconstruct parent-child relationships. On a normal
91/// operating system this is the thread, on other systems it should be whatever is the closest
92/// equivalent, e.g. for FreeRTOS it would be a task. On a single threaded bare-metal system it
93/// would be a constant as there is only the one callstack.
94#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
95pub struct ThreadId {
96    /// The globally-unique id for the process this thread is within.
97    pub process: ProcessId,
98
99    /// The process-unique id for this thread within the process.
100    raw: NonZeroU64,
101}
102
103impl ThreadId {
104    /// Creates a [`ThreadId`] from a raw value.
105    ///
106    /// Extra care needs to be taken that this is not a constant value or re-used within this
107    /// process in any way.
108    pub const fn from_raw(process: ProcessId, raw: NonZeroU64) -> Self {
109        Self { process, raw }
110    }
111
112    /// Returns the raw value of this id.
113    pub fn raw(&self) -> NonZeroU64 {
114        self.raw
115    }
116}
117
118impl fmt::Display for ThreadId {
119    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
120        let Self { process, raw } = self;
121        write!(f, "{process}:{raw:016x}")
122    }
123}
124
125/// Errors that can occur while parsing [`ThreadId`] from a string.
126#[derive(Clone, Debug)]
127pub enum ParseThreadIdError {
128    /// The string is missing a `:` separator.
129    MissingSeparator,
130
131    /// The embedded [`ProcessId`] failed to parse.
132    InvalidProcessId(core::num::ParseIntError),
133
134    /// The embedded [`ThreadId`] failed to parse.
135    InvalidThreadId(core::num::ParseIntError),
136
137    /// The embedded [`ThreadId`] had a zero value.
138    ZeroThreadId,
139}
140
141impl fmt::Display for ParseThreadIdError {
142    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
143        match self {
144            Self::MissingSeparator => f.write_str("missing ':' separator"),
145            Self::InvalidProcessId(_) => f.write_str("failed to parse process id"),
146            Self::InvalidThreadId(_) => f.write_str("failed to parse thread id"),
147            Self::ZeroThreadId => f.write_str("zero thread id"),
148        }
149    }
150}
151
152impl core::error::Error for ParseThreadIdError {
153    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
154        match self {
155            Self::MissingSeparator => None,
156            Self::InvalidProcessId(error) => Some(error),
157            Self::InvalidThreadId(error) => Some(error),
158            Self::ZeroThreadId => None,
159        }
160    }
161}
162
163impl FromStr for ThreadId {
164    type Err = ParseThreadIdError;
165
166    fn from_str(s: &str) -> Result<Self, Self::Err> {
167        let Some((process, thread)) = s.split_once(":") else {
168            return Err(ParseThreadIdError::MissingSeparator);
169        };
170        let process = ProcessId::from_str(process).map_err(ParseThreadIdError::InvalidProcessId)?;
171        let thread = NonZeroU64::new(
172            u64::from_str_radix(thread, 16).map_err(ParseThreadIdError::InvalidThreadId)?,
173        )
174        .ok_or(ParseThreadIdError::ZeroThreadId)?;
175        Ok(Self::from_raw(process, thread))
176    }
177}
178
179impl serde::Serialize for ThreadId {
180    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
181    where
182        S: serde::Serializer,
183    {
184        let mut bytes = [0u8; 49];
185
186        hex::encode_to_slice(self.process.to_raw().to_le_bytes(), &mut bytes[..32]).unwrap();
187        bytes[32] = b':';
188        hex::encode_to_slice(self.raw().get().to_le_bytes(), &mut bytes[33..]).unwrap();
189
190        serializer.serialize_str(str::from_utf8(&bytes).unwrap())
191    }
192}
193
194impl<'de> serde::Deserialize<'de> for ThreadId {
195    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
196    where
197        D: serde::Deserializer<'de>,
198    {
199        use serde::de::Error;
200
201        let string = <&str>::deserialize(deserializer)?;
202
203        if string.len() != 49 {
204            return Err(D::Error::invalid_length(
205                string.len(),
206                &"expected 49 byte string",
207            ));
208        }
209
210        let bytes = string.as_bytes();
211
212        if bytes[32] != b':' {
213            return Err(D::Error::invalid_value(
214                serde::de::Unexpected::Str(string),
215                &"expected : separator at byte 32",
216            ));
217        }
218
219        let mut process = [0; 16];
220        hex::decode_to_slice(&bytes[..32], &mut process).map_err(D::Error::custom)?;
221        let process = ProcessId::from_raw(u128::from_le_bytes(process));
222
223        let mut thread = [0; 8];
224        hex::decode_to_slice(&bytes[33..], &mut thread).map_err(D::Error::custom)?;
225        let thread = NonZeroU64::new(u64::from_le_bytes(thread))
226            .ok_or_else(|| D::Error::custom("zero thread id"))?;
227
228        Ok(Self::from_raw(process, thread))
229    }
230}
231
232/// A process-unique id for a span.
233#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
234pub struct SpanId(pub u64);
235
236#[cfg(feature = "enable")]
237impl SpanId {
238    #[inline]
239    #[doc(hidden)]
240    /// Creates a non-zero [`SpanId`].
241    pub fn next_id() -> Self {
242        use core::sync::atomic;
243        static COUNTER: atomic::AtomicU64 = atomic::AtomicU64::new(1);
244        SpanId(COUNTER.fetch_add(1, atomic::Ordering::Relaxed))
245    }
246}
247
248impl fmt::Display for SpanId {
249    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
250        write!(f, "{:016x}", self.0)
251    }
252}
253
254impl FromStr for SpanId {
255    type Err = core::num::ParseIntError;
256
257    fn from_str(s: &str) -> Result<Self, Self::Err> {
258        u64::from_str_radix(s, 16).map(SpanId)
259    }
260}
261
262impl serde::Serialize for SpanId {
263    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
264    where
265        S: serde::Serializer,
266    {
267        let mut hex_bytes = [0u8; size_of::<u64>() * 2];
268        hex::encode_to_slice(self.0.to_le_bytes(), &mut hex_bytes).unwrap();
269
270        serializer.serialize_str(str::from_utf8(&hex_bytes).unwrap())
271    }
272}
273
274impl<'de> serde::Deserialize<'de> for SpanId {
275    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
276    where
277        D: serde::Deserializer<'de>,
278    {
279        let bytes: [u8; size_of::<u64>()] = hex::serde::deserialize(deserializer)?;
280
281        Ok(SpanId(u64::from_le_bytes(bytes)))
282    }
283}
284
285/// A struct representing the context of a span, including its [`ProcessId`] and [`SpanId`].
286#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
287pub struct SpanContext {
288    /// The id of the process this span belongs to.
289    pub process_id: ProcessId,
290    /// The unique id of this span.
291    pub span_id: SpanId,
292}
293
294impl SpanContext {
295    /// Creates a new `SpanContext` with the given [`ProcessId`] and [`SpanId`].
296    ///
297    /// # Examples
298    ///
299    /// ```
300    /// use veecle_telemetry::{ProcessId, SpanId, SpanContext};
301    ///
302    /// let span_context = SpanContext::new(ProcessId::from_raw(12), SpanId(13));
303    /// ```
304    pub fn new(process_id: ProcessId, span_id: SpanId) -> Self {
305        Self {
306            process_id,
307            span_id,
308        }
309    }
310}
311
312impl fmt::Display for SpanContext {
313    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
314        let Self {
315            process_id,
316            span_id,
317        } = self;
318        write!(f, "{process_id}:{span_id}")
319    }
320}
321
322/// Errors that can occur while parsing [`SpanContext`] from a string.
323#[derive(Clone, Debug)]
324pub enum ParseSpanContextError {
325    /// The string is missing a `:` separator.
326    MissingSeparator,
327
328    /// The embedded [`ProcessId`] failed to parse.
329    InvalidProcessId(core::num::ParseIntError),
330
331    /// The embedded [`SpanId`] failed to parse.
332    InvalidSpanId(core::num::ParseIntError),
333}
334
335impl fmt::Display for ParseSpanContextError {
336    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
337        match self {
338            Self::MissingSeparator => f.write_str("missing ':' separator"),
339            Self::InvalidProcessId(_) => f.write_str("failed to parse process id"),
340            Self::InvalidSpanId(_) => f.write_str("failed to parse span id"),
341        }
342    }
343}
344
345impl core::error::Error for ParseSpanContextError {
346    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
347        match self {
348            Self::MissingSeparator => None,
349            Self::InvalidProcessId(error) => Some(error),
350            Self::InvalidSpanId(error) => Some(error),
351        }
352    }
353}
354
355impl FromStr for SpanContext {
356    type Err = ParseSpanContextError;
357
358    fn from_str(s: &str) -> Result<Self, Self::Err> {
359        let Some((process_id, span_id)) = s.split_once(":") else {
360            return Err(ParseSpanContextError::MissingSeparator);
361        };
362        let process_id =
363            ProcessId::from_str(process_id).map_err(ParseSpanContextError::InvalidProcessId)?;
364        let span_id = SpanId::from_str(span_id).map_err(ParseSpanContextError::InvalidSpanId)?;
365        Ok(Self {
366            process_id,
367            span_id,
368        })
369    }
370}
371
372impl serde::Serialize for SpanContext {
373    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
374    where
375        S: serde::Serializer,
376    {
377        let mut bytes = [0u8; 49];
378
379        hex::encode_to_slice(self.process_id.to_raw().to_le_bytes(), &mut bytes[..32]).unwrap();
380        bytes[32] = b':';
381        hex::encode_to_slice(self.span_id.0.to_le_bytes(), &mut bytes[33..]).unwrap();
382
383        serializer.serialize_str(str::from_utf8(&bytes).unwrap())
384    }
385}
386
387impl<'de> serde::Deserialize<'de> for SpanContext {
388    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
389    where
390        D: serde::Deserializer<'de>,
391    {
392        use serde::de::Error;
393
394        let string = <&str>::deserialize(deserializer)?;
395
396        if string.len() != 49 {
397            return Err(D::Error::invalid_length(
398                string.len(),
399                &"expected 49 byte string",
400            ));
401        }
402
403        let bytes = string.as_bytes();
404
405        if bytes[32] != b':' {
406            return Err(D::Error::invalid_value(
407                serde::de::Unexpected::Str(string),
408                &"expected : separator at byte 32",
409            ));
410        }
411
412        let mut process = [0; 16];
413        hex::decode_to_slice(&bytes[..32], &mut process).map_err(D::Error::custom)?;
414
415        let mut span = [0; 8];
416        hex::decode_to_slice(&bytes[33..], &mut span).map_err(D::Error::custom)?;
417
418        Ok(Self {
419            process_id: ProcessId::from_raw(u128::from_le_bytes(process)),
420            span_id: SpanId(u64::from_le_bytes(span)),
421        })
422    }
423}
424
425#[cfg(all(test, feature = "std"))]
426mod tests {
427    use std::collections::HashSet;
428    use std::format;
429    use std::vec::Vec;
430
431    use test_case::test_case;
432
433    use super::*;
434
435    #[test]
436    #[cfg(not(miri))] // VERY slow with Miri.
437    #[allow(clippy::needless_collect)]
438    fn unique_id() {
439        let handles = std::iter::repeat_with(|| {
440            std::thread::spawn(|| {
441                std::iter::repeat_with(SpanId::next_id)
442                    .take(1000)
443                    .collect::<Vec<_>>()
444            })
445        })
446        .take(32)
447        .collect::<Vec<_>>();
448
449        let k = handles
450            .into_iter()
451            .flat_map(|h| h.join().unwrap())
452            .collect::<HashSet<_>>();
453
454        assert_eq!(k.len(), 32 * 1000);
455    }
456
457    #[test]
458    fn span_id_next_id_produces_non_zero_values() {
459        let ids: Vec<SpanId> = (0..100).map(|_| SpanId::next_id()).collect();
460
461        for id in &ids {
462            assert_ne!(id.0, 0, "SpanId::next_id() should not produce zero values");
463        }
464
465        let mut unique_ids = HashSet::new();
466        for id in &ids {
467            assert!(
468                unique_ids.insert(id.0),
469                "SpanId::next_id() should produce unique values"
470            );
471        }
472    }
473
474    #[test_case(SpanId(0), r#""0000000000000000""#, "0000000000000000")]
475    #[test_case(SpanId(1), r#""0100000000000000""#, "0000000000000001")]
476    #[test_case(SpanId(0x123), r#""2301000000000000""#, "0000000000000123")]
477    #[test_case(SpanId(u64::MAX), r#""ffffffffffffffff""#, "ffffffffffffffff")]
478    #[test_case(
479        SpanId(0xFEDCBA9876543210),
480        r#""1032547698badcfe""#,
481        "fedcba9876543210"
482    )]
483    #[test_case(
484        ProcessId::from_raw(0),
485        r#""00000000000000000000000000000000""#,
486        "00000000000000000000000000000000"
487    )]
488    #[test_case(
489        ProcessId::from_raw(1),
490        r#""01000000000000000000000000000000""#,
491        "00000000000000000000000000000001"
492    )]
493    #[test_case(
494        ProcessId::from_raw(0x123),
495        r#""23010000000000000000000000000000""#,
496        "00000000000000000000000000000123"
497    )]
498    #[test_case(
499        ProcessId::from_raw(0x123456789ABCDEF0FEDCBA9876543210),
500        r#""1032547698badcfef0debc9a78563412""#,
501        "123456789abcdef0fedcba9876543210"
502    )]
503    #[test_case(
504        ProcessId::from_raw(u128::MAX),
505        r#""ffffffffffffffffffffffffffffffff""#,
506        "ffffffffffffffffffffffffffffffff"
507    )]
508    #[test_case(
509        ThreadId::from_raw(ProcessId::from_raw(0), NonZeroU64::new(1).unwrap()),
510        r#""00000000000000000000000000000000:0100000000000000""#,
511        "00000000000000000000000000000000:0000000000000001"
512    )]
513    #[test_case(
514        ThreadId::from_raw(ProcessId::from_raw(0x123), NonZeroU64::new(0x456).unwrap()),
515        r#""23010000000000000000000000000000:5604000000000000""#,
516        "00000000000000000000000000000123:0000000000000456"
517    )]
518    #[test_case(
519        ThreadId::from_raw(ProcessId::from_raw(u128::MAX), NonZeroU64::new(u64::MAX).unwrap()),
520        r#""ffffffffffffffffffffffffffffffff:ffffffffffffffff""#,
521        "ffffffffffffffffffffffffffffffff:ffffffffffffffff"
522    )]
523    #[test_case(
524        ThreadId::from_raw(
525            ProcessId::from_raw(0x123456789ABCDEF0FEDCBA9876543210),
526            NonZeroU64::new(0xFEDCBA9876543210).unwrap(),
527        ),
528        r#""1032547698badcfef0debc9a78563412:1032547698badcfe""#,
529        "123456789abcdef0fedcba9876543210:fedcba9876543210"
530    )]
531    #[test_case(
532        SpanContext::new(ProcessId::from_raw(0), SpanId(0)),
533        r#""00000000000000000000000000000000:0000000000000000""#,
534        "00000000000000000000000000000000:0000000000000000"
535    )]
536    #[test_case(
537        SpanContext::new(ProcessId::from_raw(0x123), SpanId(0x456)),
538        r#""23010000000000000000000000000000:5604000000000000""#,
539        "00000000000000000000000000000123:0000000000000456"
540    )]
541    #[test_case(
542        SpanContext::new(ProcessId::from_raw(u128::MAX), SpanId(u64::MAX)),
543        r#""ffffffffffffffffffffffffffffffff:ffffffffffffffff""#,
544        "ffffffffffffffffffffffffffffffff:ffffffffffffffff"
545    )]
546    #[test_case(
547        SpanContext::new(
548            ProcessId::from_raw(0x123456789ABCDEF0FEDCBA9876543210),
549            SpanId(0xFEDCBA9876543210)
550        ),
551        r#""1032547698badcfef0debc9a78563412:1032547698badcfe""#,
552        "123456789abcdef0fedcba9876543210:fedcba9876543210"
553    )]
554    fn serialization<T>(value: T, expected_json: &str, expected_display: &str)
555    where
556        T: fmt::Display
557            + serde::Serialize
558            + FromStr<Err: fmt::Debug>
559            + serde::de::DeserializeOwned
560            + fmt::Debug
561            + Eq,
562    {
563        assert_eq!(serde_json::to_string(&value).unwrap(), expected_json);
564        assert_eq!(value, serde_json::from_str::<T>(expected_json).unwrap());
565
566        assert_eq!(format!("{value}"), expected_display);
567        assert_eq!(value, T::from_str(expected_display).unwrap());
568    }
569
570    #[test_case("")]
571    #[test_case("xyz")]
572    fn span_id_from_str_error(input: &str) {
573        assert!(SpanId::from_str(input).is_err());
574    }
575
576    #[test_case("")]
577    #[test_case("ffffffffffffffffffffffffffffffff0")]
578    #[test_case("xyz")]
579    fn process_id_from_str_error(input: &str) {
580        assert!(ProcessId::from_str(input).is_err());
581    }
582
583    #[test_case("")]
584    #[test_case("00000000000000000000000000000000")]
585    #[test_case("00000000000000000000000000000000:0000000000000000")]
586    #[test_case("00000000000000000000000000000000:xyz")]
587    #[test_case("00000000000000000000000000000001:")]
588    #[test_case(":0000000000000001")]
589    #[test_case("xyz")]
590    #[test_case("xyz:0000000000000001")]
591    fn thread_id_from_str_error(input: &str) {
592        assert!(ThreadId::from_str(input).is_err());
593    }
594
595    #[test_case("")]
596    #[test_case("00000000000000000000000000000000")]
597    #[test_case("00000000000000000000000000000000:xyz")]
598    #[test_case("00000000000000000000000000000001:")]
599    #[test_case(":0000000000000000")]
600    #[test_case("xyz")]
601    #[test_case("xyz:0000000000000000")]
602    fn span_context_from_str_error(input: &str) {
603        assert!(SpanContext::from_str(input).is_err());
604    }
605}
veecle_telemetry/id.rs

veecle_telemetry/
id.rs
