opentelemetry_proto/transform/
trace.rs

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::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::trace::SpanData> for Span {
49        fn from(source_span: opentelemetry_sdk::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_scope
105                        .schema_url()
106                        .map(ToOwned::to_owned)
107                        .unwrap_or_default(),
108                    scope: Some((source_span.instrumentation_scope, None).into()),
109                    spans: vec![Span {
110                        trace_id: source_span.span_context.trace_id().to_bytes().to_vec(),
111                        span_id: source_span.span_context.span_id().to_bytes().to_vec(),
112                        trace_state: source_span.span_context.trace_state().header(),
113                        parent_span_id: {
114                            if source_span.parent_span_id != SpanId::INVALID {
115                                source_span.parent_span_id.to_bytes().to_vec()
116                            } else {
117                                vec![]
118                            }
119                        },
120                        flags: source_span.span_context.trace_flags().to_u8() as u32,
121                        name: source_span.name.into_owned(),
122                        kind: span_kind as i32,
123                        start_time_unix_nano: to_nanos(source_span.start_time),
124                        end_time_unix_nano: to_nanos(source_span.end_time),
125                        dropped_attributes_count: source_span.dropped_attributes_count,
126                        attributes: Attributes::from(source_span.attributes).0,
127                        dropped_events_count: source_span.events.dropped_count,
128                        events: source_span
129                            .events
130                            .into_iter()
131                            .map(|event| span::Event {
132                                time_unix_nano: to_nanos(event.timestamp),
133                                name: event.name.into(),
134                                attributes: Attributes::from(event.attributes).0,
135                                dropped_attributes_count: event.dropped_attributes_count,
136                            })
137                            .collect(),
138                        dropped_links_count: source_span.links.dropped_count,
139                        links: source_span.links.into_iter().map(Into::into).collect(),
140                        status: Some(Status {
141                            code: status::StatusCode::from(&source_span.status).into(),
142                            message: match source_span.status {
143                                trace::Status::Error { description } => description.to_string(),
144                                _ => Default::default(),
145                            },
146                        }),
147                    }],
148                }],
149            }
150        }
151    }
152
153    pub fn group_spans_by_resource_and_scope(
154        spans: Vec<SpanData>,
155        resource: &ResourceAttributesWithSchema,
156    ) -> Vec<ResourceSpans> {
157        // Group spans by their instrumentation scope
158        let scope_map = spans.iter().fold(
159            HashMap::new(),
160            |mut scope_map: HashMap<&opentelemetry::InstrumentationScope, Vec<&SpanData>>, span| {
161                let instrumentation = &span.instrumentation_scope;
162                scope_map.entry(instrumentation).or_default().push(span);
163                scope_map
164            },
165        );
166
167        // Convert the grouped spans into ScopeSpans
168        let scope_spans = scope_map
169            .into_iter()
170            .map(|(instrumentation, span_records)| ScopeSpans {
171                scope: Some((instrumentation, None).into()),
172                schema_url: resource.schema_url.clone().unwrap_or_default(),
173                spans: span_records
174                    .into_iter()
175                    .map(|span_data| span_data.clone().into())
176                    .collect(),
177            })
178            .collect();
179
180        // Wrap ScopeSpans into a single ResourceSpans
181        vec![ResourceSpans {
182            resource: Some(Resource {
183                attributes: resource.attributes.0.clone(),
184                dropped_attributes_count: 0,
185            }),
186            scope_spans,
187            schema_url: resource.schema_url.clone().unwrap_or_default(),
188        }]
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use crate::tonic::common::v1::any_value::Value;
195    use crate::transform::common::tonic::ResourceAttributesWithSchema;
196    use opentelemetry::time::now;
197    use opentelemetry::trace::{
198        SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
199    };
200    use opentelemetry::InstrumentationScope;
201    use opentelemetry::KeyValue;
202    use opentelemetry_sdk::resource::Resource;
203    use opentelemetry_sdk::trace::SpanData;
204    use opentelemetry_sdk::trace::{SpanEvents, SpanLinks};
205    use std::borrow::Cow;
206    use std::time::Duration;
207
208    fn create_test_span_data(instrumentation_name: &'static str) -> SpanData {
209        let span_context = SpanContext::new(
210            TraceId::from_u128(123),
211            SpanId::from_u64(456),
212            TraceFlags::default(),
213            false,
214            TraceState::default(),
215        );
216
217        SpanData {
218            span_context,
219            parent_span_id: SpanId::from_u64(0),
220            span_kind: SpanKind::Internal,
221            name: Cow::Borrowed("test_span"),
222            start_time: now(),
223            end_time: now() + Duration::from_secs(1),
224            attributes: vec![KeyValue::new("key", "value")],
225            dropped_attributes_count: 0,
226            events: SpanEvents::default(),
227            links: SpanLinks::default(),
228            status: Status::Unset,
229            instrumentation_scope: InstrumentationScope::builder(instrumentation_name).build(),
230        }
231    }
232
233    #[test]
234    fn test_group_spans_by_resource_and_scope_single_scope() {
235        let resource = Resource::builder_empty()
236            .with_attribute(KeyValue::new("resource_key", "resource_value"))
237            .build();
238        let span_data = create_test_span_data("lib1");
239
240        let spans = vec![span_data.clone()];
241        let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
242
243        let grouped_spans =
244            crate::transform::trace::tonic::group_spans_by_resource_and_scope(spans, &resource);
245
246        assert_eq!(grouped_spans.len(), 1);
247
248        let resource_spans = &grouped_spans[0];
249        assert_eq!(
250            resource_spans.resource.as_ref().unwrap().attributes.len(),
251            1
252        );
253        assert_eq!(
254            resource_spans.resource.as_ref().unwrap().attributes[0].key,
255            "resource_key"
256        );
257        assert_eq!(
258            resource_spans.resource.as_ref().unwrap().attributes[0]
259                .value
260                .clone()
261                .unwrap()
262                .value
263                .unwrap(),
264            Value::StringValue("resource_value".to_string())
265        );
266
267        let scope_spans = &resource_spans.scope_spans;
268        assert_eq!(scope_spans.len(), 1);
269
270        let scope_span = &scope_spans[0];
271        assert_eq!(scope_span.scope.as_ref().unwrap().name, "lib1");
272        assert_eq!(scope_span.spans.len(), 1);
273
274        assert_eq!(
275            scope_span.spans[0].trace_id,
276            span_data.span_context.trace_id().to_bytes().to_vec()
277        );
278    }
279
280    #[test]
281    fn test_group_spans_by_resource_and_scope_multiple_scopes() {
282        let resource = Resource::builder_empty()
283            .with_attribute(KeyValue::new("resource_key", "resource_value"))
284            .build();
285        let span_data1 = create_test_span_data("lib1");
286        let span_data2 = create_test_span_data("lib1");
287        let span_data3 = create_test_span_data("lib2");
288
289        let spans = vec![span_data1.clone(), span_data2.clone(), span_data3.clone()];
290        let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
291
292        let grouped_spans =
293            crate::transform::trace::tonic::group_spans_by_resource_and_scope(spans, &resource);
294
295        assert_eq!(grouped_spans.len(), 1);
296
297        let resource_spans = &grouped_spans[0];
298        assert_eq!(
299            resource_spans.resource.as_ref().unwrap().attributes.len(),
300            1
301        );
302        assert_eq!(
303            resource_spans.resource.as_ref().unwrap().attributes[0].key,
304            "resource_key"
305        );
306        assert_eq!(
307            resource_spans.resource.as_ref().unwrap().attributes[0]
308                .value
309                .clone()
310                .unwrap()
311                .value
312                .unwrap(),
313            Value::StringValue("resource_value".to_string())
314        );
315
316        let scope_spans = &resource_spans.scope_spans;
317        assert_eq!(scope_spans.len(), 2);
318
319        // Check the scope spans for both lib1 and lib2
320        let mut lib1_scope_span = None;
321        let mut lib2_scope_span = None;
322
323        for scope_span in scope_spans {
324            match scope_span.scope.as_ref().unwrap().name.as_str() {
325                "lib1" => lib1_scope_span = Some(scope_span),
326                "lib2" => lib2_scope_span = Some(scope_span),
327                _ => {}
328            }
329        }
330
331        let lib1_scope_span = lib1_scope_span.expect("lib1 scope span not found");
332        let lib2_scope_span = lib2_scope_span.expect("lib2 scope span not found");
333
334        assert_eq!(lib1_scope_span.scope.as_ref().unwrap().name, "lib1");
335        assert_eq!(lib2_scope_span.scope.as_ref().unwrap().name, "lib2");
336
337        assert_eq!(lib1_scope_span.spans.len(), 2);
338        assert_eq!(lib2_scope_span.spans.len(), 1);
339
340        assert_eq!(
341            lib1_scope_span.spans[0].trace_id,
342            span_data1.span_context.trace_id().to_bytes().to_vec()
343        );
344        assert_eq!(
345            lib1_scope_span.spans[1].trace_id,
346            span_data2.span_context.trace_id().to_bytes().to_vec()
347        );
348        assert_eq!(
349            lib2_scope_span.spans[0].trace_id,
350            span_data3.span_context.trace_id().to_bytes().to_vec()
351        );
352    }
353}
opentelemetry_proto/transform/trace.rs

opentelemetry_proto/transform/
trace.rs
