servers/http/jaeger/
v3.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
15mod otlp;
16mod summary;
17
18use std::collections::HashMap;
19use std::sync::Arc;
20
21use axum::extract::rejection::QueryRejection;
22use axum::extract::{Path, Query, State};
23use axum::http::StatusCode;
24use axum::response::{IntoResponse, Response};
25use axum::{Extension, Json};
26use common_error::ext::ErrorExt;
27use common_error::status_code::StatusCode as GreptimeStatusCode;
28use common_query::Output;
29use common_telemetry::error;
30use rust_decimal::Decimal;
31use rust_decimal::prelude::ToPrimitive;
32use serde::Deserialize;
33use serde_json::{Value, json};
34use session::context::QueryContext;
35
36use crate::error::{Error, status_code_to_http_status};
37use crate::http::extractor::TraceTableName;
38use crate::http::jaeger::{
39    QueryTraceParams, convert_string_to_boolean, convert_string_to_number, covert_to_records,
40    empty_string_as_none, operations_from_records, services_from_records, update_query_context,
41};
42use crate::metrics::METRIC_JAEGER_QUERY_ELAPSED;
43use crate::query_handler::JaegerQueryHandlerRef;
44
45/// Jaeger API v3 uses an error envelope rather than the legacy `errors` array.
46#[derive(Debug)]
47pub struct ApiError {
48    status: StatusCode,
49    message: String,
50}
51
52impl ApiError {
53    fn invalid(message: impl Into<String>) -> Self {
54        Self {
55            status: StatusCode::BAD_REQUEST,
56            message: message.into(),
57        }
58    }
59
60    fn not_found() -> Self {
61        Self {
62            status: StatusCode::NOT_FOUND,
63            message: "No traces found".into(),
64        }
65    }
66}
67
68impl From<Error> for ApiError {
69    fn from(err: Error) -> Self {
70        error!(err; "Jaeger v3 query failed");
71        Self {
72            status: status_code_to_http_status(&err.status_code()),
73            message: err.output_msg(),
74        }
75    }
76}
77
78impl From<QueryRejection> for ApiError {
79    fn from(err: QueryRejection) -> Self {
80        Self::invalid(err.body_text())
81    }
82}
83
84impl IntoResponse for ApiError {
85    fn into_response(self) -> Response {
86        (
87            self.status,
88            Json(json!({"error": {"httpCode": self.status.as_u16(), "message": self.message}})),
89        )
90            .into_response()
91    }
92}
93
94#[derive(Debug, Default, Deserialize)]
95pub struct OperationsParams {
96    service: String,
97    #[serde(rename = "spanKind", alias = "span_kind")]
98    span_kind: Option<String>,
99}
100
101#[derive(Debug, Default, Deserialize)]
102pub struct GetTraceParams {
103    #[serde(rename = "startTime", alias = "start_time")]
104    #[serde(default, deserialize_with = "empty_string_as_none")]
105    start_time: Option<String>,
106    #[serde(rename = "endTime", alias = "end_time")]
107    #[serde(default, deserialize_with = "empty_string_as_none")]
108    end_time: Option<String>,
109    // GreptimeDB returns stored spans without Jaeger's clock-skew enrichment.
110    #[serde(rename = "rawTraces", alias = "raw_traces")]
111    #[serde(default, deserialize_with = "empty_string_as_none")]
112    _raw_traces: Option<bool>,
113}
114
115#[derive(Debug, Default, Deserialize)]
116pub struct FindTracesParams {
117    #[serde(rename = "query.serviceName", alias = "query.service_name")]
118    service_name: Option<String>,
119    #[serde(rename = "query.operationName", alias = "query.operation_name")]
120    operation_name: Option<String>,
121    #[serde(rename = "query.startTimeMin", alias = "query.start_time_min")]
122    start_time_min: String,
123    #[serde(rename = "query.startTimeMax", alias = "query.start_time_max")]
124    start_time_max: String,
125    #[serde(rename = "query.durationMin", alias = "query.duration_min")]
126    #[serde(default, deserialize_with = "empty_string_as_none")]
127    duration_min: Option<String>,
128    #[serde(rename = "query.durationMax", alias = "query.duration_max")]
129    #[serde(default, deserialize_with = "empty_string_as_none")]
130    duration_max: Option<String>,
131    #[serde(
132        rename = "query.searchDepth",
133        alias = "query.search_depth",
134        alias = "query.num_traces"
135    )]
136    #[serde(default, deserialize_with = "empty_string_as_none")]
137    search_depth: Option<i32>,
138    #[serde(rename = "query.attributes")]
139    #[serde(default, deserialize_with = "empty_string_as_none")]
140    attributes: Option<String>,
141    #[serde(rename = "query.rawTraces", alias = "query.raw_traces")]
142    #[serde(default, deserialize_with = "empty_string_as_none")]
143    _raw_traces: Option<bool>,
144    #[serde(rename = "query.filter")]
145    filter: Option<String>,
146}
147
148impl FindTracesParams {
149    fn into_query(self) -> Result<QueryTraceParams, ApiError> {
150        if self.filter.is_some() {
151            return Err(ApiError::invalid("query.filter is not supported"));
152        }
153        let start = parse_timestamp(&self.start_time_min)?;
154        let end = parse_timestamp(&self.start_time_max)?;
155        if start >= end {
156            return Err(ApiError::invalid(
157                "query.startTimeMin must be before query.startTimeMax",
158            ));
159        }
160        let limit = match self.search_depth.unwrap_or(0) {
161            0 => 100,
162            limit @ 1..=10_000 => limit,
163            _ => {
164                return Err(ApiError::invalid(
165                    "query.searchDepth must be between 0 and 10000",
166                ));
167            }
168        };
169        let tags = self
170            .attributes
171            .map(|attributes| {
172                let values = serde_json::from_str::<HashMap<String, String>>(&attributes)
173                    .map_err(|err| ApiError::invalid(format!("Invalid query.attributes: {err}")))?;
174                Ok::<_, ApiError>(
175                    values
176                        .into_iter()
177                        .map(|(key, value)| {
178                            let value = Value::String(value);
179                            let value = convert_string_to_number(&value)
180                                .or_else(|| convert_string_to_boolean(&value))
181                                .unwrap_or(value);
182                            (key, value)
183                        })
184                        .collect(),
185                )
186            })
187            .transpose()?;
188        let query = QueryTraceParams {
189            service_name: self.service_name.filter(|name| !name.is_empty()),
190            operation_name: self.operation_name.filter(|name| !name.is_empty()),
191            start_time: Some(start),
192            // The backend uses an inclusive bound; API v3 specifies an exclusive bound.
193            end_time: Some(end - 1),
194            min_duration: self
195                .duration_min
196                .as_deref()
197                .map(parse_duration)
198                .transpose()?,
199            max_duration: self
200                .duration_max
201                .as_deref()
202                .map(parse_duration)
203                .transpose()?,
204            limit: Some(limit as usize),
205            tags,
206            fetch_full_trace: true,
207            ..Default::default()
208        };
209        if let (Some(min), Some(max)) = (query.min_duration, query.max_duration)
210            && min > max
211        {
212            return Err(ApiError::invalid(
213                "query.durationMin must not exceed query.durationMax",
214            ));
215        }
216        Ok(query)
217    }
218}
219
220fn parse_timestamp(value: &str) -> Result<i64, ApiError> {
221    chrono::DateTime::parse_from_rfc3339(value)
222        .map_err(|err| ApiError::invalid(format!("Invalid RFC3339 timestamp '{value}': {err}")))?
223        .timestamp_nanos_opt()
224        .ok_or_else(|| ApiError::invalid("Timestamp is outside the supported nanosecond range"))
225}
226
227fn parse_duration(value: &str) -> Result<u64, ApiError> {
228    let invalid = || ApiError::invalid(format!("Invalid Go duration '{value}'"));
229    let mut rest = value.strip_prefix('+').unwrap_or(value);
230    if rest == "0" {
231        return Ok(0);
232    }
233    if rest.is_empty() {
234        return Err(invalid());
235    }
236    let mut total = 0_u64;
237    while !rest.is_empty() {
238        let number_end = rest
239            .find(|ch: char| !ch.is_ascii_digit() && ch != '.')
240            .ok_or_else(invalid)?;
241        let amount: Decimal = rest[..number_end].parse().map_err(|_| invalid())?;
242        rest = &rest[number_end..];
243        let unit_end = rest
244            .find(|ch: char| ch.is_ascii_digit() || ch == '.')
245            .unwrap_or(rest.len());
246        let unit = match &rest[..unit_end] {
247            "ns" => 1_u64,
248            "us" | "µs" | "μs" => 1_000,
249            "ms" => 1_000_000,
250            "s" => 1_000_000_000,
251            "m" => 60_000_000_000,
252            "h" => 3_600_000_000_000,
253            _ => return Err(invalid()),
254        };
255        let nanos = amount
256            .checked_mul(Decimal::from(unit))
257            .and_then(|amount| amount.trunc().to_u64())
258            .ok_or_else(invalid)?;
259        total = total
260            .checked_add(nanos)
261            .filter(|total| *total <= i64::MAX as u64)
262            .ok_or_else(invalid)?;
263        rest = &rest[unit_end..];
264    }
265    Ok(total)
266}
267
268pub async fn handle_get_services(
269    State(handler): State<JaegerQueryHandlerRef>,
270    Extension(mut ctx): Extension<QueryContext>,
271    TraceTableName(table): TraceTableName,
272) -> Result<Json<Value>, ApiError> {
273    update_query_context(&mut ctx, table);
274    let _timer = METRIC_JAEGER_QUERY_ELAPSED
275        .with_label_values(&[&ctx.get_db_string(), "/api/v3/services"])
276        .start_timer();
277    let services = match query_records(handler.get_services(Arc::new(ctx)).await).await? {
278        Some(records) => services_from_records(records)?,
279        None => vec![],
280    };
281    Ok(Json(json!({"services": services})))
282}
283
284pub async fn handle_get_operations(
285    State(handler): State<JaegerQueryHandlerRef>,
286    params: Result<Query<OperationsParams>, QueryRejection>,
287    Extension(mut ctx): Extension<QueryContext>,
288    TraceTableName(table): TraceTableName,
289) -> Result<Json<Value>, ApiError> {
290    let Query(params) = params?;
291    if params.service.is_empty() {
292        return Err(ApiError::invalid("service is required"));
293    }
294    update_query_context(&mut ctx, table);
295    let _timer = METRIC_JAEGER_QUERY_ELAPSED
296        .with_label_values(&[&ctx.get_db_string(), "/api/v3/operations"])
297        .start_timer();
298    let operations = match query_records(
299        handler
300            .get_operations(
301                Arc::new(ctx),
302                &params.service,
303                params.span_kind.as_deref().filter(|kind| !kind.is_empty()),
304            )
305            .await,
306    )
307    .await?
308    {
309        Some(records) => operations_from_records(records, true)?,
310        None => vec![],
311    };
312    let operations: Vec<_> = operations.into_iter().map(|operation| json!({
313        "name": operation.name,
314        "spanKind": operation.span_kind.filter(|kind| !kind.is_empty()).unwrap_or_else(|| "internal".into()),
315    })).collect();
316    Ok(Json(json!({"operations": operations})))
317}
318
319pub async fn handle_get_trace(
320    State(handler): State<JaegerQueryHandlerRef>,
321    Path(trace_id): Path<String>,
322    params: Result<Query<GetTraceParams>, QueryRejection>,
323    Extension(mut ctx): Extension<QueryContext>,
324    TraceTableName(table): TraceTableName,
325) -> Result<Json<Value>, ApiError> {
326    let Query(params) = params?;
327    if !matches!(trace_id.len(), 16 | 32) || !trace_id.bytes().all(|ch| ch.is_ascii_hexdigit()) {
328        return Err(ApiError::invalid(
329            "trace_id must be a 64-bit or 128-bit hexadecimal ID",
330        ));
331    }
332    let trace_id = format!("{:0>32}", trace_id.to_ascii_lowercase());
333    let start = params
334        .start_time
335        .as_deref()
336        .map(parse_timestamp)
337        .transpose()?;
338    let end = params
339        .end_time
340        .as_deref()
341        .map(parse_timestamp)
342        .transpose()?;
343    if let (Some(start), Some(end)) = (start, end)
344        && start > end
345    {
346        return Err(ApiError::invalid("startTime must not exceed endTime"));
347    }
348    update_query_context(&mut ctx, table);
349    let _timer = METRIC_JAEGER_QUERY_ELAPSED
350        .with_label_values(&[&ctx.get_db_string(), "/api/v3/traces/{trace_id}"])
351        .start_timer();
352    // API v3 time bounds are lookup hints. Applying them as span filters would truncate the trace,
353    // so they are only validated.
354    trace_response(
355        handler
356            .get_trace(Arc::new(ctx), &trace_id, None, None, None)
357            .await,
358    )
359    .await
360}
361
362pub async fn handle_find_traces(
363    State(handler): State<JaegerQueryHandlerRef>,
364    params: Result<Query<FindTracesParams>, QueryRejection>,
365    Extension(mut ctx): Extension<QueryContext>,
366    TraceTableName(table): TraceTableName,
367) -> Result<Json<Value>, ApiError> {
368    let Query(params) = params?;
369    let query = params.into_query()?;
370    update_query_context(&mut ctx, table);
371    let _timer = METRIC_JAEGER_QUERY_ELAPSED
372        .with_label_values(&[&ctx.get_db_string(), "/api/v3/traces"])
373        .start_timer();
374    trace_response(handler.find_traces(Arc::new(ctx), query).await).await
375}
376
377pub async fn handle_find_trace_summaries(
378    State(handler): State<JaegerQueryHandlerRef>,
379    params: Result<Query<FindTracesParams>, QueryRejection>,
380    Extension(mut ctx): Extension<QueryContext>,
381    TraceTableName(table): TraceTableName,
382) -> Result<Json<Value>, ApiError> {
383    let Query(params) = params?;
384    let query = params.into_query()?;
385    update_query_context(&mut ctx, table);
386    let _timer = METRIC_JAEGER_QUERY_ELAPSED
387        .with_label_values(&[&ctx.get_db_string(), "/api/v3/trace-summaries"])
388        .start_timer();
389    let summaries = match query_records(handler.find_traces(Arc::new(ctx), query).await).await? {
390        Some(records) => summary::summarize(otlp::traces_from_records(records)?),
391        None => vec![],
392    };
393    Ok(Json(json!({"summaries": summaries})))
394}
395
396async fn query_records(
397    output: crate::error::Result<Output>,
398) -> Result<Option<crate::http::HttpRecordsOutput>, ApiError> {
399    match output {
400        Ok(output) => covert_to_records(output)
401            .await?
402            .map(Some)
403            .ok_or_else(|| ApiError {
404                status: StatusCode::INTERNAL_SERVER_ERROR,
405                message: "Unexpected Jaeger query output".into(),
406            }),
407        Err(err) if err.status_code() == GreptimeStatusCode::TableNotFound => Ok(None),
408        Err(err) => Err(err.into()),
409    }
410}
411
412async fn trace_response(output: crate::error::Result<Output>) -> Result<Json<Value>, ApiError> {
413    let records = query_records(output)
414        .await?
415        .ok_or_else(ApiError::not_found)?;
416    if records.rows.is_empty() {
417        return Err(ApiError::not_found());
418    }
419    Ok(Json(json!({"result": otlp::traces_from_records(records)?})))
420}
421
422#[cfg(test)]
423mod tests {
424    use super::*;
425
426    #[test]
427    fn test_find_traces_parameters() {
428        for (start, end, service, operation, depth, duration) in [
429            (
430                "start_time_min",
431                "start_time_max",
432                "service_name",
433                "operation_name",
434                "num_traces",
435                "duration_min",
436            ),
437            (
438                "startTimeMin",
439                "startTimeMax",
440                "serviceName",
441                "operationName",
442                "searchDepth",
443                "durationMin",
444            ),
445        ] {
446            let uri = format!("/api/v3/traces?query.{start}=2026-01-01T00:00:00.000000001Z&query.{end}=2026-01-01T00:00:01Z&query.{service}=checkout&query.{operation}=GET&query.{depth}=7&query.{duration}=1.5ms&query.attributes=%7B%22error%22%3A%22true%22%7D").parse().unwrap();
447            let Query(params) = Query::<FindTracesParams>::try_from_uri(&uri).unwrap();
448            let query = params.into_query().unwrap();
449            assert_eq!(query.service_name.as_deref(), Some("checkout"));
450            assert_eq!(query.operation_name.as_deref(), Some("GET"));
451            assert_eq!(query.limit, Some(7));
452            assert_eq!(query.min_duration, Some(1_500_000));
453            assert_eq!(query.start_time, Some(1767225600000000001));
454            assert_eq!(query.end_time, Some(1767225600999999999));
455            assert_eq!(query.tags.unwrap()["error"], json!(true));
456            assert!(query.fetch_full_trace);
457        }
458        let uri = "/api/v3/traces?query.startTimeMin=2026-01-01T00:00:00Z&query.startTimeMax=2026-01-01T01:00:00Z&query.durationMin=&query.durationMax=&query.searchDepth=&query.rawTraces=&query.attributes=".parse().unwrap();
459        let Query(params) = Query::<FindTracesParams>::try_from_uri(&uri).unwrap();
460        let query = params.into_query().unwrap();
461        assert_eq!(query.service_name, None);
462        assert_eq!(query.limit, Some(100));
463        for name in [
464            "query.searchDepth",
465            "query.search_depth",
466            "query.num_traces",
467        ] {
468            for (depth, expected) in [(0, 100), (1, 1), (10_000, 10_000)] {
469                let uri = format!("/api/v3/traces?query.startTimeMin=2026-01-01T00:00:00Z&query.startTimeMax=2026-01-01T01:00:00Z&{name}={depth}").parse().unwrap();
470                let Query(params) = Query::<FindTracesParams>::try_from_uri(&uri).unwrap();
471                assert_eq!(params.into_query().unwrap().limit, Some(expected));
472            }
473        }
474    }
475
476    #[test]
477    fn test_go_durations() {
478        for (value, expected) in [
479            ("0", 0),
480            ("1.5ms", 1_500_000),
481            ("1h2m3.5s", 3_723_500_000_000),
482            ("+.5us", 500),
483            ("1.µs", 1_000),
484            ("1μs", 1_000),
485            ("0.1ns", 0),
486            ("9223372036854775807ns", i64::MAX as u64),
487        ] {
488            assert_eq!(parse_duration(value).unwrap(), expected, "{value}");
489        }
490        for value in [
491            "",
492            "1",
493            "-1s",
494            "1.2.3s",
495            "s",
496            "1 s",
497            "1d",
498            "1s-1s",
499            "9223372036854775808ns",
500        ] {
501            assert!(parse_duration(value).is_err(), "{value}");
502        }
503    }
504
505    #[test]
506    fn test_invalid_find_traces_parameters() {
507        for extra in [
508            "query.searchDepth=-1",
509            "query.searchDepth=10001",
510            "query.searchDepth=2147483647",
511            "query.durationMin=-1s",
512            "query.durationMin=2s&query.durationMax=1s",
513            "query.durationMax=999999999999h",
514            "query.attributes=%7B%22error%22%3Atrue%7D",
515            "query.attributes=invalid",
516            "query.filter=%7B%7D",
517        ] {
518            let uri = format!("/api/v3/traces?query.startTimeMin=2026-01-01T00:00:00Z&query.startTimeMax=2026-01-01T01:00:00Z&{extra}").parse().unwrap();
519            let Query(params) = Query::<FindTracesParams>::try_from_uri(&uri).unwrap();
520            assert_eq!(
521                params.into_query().unwrap_err().status,
522                StatusCode::BAD_REQUEST,
523                "{extra}"
524            );
525        }
526        for (start, end) in [
527            ("invalid", "2026-01-01T00:00:00Z"),
528            ("2500-01-01T00:00:00Z", "2500-01-02T00:00:00Z"),
529            ("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z"),
530        ] {
531            assert!(
532                FindTracesParams {
533                    start_time_min: start.into(),
534                    start_time_max: end.into(),
535                    ..Default::default()
536                }
537                .into_query()
538                .is_err()
539            );
540        }
541    }
542}
servers/http/jaeger/v3.rs

servers/http/jaeger/
v3.rs
