servers/otlp/
trace.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
15pub mod attributes;
16pub mod span;
17pub mod v0;
18pub mod v1;
19
20use std::collections::HashSet;
21
22use api::v1::RowInsertRequests;
23// Column names. The `greptime_trace_v1` fixed columns are canonically defined
24// in `common_catalog::consts` (shared with the read-time graph derivation).
25pub use common_catalog::consts::{
26    DURATION_NANO_COLUMN, SERVICE_NAME_COLUMN, SPAN_KIND_COLUMN,
27    SPAN_STATUS_CODE_COLUMN as SPAN_STATUS_CODE, TRACE_TIMESTAMP_COLUMN as TIMESTAMP_COLUMN,
28};
29pub use common_catalog::consts::{
30    PARENT_SPAN_ID_COLUMN, SPAN_ID_COLUMN, SPAN_NAME_COLUMN, TRACE_ID_COLUMN,
31};
32use pipeline::{GreptimePipelineParams, PipelineWay};
33use session::context::QueryContextRef;
34
35use crate::error::{NotSupportedSnafu, Result};
36use crate::otlp::trace::span::TraceSpan;
37use crate::query_handler::PipelineHandlerRef;
38
39pub const SPAN_STATUS_MESSAGE_COLUMN: &str = "span_status_message";
40pub const SPAN_ATTRIBUTES_COLUMN: &str = "span_attributes";
41pub const SPAN_EVENTS_COLUMN: &str = "span_events";
42pub const SCOPE_NAME_COLUMN: &str = "scope_name";
43pub const SCOPE_VERSION_COLUMN: &str = "scope_version";
44pub const RESOURCE_ATTRIBUTES_COLUMN: &str = "resource_attributes";
45pub const TRACE_STATE_COLUMN: &str = "trace_state";
46
47// const keys
48pub const KEY_SERVICE_NAME: &str = "service.name";
49pub const KEY_SERVICE_NAMESPACE: &str = "service.namespace";
50pub const KEY_SERVICE_INSTANCE_ID: &str = "service.instance.id";
51pub const KEY_HOST_ID: &str = "host.id";
52pub const KEY_HOST_NAME: &str = "host.name";
53pub const KEY_CONTAINER_ID: &str = "container.id";
54pub const KEY_CONTAINER_NAME: &str = "container.name";
55pub const KEY_K8S_POD_UID: &str = "k8s.pod.uid";
56pub const KEY_K8S_POD_NAME: &str = "k8s.pod.name";
57pub const KEY_K8S_CONTAINER_NAME: &str = "k8s.container.name";
58pub const KEY_K8S_NAMESPACE_NAME: &str = "k8s.namespace.name";
59pub const KEY_K8S_NODE_NAME: &str = "k8s.node.name";
60pub const KEY_SPAN_KIND: &str = "span.kind";
61
62// jaeger const keys, not sure if they are general
63pub const KEY_OTEL_SCOPE_NAME: &str = "otel.scope.name";
64pub const KEY_OTEL_SCOPE_VERSION: &str = "otel.scope.version";
65pub const KEY_OTEL_STATUS_CODE: &str = "otel.status_code";
66pub const KEY_OTEL_STATUS_MESSAGE: &str = "otel.status_description";
67pub const KEY_OTEL_STATUS_ERROR_KEY: &str = "error";
68pub const KEY_OTEL_TRACE_STATE: &str = "w3c.tracestate";
69
70/// The span kind prefix in the database.
71/// If the span kind is `server`, it will be stored as `SPAN_KIND_SERVER` in the database.
72pub const SPAN_KIND_PREFIX: &str = "SPAN_KIND_";
73
74// The span status code prefix in the database.
75pub const SPAN_STATUS_PREFIX: &str = "STATUS_CODE_";
76pub const SPAN_STATUS_UNSET: &str = "STATUS_CODE_UNSET";
77pub use common_catalog::consts::SPAN_STATUS_ERROR;
78
79/// Deduplicated auxiliary trace entities derived from successfully ingested
80/// spans.
81///
82/// The main trace table is written first. Once a span is confirmed accepted, we
83/// record the service and operation tuples here so the auxiliary tables can be
84/// updated separately without affecting span acceptance accounting.
85#[derive(Debug, Default)]
86pub struct TraceAuxData {
87    pub services: HashSet<String>,
88    pub operations: HashSet<(String, String, String)>,
89}
90
91impl TraceAuxData {
92    /// Records the auxiliary service and operation rows implied by one accepted
93    /// span.
94    pub fn observe_span(&mut self, span: &TraceSpan) {
95        if let Some(service_name) = &span.service_name {
96            self.services.insert(service_name.clone());
97            self.operations.insert((
98                service_name.clone(),
99                span.span_name.clone(),
100                span.span_kind.clone(),
101            ));
102        }
103    }
104
105    /// Returns true when no auxiliary table updates are needed.
106    pub fn is_empty(&self) -> bool {
107        self.services.is_empty() && self.operations.is_empty()
108    }
109}
110
111/// Convert a subset of trace spans to GreptimeDB row insert requests.
112pub fn to_grpc_insert_requests_from_spans(
113    spans: &[TraceSpan],
114    pipeline: &PipelineWay,
115    pipeline_params: &GreptimePipelineParams,
116    table_name: &str,
117    query_ctx: &QueryContextRef,
118    pipeline_handler: PipelineHandlerRef,
119) -> Result<(RowInsertRequests, usize)> {
120    match pipeline {
121        PipelineWay::OtlpTraceDirectV0 => v0::v0_to_grpc_main_insert_requests(
122            spans,
123            pipeline,
124            pipeline_params,
125            table_name,
126            query_ctx,
127            pipeline_handler,
128        ),
129        PipelineWay::OtlpTraceDirectV1 => v1::v1_to_grpc_main_insert_requests(
130            spans,
131            pipeline,
132            pipeline_params,
133            table_name,
134            query_ctx,
135            pipeline_handler,
136        ),
137        _ => NotSupportedSnafu {
138            feat: "Unsupported pipeline for trace",
139        }
140        .fail(),
141    }
142}
143
144/// Build insert requests for the auxiliary trace tables derived from accepted
145/// spans.
146///
147/// "Aux" here refers to the trace service and trace operation tables, not the
148/// main trace span table itself.
149pub fn to_grpc_insert_requests_for_aux_tables(
150    aux_data: TraceAuxData,
151    pipeline: &PipelineWay,
152    table_name: &str,
153) -> Result<(RowInsertRequests, usize)> {
154    match pipeline {
155        PipelineWay::OtlpTraceDirectV0 => v0::build_aux_table_requests(aux_data, table_name),
156        PipelineWay::OtlpTraceDirectV1 => v1::build_aux_table_requests(aux_data, table_name),
157        _ => NotSupportedSnafu {
158            feat: "Unsupported pipeline for trace",
159        }
160        .fail(),
161    }
162}
servers/otlp/trace.rs

servers/otlp/
trace.rs
