servers/http/jaeger/v3/
otlp.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::collections::{HashMap, HashSet};
16
17use axum::http::StatusCode;
18use opentelemetry_proto::tonic::common::v1::{
19    AnyValue, ArrayValue, InstrumentationScope, KeyValue, KeyValueList, any_value,
20};
21use opentelemetry_proto::tonic::resource::v1::Resource;
22use opentelemetry_proto::tonic::trace::v1::span::{Event, Link, SpanKind};
23use opentelemetry_proto::tonic::trace::v1::status::StatusCode as SpanStatusCode;
24use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span, Status, TracesData};
25use serde_json::Value;
26
27use crate::http::HttpRecordsOutput;
28use crate::http::jaeger::v3::ApiError;
29use crate::otlp::trace::{
30    KEY_SERVICE_NAME, PARENT_SPAN_ID_COLUMN, RESOURCE_ATTRIBUTES_COLUMN, SCOPE_ATTRIBUTES_COLUMN,
31    SCOPE_NAME_COLUMN, SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN,
32    SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_LINKS_COLUMN, SPAN_NAME_COLUMN,
33    SPAN_STATUS_CODE, SPAN_STATUS_MESSAGE_COLUMN, TIMESTAMP_COLUMN, TIMESTAMP_END_COLUMN,
34    TRACE_ID_COLUMN, TRACE_STATE_COLUMN,
35};
36
37/// Converts stored trace rows directly to OTLP, without the legacy microsecond conversion.
38pub fn traces_from_records(records: HttpRecordsOutput) -> Result<TracesData, ApiError> {
39    let binary_columns: HashSet<_> = records
40        .schema
41        .column_schemas
42        .iter()
43        .filter(|column| column.data_type == "Binary")
44        .map(|column| column.name.as_str())
45        .collect();
46    let mut resource_spans = Vec::with_capacity(records.rows.len());
47    for row in &records.rows {
48        let cells: HashMap<_, _> = records
49            .schema
50            .column_schemas
51            .iter()
52            .zip(row)
53            .map(|(column, cell)| (column.name.as_str(), cell))
54            .collect();
55        let string = |name| cells.get(name).and_then(|v| v.as_str()).unwrap_or_default();
56        let timestamp = |name| {
57            cells
58                .get(name)
59                .and_then(|v| v.as_u64())
60                .ok_or_else(|| invalid_data(format!("Invalid timestamp column: {name}")))
61        };
62        let span = Span {
63            trace_id: decode_id(string(TRACE_ID_COLUMN), 16)?,
64            span_id: decode_id(string(SPAN_ID_COLUMN), 8)?,
65            parent_span_id: if string(PARENT_SPAN_ID_COLUMN).is_empty() {
66                vec![]
67            } else {
68                decode_id(string(PARENT_SPAN_ID_COLUMN), 8)?
69            },
70            trace_state: string(TRACE_STATE_COLUMN).into(),
71            name: string(SPAN_NAME_COLUMN).into(),
72            kind: SpanKind::from_str_name(string(SPAN_KIND_COLUMN)).unwrap_or(SpanKind::Unspecified)
73                as i32,
74            start_time_unix_nano: timestamp(TIMESTAMP_COLUMN)?,
75            end_time_unix_nano: timestamp(TIMESTAMP_END_COLUMN)?,
76            attributes: attributes(&cells, SPAN_ATTRIBUTES_COLUMN, &binary_columns)?,
77            events: events(cells.get(SPAN_EVENTS_COLUMN).copied())?,
78            links: links(cells.get(SPAN_LINKS_COLUMN).copied())?,
79            status: Some(Status {
80                code: SpanStatusCode::from_str_name(string(SPAN_STATUS_CODE))
81                    .unwrap_or(SpanStatusCode::Unset) as i32,
82                message: string(SPAN_STATUS_MESSAGE_COLUMN).into(),
83            }),
84            ..Default::default()
85        };
86        let mut resource_attributes =
87            attributes(&cells, RESOURCE_ATTRIBUTES_COLUMN, &binary_columns)?;
88        if !resource_attributes
89            .iter()
90            .any(|attr| attr.key == KEY_SERVICE_NAME)
91            && !string(SERVICE_NAME_COLUMN).is_empty()
92        {
93            resource_attributes.push(KeyValue {
94                key: KEY_SERVICE_NAME.into(),
95                value: Some(AnyValue {
96                    value: Some(any_value::Value::StringValue(
97                        string(SERVICE_NAME_COLUMN).into(),
98                    )),
99                }),
100                ..Default::default()
101            });
102        }
103        // Each row retains its own resource and scope. Service name alone does not identify
104        // a resource: different instances of one service can have different attributes.
105        resource_spans.push(ResourceSpans {
106            resource: Some(Resource {
107                attributes: resource_attributes,
108                ..Default::default()
109            }),
110            scope_spans: vec![ScopeSpans {
111                scope: Some(InstrumentationScope {
112                    name: string(SCOPE_NAME_COLUMN).into(),
113                    version: string(SCOPE_VERSION_COLUMN).into(),
114                    attributes: attributes(&cells, SCOPE_ATTRIBUTES_COLUMN, &binary_columns)?,
115                    ..Default::default()
116                }),
117                spans: vec![span],
118                ..Default::default()
119            }],
120            ..Default::default()
121        });
122    }
123    Ok(TracesData { resource_spans })
124}
125
126fn invalid_data(message: impl Into<String>) -> ApiError {
127    ApiError {
128        status: StatusCode::INTERNAL_SERVER_ERROR,
129        message: message.into(),
130    }
131}
132
133fn decode_id(value: &str, length: usize) -> Result<Vec<u8>, ApiError> {
134    let bytes = hex::decode(value).map_err(|_| invalid_data("Invalid stored span or trace ID"))?;
135    if bytes.len() != length {
136        return Err(invalid_data("Invalid stored span or trace ID length"));
137    }
138    Ok(bytes)
139}
140
141fn attributes(
142    cells: &HashMap<&str, &Value>,
143    name: &str,
144    binary_columns: &HashSet<&str>,
145) -> Result<Vec<KeyValue>, ApiError> {
146    if let Some(value) = cells.get(name) {
147        return object_attributes(value);
148    }
149    let prefix = format!("{name}.");
150    let mut attrs = cells
151        .iter()
152        .filter_map(|(column, value)| {
153            column
154                .strip_prefix(&prefix)
155                .filter(|_| !value.is_null())
156                .map(|key| (key, *value, binary_columns.contains(column)))
157        })
158        .map(|(key, value, is_binary)| {
159            let value = if is_binary {
160                AnyValue {
161                    value: Some(any_value::Value::BytesValue(
162                        serde_json::from_value(value.clone())
163                            .map_err(|_| invalid_data("Invalid stored binary attribute"))?,
164                    )),
165                }
166            } else {
167                any_value(value)?
168            };
169            Ok(KeyValue {
170                key: key.into(),
171                value: Some(value),
172                ..Default::default()
173            })
174        })
175        .collect::<Result<Vec<_>, ApiError>>()?;
176    attrs.sort_unstable_by(|a, b| a.key.cmp(&b.key));
177    Ok(attrs)
178}
179
180fn object_attributes(value: &Value) -> Result<Vec<KeyValue>, ApiError> {
181    match value {
182        Value::Null => Ok(vec![]),
183        Value::Object(values) => values
184            .iter()
185            .map(|(key, value)| {
186                Ok(KeyValue {
187                    key: key.clone(),
188                    value: Some(any_value(value)?),
189                    ..Default::default()
190                })
191            })
192            .collect(),
193        _ => Err(invalid_data("Stored attributes must be a JSON object")),
194    }
195}
196
197fn any_value(value: &Value) -> Result<AnyValue, ApiError> {
198    use any_value::Value as OtlpValue;
199
200    let value = match value {
201        Value::Null => None,
202        Value::String(value) => Some(OtlpValue::StringValue(value.clone())),
203        Value::Bool(value) => Some(OtlpValue::BoolValue(*value)),
204        Value::Number(value) => Some(if let Some(value) = value.as_i64() {
205            OtlpValue::IntValue(value)
206        } else {
207            OtlpValue::DoubleValue(
208                value
209                    .as_f64()
210                    .ok_or_else(|| invalid_data("Invalid numeric attribute"))?,
211            )
212        }),
213        Value::Array(values) => Some(OtlpValue::ArrayValue(ArrayValue {
214            values: values.iter().map(any_value).collect::<Result<_, _>>()?,
215        })),
216        Value::Object(_) => Some(OtlpValue::KvlistValue(KeyValueList {
217            values: object_attributes(value)?,
218        })),
219    };
220    Ok(AnyValue { value })
221}
222
223fn events(value: Option<&Value>) -> Result<Vec<Event>, ApiError> {
224    let Some(value) = value.filter(|value| !value.is_null()) else {
225        return Ok(vec![]);
226    };
227    let values = value
228        .as_array()
229        .ok_or_else(|| invalid_data("Invalid stored span events"))?;
230    values
231        .iter()
232        .map(|event| {
233            let time = event["time"]
234                .as_str()
235                .ok_or_else(|| invalid_data("Missing span event time"))?;
236            let time = chrono::DateTime::parse_from_str(time, "%Y-%m-%d %H:%M:%S%.f%z")
237                .ok()
238                .and_then(|time| time.timestamp_nanos_opt())
239                .and_then(|time| u64::try_from(time).ok())
240                .ok_or_else(|| invalid_data("Invalid stored span event time"))?;
241            Ok(Event {
242                time_unix_nano: time,
243                name: event["name"]
244                    .as_str()
245                    .ok_or_else(|| invalid_data("Missing span event name"))?
246                    .into(),
247                attributes: object_attributes(&event["attributes"])?,
248                ..Default::default()
249            })
250        })
251        .collect()
252}
253
254fn links(value: Option<&Value>) -> Result<Vec<Link>, ApiError> {
255    let Some(value) = value.filter(|value| !value.is_null()) else {
256        return Ok(vec![]);
257    };
258    let values = value
259        .as_array()
260        .ok_or_else(|| invalid_data("Invalid stored span links"))?;
261    values
262        .iter()
263        .map(|link| {
264            Ok(Link {
265                trace_id: decode_id(link["trace_id"].as_str().unwrap_or_default(), 16)?,
266                span_id: decode_id(link["span_id"].as_str().unwrap_or_default(), 8)?,
267                trace_state: link["trace_state"].as_str().unwrap_or_default().into(),
268                attributes: object_attributes(&link["attributes"])?,
269                ..Default::default()
270            })
271        })
272        .collect()
273}
274
275#[cfg(test)]
276mod tests {
277    use serde_json::json;
278
279    use super::*;
280
281    #[test]
282    fn test_event_timestamp_precision() {
283        for (time, expected) in [
284            ("1970-01-01 00:00:01+0000", 1_000_000_000),
285            ("1970-01-01 00:00:01.123+0000", 1_123_000_000),
286            ("1970-01-01 00:00:01.123456+0000", 1_123_456_000),
287            ("1970-01-01 00:00:01.123456789+0000", 1_123_456_789),
288        ] {
289            let value = json!([{"name":"event", "time":time, "attributes":{}}]);
290            let events = events(Some(&value)).unwrap();
291            assert_eq!(events[0].time_unix_nano, expected);
292        }
293    }
294}
servers/http/jaeger/v3/otlp.rs

servers/http/jaeger/v3/
otlp.rs
