opentelemetry_proto/transform/
trace.rs

Help
1#[cfg(feature = "gen-tonic-messages")]
2pub mod tonic {
3    use crate::proto::tonic::resource::v1::Resource;
4    use crate::proto::tonic::trace::v1::{span, status, ResourceSpans, ScopeSpans, Span, Status};
5    use crate::transform::common::{
6        to_nanos,
7        tonic::{Attributes, ResourceAttributesWithSchema},
8    };
9    use opentelemetry::trace;
10    use opentelemetry::trace::{Link, SpanId, SpanKind};
11    use opentelemetry_sdk::export::trace::SpanData;
12    use std::collections::HashMap;
13
14    impl From<SpanKind> for span::SpanKind {
15        fn from(span_kind: SpanKind) -> Self {
16            match span_kind {
17                SpanKind::Client => span::SpanKind::Client,
18                SpanKind::Consumer => span::SpanKind::Consumer,
19                SpanKind::Internal => span::SpanKind::Internal,
20                SpanKind::Producer => span::SpanKind::Producer,
21                SpanKind::Server => span::SpanKind::Server,
22            }
23        }
24    }
25
26    impl From<&trace::Status> for status::StatusCode {
27        fn from(status: &trace::Status) -> Self {
28            match status {
29                trace::Status::Ok => status::StatusCode::Ok,
30                trace::Status::Unset => status::StatusCode::Unset,
31                trace::Status::Error { .. } => status::StatusCode::Error,
32            }
33        }
34    }
35
36    impl From<Link> for span::Link {
37        fn from(link: Link) -> Self {
38            span::Link {
39                trace_id: link.span_context.trace_id().to_bytes().to_vec(),
40                span_id: link.span_context.span_id().to_bytes().to_vec(),
41                trace_state: link.span_context.trace_state().header(),
42                attributes: Attributes::from(link.attributes).0,
43                dropped_attributes_count: link.dropped_attributes_count,
44                flags: link.span_context.trace_flags().to_u8() as u32,
45            }
46        }
47    }
48    impl From<opentelemetry_sdk::export::trace::SpanData> for Span {
49        fn from(source_span: opentelemetry_sdk::export::trace::SpanData) -> Self {
50            let span_kind: span::SpanKind = source_span.span_kind.into();
51            Span {
52                trace_id: source_span.span_context.trace_id().to_bytes().to_vec(),
53                span_id: source_span.span_context.span_id().to_bytes().to_vec(),
54                trace_state: source_span.span_context.trace_state().header(),
55                parent_span_id: {
56                    if source_span.parent_span_id != SpanId::INVALID {
57                        source_span.parent_span_id.to_bytes().to_vec()
58                    } else {
59                        vec![]
60                    }
61                },
62                flags: source_span.span_context.trace_flags().to_u8() as u32,
63                name: source_span.name.into_owned(),
64                kind: span_kind as i32,
65                start_time_unix_nano: to_nanos(source_span.start_time),
66                end_time_unix_nano: to_nanos(source_span.end_time),
67                dropped_attributes_count: source_span.dropped_attributes_count,
68                attributes: Attributes::from(source_span.attributes).0,
69                dropped_events_count: source_span.events.dropped_count,
70                events: source_span
71                    .events
72                    .into_iter()
73                    .map(|event| span::Event {
74                        time_unix_nano: to_nanos(event.timestamp),
75                        name: event.name.into(),
76                        attributes: Attributes::from(event.attributes).0,
77                        dropped_attributes_count: event.dropped_attributes_count,
78                    })
79                    .collect(),
80                dropped_links_count: source_span.links.dropped_count,
81                links: source_span.links.into_iter().map(Into::into).collect(),
82                status: Some(Status {
83                    code: status::StatusCode::from(&source_span.status).into(),
84                    message: match source_span.status {
85                        trace::Status::Error { description } => description.to_string(),
86                        _ => Default::default(),
87                    },
88                }),
89            }
90        }
91    }
92
93    impl ResourceSpans {
94        pub fn new(source_span: SpanData, resource: &ResourceAttributesWithSchema) -> Self {
95            let span_kind: span::SpanKind = source_span.span_kind.into();
96            ResourceSpans {
97                resource: Some(Resource {
98                    attributes: resource.attributes.0.clone(),
99                    dropped_attributes_count: 0,
100                }),
101                schema_url: resource.schema_url.clone().unwrap_or_default(),
102                scope_spans: vec![ScopeSpans {
103                    schema_url: source_span
104                        .instrumentation_lib
105                        .schema_url
106                        .as_ref()
107                        .map(ToString::to_string)
108                        .unwrap_or_default(),
109                    scope: Some((source_span.instrumentation_lib, None).into()),
110                    spans: vec![Span {
111                        trace_id: source_span.span_context.trace_id().to_bytes().to_vec(),
112                        span_id: source_span.span_context.span_id().to_bytes().to_vec(),
113                        trace_state: source_span.span_context.trace_state().header(),
114                        parent_span_id: {
115                            if source_span.parent_span_id != SpanId::INVALID {
116                                source_span.parent_span_id.to_bytes().to_vec()
117                            } else {
118                                vec![]
119                            }
120                        },
121                        flags: source_span.span_context.trace_flags().to_u8() as u32,
122                        name: source_span.name.into_owned(),
123                        kind: span_kind as i32,
124                        start_time_unix_nano: to_nanos(source_span.start_time),
125                        end_time_unix_nano: to_nanos(source_span.end_time),
126                        dropped_attributes_count: source_span.dropped_attributes_count,
127                        attributes: Attributes::from(source_span.attributes).0,
128                        dropped_events_count: source_span.events.dropped_count,
129                        events: source_span
130                            .events
131                            .into_iter()
132                            .map(|event| span::Event {
133                                time_unix_nano: to_nanos(event.timestamp),
134                                name: event.name.into(),
135                                attributes: Attributes::from(event.attributes).0,
136                                dropped_attributes_count: event.dropped_attributes_count,
137                            })
138                            .collect(),
139                        dropped_links_count: source_span.links.dropped_count,
140                        links: source_span.links.into_iter().map(Into::into).collect(),
141                        status: Some(Status {
142                            code: status::StatusCode::from(&source_span.status).into(),
143                            message: match source_span.status {
144                                trace::Status::Error { description } => description.to_string(),
145                                _ => Default::default(),
146                            },
147                        }),
148                    }],
149                }],
150            }
151        }
152    }
153
154    pub fn group_spans_by_resource_and_scope(
155        spans: Vec<SpanData>,
156        resource: &ResourceAttributesWithSchema,
157    ) -> Vec<ResourceSpans> {
158        // Group spans by their instrumentation library
159        let scope_map = spans.iter().fold(
160            HashMap::new(),
161            |mut scope_map: HashMap<&opentelemetry_sdk::InstrumentationLibrary, Vec<&SpanData>>,
162             span| {
163                let instrumentation = &span.instrumentation_lib;
164                scope_map.entry(instrumentation).or_default().push(span);
165                scope_map
166            },
167        );
168
169        // Convert the grouped spans into ScopeSpans
170        let scope_spans = scope_map
171            .into_iter()
172            .map(|(instrumentation, span_records)| ScopeSpans {
173                scope: Some((instrumentation, None).into()),
174                schema_url: resource.schema_url.clone().unwrap_or_default(),
175                spans: span_records
176                    .into_iter()
177                    .map(|span_data| span_data.clone().into())
178                    .collect(),
179            })
180            .collect();
181
182        // Wrap ScopeSpans into a single ResourceSpans
183        vec![ResourceSpans {
184            resource: Some(Resource {
185                attributes: resource.attributes.0.clone(),
186                dropped_attributes_count: 0,
187            }),
188            scope_spans,
189            schema_url: resource.schema_url.clone().unwrap_or_default(),
190        }]
191    }
192}
193
194#[cfg(test)]
195mod tests {
196    use crate::tonic::common::v1::any_value::Value;
197    use crate::transform::common::tonic::ResourceAttributesWithSchema;
198    use opentelemetry::trace::{
199        SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
200    };
201    use opentelemetry::KeyValue;
202    use opentelemetry_sdk::export::trace::SpanData;
203    use opentelemetry_sdk::resource::Resource;
204    use opentelemetry_sdk::trace::{SpanEvents, SpanLinks};
205    use opentelemetry_sdk::InstrumentationLibrary;
206    use std::borrow::Cow;
207    use std::time::{Duration, SystemTime};
208
209    fn create_test_span_data(instrumentation_name: &'static str) -> SpanData {
210        let span_context = SpanContext::new(
211            TraceId::from_u128(123),
212            SpanId::from_u64(456),
213            TraceFlags::default(),
214            false,
215            TraceState::default(),
216        );
217
218        SpanData {
219            span_context,
220            parent_span_id: SpanId::from_u64(0),
221            span_kind: SpanKind::Internal,
222            name: Cow::Borrowed("test_span"),
223            start_time: SystemTime::now(),
224            end_time: SystemTime::now() + Duration::from_secs(1),
225            attributes: vec![KeyValue::new("key", "value")],
226            dropped_attributes_count: 0,
227            events: SpanEvents::default(),
228            links: SpanLinks::default(),
229            status: Status::Unset,
230            instrumentation_lib: InstrumentationLibrary::builder(instrumentation_name).build(),
231        }
232    }
233
234    #[test]
235    fn test_group_spans_by_resource_and_scope_single_scope() {
236        let resource = Resource::new(vec![KeyValue::new("resource_key", "resource_value")]);
237        let span_data = create_test_span_data("lib1");
238
239        let spans = vec![span_data.clone()];
240        let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
241
242        let grouped_spans =
243            crate::transform::trace::tonic::group_spans_by_resource_and_scope(spans, &resource);
244
245        assert_eq!(grouped_spans.len(), 1);
246
247        let resource_spans = &grouped_spans[0];
248        assert_eq!(
249            resource_spans.resource.as_ref().unwrap().attributes.len(),
250            1
251        );
252        assert_eq!(
253            resource_spans.resource.as_ref().unwrap().attributes[0].key,
254            "resource_key"
255        );
256        assert_eq!(
257            resource_spans.resource.as_ref().unwrap().attributes[0]
258                .value
259                .clone()
260                .unwrap()
261                .value
262                .unwrap(),
263            Value::StringValue("resource_value".to_string())
264        );
265
266        let scope_spans = &resource_spans.scope_spans;
267        assert_eq!(scope_spans.len(), 1);
268
269        let scope_span = &scope_spans[0];
270        assert_eq!(scope_span.scope.as_ref().unwrap().name, "lib1");
271        assert_eq!(scope_span.spans.len(), 1);
272
273        assert_eq!(
274            scope_span.spans[0].trace_id,
275            span_data.span_context.trace_id().to_bytes().to_vec()
276        );
277    }
278
279    #[test]
280    fn test_group_spans_by_resource_and_scope_multiple_scopes() {
281        let resource = Resource::new(vec![KeyValue::new("resource_key", "resource_value")]);
282        let span_data1 = create_test_span_data("lib1");
283        let span_data2 = create_test_span_data("lib1");
284        let span_data3 = create_test_span_data("lib2");
285
286        let spans = vec![span_data1.clone(), span_data2.clone(), span_data3.clone()];
287        let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
288
289        let grouped_spans =
290            crate::transform::trace::tonic::group_spans_by_resource_and_scope(spans, &resource);
291
292        assert_eq!(grouped_spans.len(), 1);
293
294        let resource_spans = &grouped_spans[0];
295        assert_eq!(
296            resource_spans.resource.as_ref().unwrap().attributes.len(),
297            1
298        );
299        assert_eq!(
300            resource_spans.resource.as_ref().unwrap().attributes[0].key,
301            "resource_key"
302        );
303        assert_eq!(
304            resource_spans.resource.as_ref().unwrap().attributes[0]
305                .value
306                .clone()
307                .unwrap()
308                .value
309                .unwrap(),
310            Value::StringValue("resource_value".to_string())
311        );
312
313        let scope_spans = &resource_spans.scope_spans;
314        assert_eq!(scope_spans.len(), 2);
315
316        // Check the scope spans for both lib1 and lib2
317        let mut lib1_scope_span = None;
318        let mut lib2_scope_span = None;
319
320        for scope_span in scope_spans {
321            match scope_span.scope.as_ref().unwrap().name.as_str() {
322                "lib1" => lib1_scope_span = Some(scope_span),
323                "lib2" => lib2_scope_span = Some(scope_span),
324                _ => {}
325            }
326        }
327
328        let lib1_scope_span = lib1_scope_span.expect("lib1 scope span not found");
329        let lib2_scope_span = lib2_scope_span.expect("lib2 scope span not found");
330
331        assert_eq!(lib1_scope_span.scope.as_ref().unwrap().name, "lib1");
332        assert_eq!(lib2_scope_span.scope.as_ref().unwrap().name, "lib2");
333
334        assert_eq!(lib1_scope_span.spans.len(), 2);
335        assert_eq!(lib2_scope_span.spans.len(), 1);
336
337        assert_eq!(
338            lib1_scope_span.spans[0].trace_id,
339            span_data1.span_context.trace_id().to_bytes().to_vec()
340        );
341        assert_eq!(
342            lib1_scope_span.spans[1].trace_id,
343            span_data2.span_context.trace_id().to_bytes().to_vec()
344        );
345        assert_eq!(
346            lib2_scope_span.spans[0].trace_id,
347            span_data3.span_context.trace_id().to_bytes().to_vec()
348        );
349    }
350}
opentelemetry_proto/transform/trace.rs

opentelemetry_proto/transform/
trace.rs
