servers/otlp/trace/
span.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt::Display;
16
17use common_time::timestamp::Timestamp;
18use itertools::Itertools;
19use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
20use opentelemetry_proto::tonic::common::v1::InstrumentationScope;
21use opentelemetry_proto::tonic::trace::v1::span::{Event, Link};
22use opentelemetry_proto::tonic::trace::v1::{Span, Status};
23use serde::Serialize;
24
25use crate::otlp::trace::attributes::{Attributes, SharedAttributes};
26use crate::otlp::utils::{bytes_to_hex_string, resource_service_name};
27
28#[derive(Debug, Clone)]
29pub struct TraceSpan {
30    // the following are tags
31    pub service_name: Option<String>,
32    pub trace_id: String,
33    pub span_id: String,
34    pub parent_span_id: Option<String>,
35
36    // the following are fields
37    pub resource_attributes: SharedAttributes,
38    pub scope_name: String,
39    pub scope_version: String,
40    pub scope_attributes: SharedAttributes,
41    pub trace_state: String,
42    pub span_name: String,
43    pub span_kind: String,
44    pub span_status_code: String,
45    pub span_status_message: String,
46    pub span_attributes: Attributes,
47    pub span_events: SpanEvents,  // TODO(yuanbohan): List in the future
48    pub span_links: SpanLinks,    // TODO(yuanbohan): List in the future
49    pub start_in_nanosecond: u64, // this is also the Timestamp Index
50    pub end_in_nanosecond: u64,
51}
52
53pub type TraceSpans = Vec<TraceSpan>;
54
55#[derive(Debug, Clone)]
56pub struct TraceSpanGroup {
57    pub service_name: Option<String>,
58    pub resource_attributes: SharedAttributes,
59    pub scope_name: String,
60    pub scope_version: String,
61    pub scope_attributes: SharedAttributes,
62    pub spans: TraceSpans,
63}
64
65pub type TraceSpanGroups = Vec<TraceSpanGroup>;
66
67#[derive(Debug, Clone, Serialize)]
68pub struct SpanLink {
69    pub trace_id: String,
70    pub span_id: String,
71    pub trace_state: String,
72    pub attributes: Attributes, // TODO(yuanbohan): Map in the future
73}
74
75impl From<Link> for SpanLink {
76    fn from(link: Link) -> Self {
77        Self {
78            trace_id: bytes_to_hex_string(&link.trace_id),
79            span_id: bytes_to_hex_string(&link.span_id),
80            trace_state: link.trace_state,
81            attributes: Attributes::from(link.attributes),
82        }
83    }
84}
85
86impl From<SpanLink> for jsonb::Value<'static> {
87    fn from(value: SpanLink) -> jsonb::Value<'static> {
88        jsonb::Value::Object(
89            vec![
90                (
91                    "trace_id".to_string(),
92                    jsonb::Value::String(value.trace_id.into()),
93                ),
94                (
95                    "span_id".to_string(),
96                    jsonb::Value::String(value.span_id.into()),
97                ),
98                (
99                    "trace_state".to_string(),
100                    jsonb::Value::String(value.trace_state.into()),
101                ),
102                ("attributes".to_string(), value.attributes.into()),
103            ]
104            .into_iter()
105            .collect(),
106        )
107    }
108}
109
110#[derive(Debug, Clone, Serialize)]
111pub struct SpanLinks(Vec<SpanLink>);
112
113impl From<Vec<Link>> for SpanLinks {
114    fn from(value: Vec<Link>) -> Self {
115        let links = value.into_iter().map(SpanLink::from).collect_vec();
116        Self(links)
117    }
118}
119
120impl From<SpanLinks> for jsonb::Value<'static> {
121    fn from(value: SpanLinks) -> jsonb::Value<'static> {
122        jsonb::Value::Array(value.0.into_iter().map(Into::into).collect())
123    }
124}
125
126impl Display for SpanLinks {
127    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
128        write!(f, "{}", serde_json::to_string(self).unwrap_or_default())
129    }
130}
131
132impl SpanLinks {
133    pub fn get_ref(&self) -> &Vec<SpanLink> {
134        &self.0
135    }
136
137    pub fn get_mut(&mut self) -> &mut Vec<SpanLink> {
138        &mut self.0
139    }
140}
141
142#[derive(Debug, Clone, Serialize)]
143pub struct SpanEvent {
144    pub name: String,
145    pub time: String,
146    pub attributes: Attributes, // TODO(yuanbohan): Map in the future
147}
148
149impl From<Event> for SpanEvent {
150    fn from(event: Event) -> Self {
151        Self {
152            name: event.name,
153            time: Timestamp::new_nanosecond(event.time_unix_nano as i64).to_iso8601_string(),
154            attributes: Attributes::from(event.attributes),
155        }
156    }
157}
158
159impl From<SpanEvent> for jsonb::Value<'static> {
160    fn from(value: SpanEvent) -> jsonb::Value<'static> {
161        jsonb::Value::Object(
162            vec![
163                ("name".to_string(), jsonb::Value::String(value.name.into())),
164                ("time".to_string(), jsonb::Value::String(value.time.into())),
165                ("attributes".to_string(), value.attributes.into()),
166            ]
167            .into_iter()
168            .collect(),
169        )
170    }
171}
172
173#[derive(Debug, Clone, Serialize)]
174pub struct SpanEvents(Vec<SpanEvent>);
175
176impl From<Vec<Event>> for SpanEvents {
177    fn from(value: Vec<Event>) -> Self {
178        let events = value.into_iter().map(SpanEvent::from).collect_vec();
179        Self(events)
180    }
181}
182
183impl From<SpanEvents> for jsonb::Value<'static> {
184    fn from(value: SpanEvents) -> jsonb::Value<'static> {
185        jsonb::Value::Array(value.0.into_iter().map(Into::into).collect())
186    }
187}
188
189impl Display for SpanEvents {
190    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
191        write!(f, "{}", serde_json::to_string(self).unwrap_or_default())
192    }
193}
194
195impl SpanEvents {
196    pub fn get_ref(&self) -> &Vec<SpanEvent> {
197        &self.0
198    }
199
200    pub fn get_mut(&mut self) -> &mut Vec<SpanEvent> {
201        &mut self.0
202    }
203}
204
205pub fn parse_span(
206    service_name: Option<String>,
207    resource_attrs: &SharedAttributes,
208    scope_name: &str,
209    scope_version: &str,
210    scope_attrs: &SharedAttributes,
211    span: Span,
212) -> TraceSpan {
213    let (span_status_code, span_status_message) = status_to_string(&span.status);
214    let span_kind = span.kind().as_str_name().into();
215    TraceSpan {
216        service_name,
217        trace_id: bytes_to_hex_string(&span.trace_id),
218        span_id: bytes_to_hex_string(&span.span_id),
219        parent_span_id: if span.parent_span_id.is_empty() {
220            None
221        } else {
222            Some(bytes_to_hex_string(&span.parent_span_id))
223        },
224
225        resource_attributes: resource_attrs.clone(),
226        trace_state: span.trace_state,
227
228        scope_name: scope_name.to_string(),
229        scope_version: scope_version.to_string(),
230        scope_attributes: scope_attrs.clone(),
231
232        span_name: span.name,
233        span_kind,
234        span_status_code,
235        span_status_message,
236        span_attributes: Attributes::from(span.attributes),
237        span_events: SpanEvents::from(span.events),
238        span_links: SpanLinks::from(span.links),
239
240        start_in_nanosecond: span.start_time_unix_nano,
241        end_in_nanosecond: span.end_time_unix_nano,
242    }
243}
244
245pub fn status_to_string(status: &Option<Status>) -> (String, String) {
246    match status {
247        Some(status) => (status.code().as_str_name().into(), status.message.clone()),
248        None => ("".into(), "".into()),
249    }
250}
251
252/// Convert OpenTelemetry traces to SpanTraces
253///
254/// See
255/// <https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto>
256/// for data structure of OTLP traces.
257pub fn parse(request: ExportTraceServiceRequest) -> TraceSpanGroups {
258    let group_size = request
259        .resource_spans
260        .iter()
261        .flat_map(|res| res.scope_spans.iter())
262        .count();
263    let mut groups = Vec::with_capacity(group_size);
264    for resource_spans in request.resource_spans {
265        let resource_attrs = resource_spans
266            .resource
267            .map(|r| r.attributes)
268            .unwrap_or_default();
269
270        let service_name = resource_service_name(&resource_attrs);
271
272        let resource_attrs = SharedAttributes::from(Attributes::from(resource_attrs));
273        for scope_spans in resource_spans.scope_spans {
274            let InstrumentationScope {
275                name: scope_name,
276                version: scope_version,
277                attributes: scope_attrs,
278                ..
279            } = scope_spans.scope.unwrap_or_default();
280            let scope_attrs = SharedAttributes::from(Attributes::from(scope_attrs));
281            let mut spans = Vec::with_capacity(scope_spans.spans.len());
282            for span in scope_spans.spans {
283                spans.push(parse_span(
284                    service_name.clone(),
285                    &resource_attrs,
286                    &scope_name,
287                    &scope_version,
288                    &scope_attrs,
289                    span,
290                ));
291            }
292            groups.push(TraceSpanGroup {
293                service_name: service_name.clone(),
294                resource_attributes: resource_attrs.clone(),
295                scope_name,
296                scope_version,
297                scope_attributes: scope_attrs,
298                spans,
299            });
300        }
301    }
302    groups
303}
304
305#[cfg(test)]
306mod tests {
307    use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
308    use opentelemetry_proto::tonic::common::v1::{
309        AnyValue, InstrumentationScope, KeyValue, any_value,
310    };
311    use opentelemetry_proto::tonic::resource::v1::Resource;
312    use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span, Status};
313
314    use crate::otlp::trace::KEY_SERVICE_NAME;
315    use crate::otlp::trace::span::{bytes_to_hex_string, parse, status_to_string};
316
317    fn make_kv(key: &str, value: &str) -> KeyValue {
318        KeyValue {
319            key: key.to_string(),
320            value: Some(AnyValue {
321                value: Some(any_value::Value::StringValue(value.to_string())),
322            }),
323            ..Default::default()
324        }
325    }
326
327    fn make_span(trace_id: u8, span_id: u8) -> Span {
328        Span {
329            trace_id: vec![trace_id; 16],
330            span_id: vec![span_id; 8],
331            ..Default::default()
332        }
333    }
334
335    #[test]
336    fn test_bytes_to_hex_string() {
337        assert_eq!(
338            "24fe79948641b110a29bc27859307e8d",
339            bytes_to_hex_string(&[
340                36, 254, 121, 148, 134, 65, 177, 16, 162, 155, 194, 120, 89, 48, 126, 141,
341            ])
342        );
343
344        assert_eq!(
345            "baffeedd7b8debc0",
346            bytes_to_hex_string(&[186, 255, 238, 221, 123, 141, 235, 192,])
347        );
348    }
349
350    #[test]
351    fn test_status_to_string() {
352        let message = String::from("status message");
353        let status = Status {
354            code: 1,
355            message: message.clone(),
356        };
357
358        assert_eq!(
359            ("STATUS_CODE_OK".into(), message),
360            status_to_string(&Some(status)),
361        );
362    }
363
364    #[test]
365    fn test_parse_preserves_resource_scope_groups() {
366        let request = ExportTraceServiceRequest {
367            resource_spans: vec![
368                ResourceSpans {
369                    resource: Some(Resource {
370                        attributes: vec![make_kv(KEY_SERVICE_NAME, "svc-a")],
371                        ..Default::default()
372                    }),
373                    scope_spans: vec![
374                        ScopeSpans {
375                            scope: Some(InstrumentationScope {
376                                name: "scope-1".to_string(),
377                                attributes: vec![make_kv("scope.key", "scope-1-value")],
378                                ..Default::default()
379                            }),
380                            spans: vec![make_span(0x11, 0x21), make_span(0x12, 0x22)],
381                            ..Default::default()
382                        },
383                        ScopeSpans {
384                            scope: Some(InstrumentationScope {
385                                name: "scope-2".to_string(),
386                                attributes: vec![make_kv("scope.key", "scope-2-value")],
387                                ..Default::default()
388                            }),
389                            spans: vec![make_span(0x13, 0x23)],
390                            ..Default::default()
391                        },
392                    ],
393                    ..Default::default()
394                },
395                ResourceSpans {
396                    resource: Some(Resource {
397                        attributes: vec![make_kv(KEY_SERVICE_NAME, "svc-b")],
398                        ..Default::default()
399                    }),
400                    scope_spans: vec![ScopeSpans {
401                        scope: Some(InstrumentationScope {
402                            name: "scope-3".to_string(),
403                            ..Default::default()
404                        }),
405                        spans: vec![make_span(0x14, 0x24)],
406                        ..Default::default()
407                    }],
408                    ..Default::default()
409                },
410            ],
411        };
412
413        let groups = parse(request);
414        assert_eq!(groups.len(), 3);
415        assert_eq!(groups[0].service_name.as_deref(), Some("svc-a"));
416        assert_eq!(groups[0].scope_name, "scope-1");
417        assert_eq!(groups[0].spans.len(), 2);
418        assert_eq!(groups[1].scope_name, "scope-2");
419        assert_eq!(groups[1].spans.len(), 1);
420        assert_eq!(groups[2].service_name.as_deref(), Some("svc-b"));
421        assert_eq!(groups[2].scope_name, "scope-3");
422        assert_eq!(
423            groups[0].scope_attributes.as_ref().get_ref(),
424            &[make_kv("scope.key", "scope-1-value")]
425        );
426
427        // Scopes of one resource share a single copy of the resource attributes,
428        // spans share their group's attributes, and unrelated resources and
429        // scopes stay separate.
430        assert!(std::ptr::eq(
431            groups[0].resource_attributes.as_ref(),
432            groups[1].resource_attributes.as_ref(),
433        ));
434        assert!(!std::ptr::eq(
435            groups[0].resource_attributes.as_ref(),
436            groups[2].resource_attributes.as_ref(),
437        ));
438        assert!(!std::ptr::eq(
439            groups[0].scope_attributes.as_ref(),
440            groups[1].scope_attributes.as_ref(),
441        ));
442        for group in &groups {
443            for span in &group.spans {
444                assert!(std::ptr::eq(
445                    group.resource_attributes.as_ref(),
446                    span.resource_attributes.as_ref()
447                ));
448                assert!(std::ptr::eq(
449                    group.scope_attributes.as_ref(),
450                    span.scope_attributes.as_ref()
451                ));
452            }
453        }
454    }
455}
servers/otlp/trace/span.rs

servers/otlp/trace/
span.rs
