Skip to content

Commit 6caff50

Browse files
shuiyisongshuiyisong
authored
chore: improve search traces and jaeger resp (#7166)
* chore: add jaeger field in trace query Signed-off-by: shuiyisong <[email protected]> * chore: update search v1 with tags Signed-off-by: shuiyisong <[email protected]> * chore: update col matching using col names Signed-off-by: shuiyisong <[email protected]> * chore: minify code with macro Signed-off-by: shuiyisong <[email protected]> * chore: fix test Signed-off-by: shuiyisong <[email protected]> * chore: change macro to inline function Signed-off-by: shuiyisong <[email protected]> * chore: fix filter with tags & add test Signed-off-by: shuiyisong <[email protected]> --------- Signed-off-by: shuiyisong <[email protected]>
1 parent 421f4ee commit 6caff50

File tree

6 files changed

+247
-30
lines changed

6 files changed

+247
-30
lines changed

‎src/frontend/src/instance/jaeger.rs‎

Lines changed: 74 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
// See the License for the specific language governing permissions and
1313
// limitations under the License.
1414

15-
use std::collections::HashMap;
15+
use std::collections::{HashMap, HashSet};
1616
use std::sync::Arc;
1717

1818
use async_trait::async_trait;
@@ -28,6 +28,7 @@ use common_function::scalars::udf::create_udf;
2828
use common_query::{Output, OutputData};
2929
use common_recordbatch::adapter::RecordBatchStreamAdapter;
3030
use common_recordbatch::util;
31+
use common_telemetry::warn;
3132
use datafusion::dataframe::DataFrame;
3233
use datafusion::execution::SessionStateBuilder;
3334
use datafusion::execution::context::SessionContext;
@@ -42,8 +43,9 @@ use servers::error::{
4243
};
4344
use servers::http::jaeger::{JAEGER_QUERY_TABLE_NAME_KEY, QueryTraceParams};
4445
use servers::otlp::trace::{
45-
DURATION_NANO_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_KIND_COLUMN,
46-
SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, TIMESTAMP_COLUMN, TRACE_ID_COLUMN,
46+
DURATION_NANO_COLUMN, KEY_OTEL_STATUS_ERROR_KEY, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN,
47+
SPAN_KIND_COLUMN, SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, SPAN_STATUS_ERROR,
48+
TIMESTAMP_COLUMN, TRACE_ID_COLUMN,
4749
};
4850
use servers::query_handler::JaegerQueryHandler;
4951
use session::context::QueryContextRef;
@@ -322,6 +324,7 @@ async fn query_trace_table(
322324
})?;
323325

324326
let is_data_model_v1 = table
327+
.clone()
325328
.table_info()
326329
.meta
327330
.options
@@ -330,6 +333,14 @@ async fn query_trace_table(
330333
.map(|s| s.as_str())
331334
== Some(TABLE_DATA_MODEL_TRACE_V1);
332335

336+
// collect to set
337+
let col_names = table
338+
.table_info()
339+
.meta
340+
.field_column_names()
341+
.map(|s| format!("\"{}\"", s))
342+
.collect::<HashSet<String>>();
343+
333344
let df_context = create_df_context(query_engine)?;
334345

335346
let dataframe = df_context
@@ -342,7 +353,7 @@ async fn query_trace_table(
342353
let dataframe = filters
343354
.into_iter()
344355
.chain(tags.map_or(Ok(vec![]), |t| {
345-
tags_filters(&dataframe, t, is_data_model_v1)
356+
tags_filters(&dataframe, t, is_data_model_v1, &col_names)
346357
})?)
347358
.try_fold(dataframe, |df, expr| {
348359
df.filter(expr).context(DataFusionSnafu)
@@ -472,23 +483,73 @@ fn json_tag_filters(
472483
Ok(filters)
473484
}
474485

475-
fn flatten_tag_filters(tags: HashMap<String, JsonValue>) -> ServerResult<Vec<Expr>> {
486+
/// Helper function to check if span_key or resource_key exists in col_names and create an expression.
487+
/// If neither exists, logs a warning and returns None.
488+
#[inline]
489+
fn check_col_and_build_expr<F>(
490+
span_key: String,
491+
resource_key: String,
492+
key: &str,
493+
col_names: &HashSet<String>,
494+
expr_builder: F,
495+
) -> Option<Expr>
496+
where
497+
F: FnOnce(String) -> Expr,
498+
{
499+
if col_names.contains(&span_key) {
500+
return Some(expr_builder(span_key));
501+
}
502+
if col_names.contains(&resource_key) {
503+
return Some(expr_builder(resource_key));
504+
}
505+
warn!("tag key {} not found in table columns", key);
506+
None
507+
}
508+
509+
fn flatten_tag_filters(
510+
tags: HashMap<String, JsonValue>,
511+
col_names: &HashSet<String>,
512+
) -> ServerResult<Vec<Expr>> {
476513
let filters = tags
477514
.into_iter()
478515
.filter_map(|(key, value)| {
479-
let key = format!("\"span_attributes.{}\"", key);
516+
if key == KEY_OTEL_STATUS_ERROR_KEY && value == JsonValue::Bool(true) {
517+
return Some(col(SPAN_STATUS_CODE).eq(lit(SPAN_STATUS_ERROR)));
518+
}
519+
520+
// TODO(shuiyisong): add more precise mapping from key to col name
521+
let span_key = format!("\"span_attributes.{}\"", key);
522+
let resource_key = format!("\"resource_attributes.{}\"", key);
480523
match value {
481-
JsonValue::String(value) => Some(col(key).eq(lit(value))),
524+
JsonValue::String(value) => {
525+
check_col_and_build_expr(span_key, resource_key, &key, col_names, |k| {
526+
col(k).eq(lit(value))
527+
})
528+
}
482529
JsonValue::Number(value) => {
483530
if value.is_f64() {
484531
// safe to unwrap as checked previously
485-
Some(col(key).eq(lit(value.as_f64().unwrap())))
532+
let value = value.as_f64().unwrap();
533+
check_col_and_build_expr(span_key, resource_key, &key, col_names, |k| {
534+
col(k).eq(lit(value))
535+
})
486536
} else {
487-
Some(col(key).eq(lit(value.as_i64().unwrap())))
537+
let value = value.as_i64().unwrap();
538+
check_col_and_build_expr(span_key, resource_key, &key, col_names, |k| {
539+
col(k).eq(lit(value))
540+
})
488541
}
489542
}
490-
JsonValue::Bool(value) => Some(col(key).eq(lit(value))),
491-
JsonValue::Null => Some(col(key).is_null()),
543+
JsonValue::Bool(value) => {
544+
check_col_and_build_expr(span_key, resource_key, &key, col_names, |k| {
545+
col(k).eq(lit(value))
546+
})
547+
}
548+
JsonValue::Null => {
549+
check_col_and_build_expr(span_key, resource_key, &key, col_names, |k| {
550+
col(k).is_null()
551+
})
552+
}
492553
// not supported at the moment
493554
JsonValue::Array(_value) => None,
494555
JsonValue::Object(_value) => None,
@@ -502,9 +563,10 @@ fn tags_filters(
502563
dataframe: &DataFrame,
503564
tags: HashMap<String, JsonValue>,
504565
is_data_model_v1: bool,
566+
col_names: &HashSet<String>,
505567
) -> ServerResult<Vec<Expr>> {
506568
if is_data_model_v1 {
507-
flatten_tag_filters(tags)
569+
flatten_tag_filters(tags, col_names)
508570
} else {
509571
json_tag_filters(dataframe, tags)
510572
}

‎src/servers/src/http/jaeger.rs‎

Lines changed: 42 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,10 +40,12 @@ use crate::http::extractor::TraceTableName;
4040
use crate::metrics::METRIC_JAEGER_QUERY_ELAPSED;
4141
use crate::otlp::trace::{
4242
DURATION_NANO_COLUMN, KEY_OTEL_SCOPE_NAME, KEY_OTEL_SCOPE_VERSION, KEY_OTEL_STATUS_CODE,
43-
KEY_SERVICE_NAME, KEY_SPAN_KIND, RESOURCE_ATTRIBUTES_COLUMN, SCOPE_NAME_COLUMN,
44-
SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_EVENTS_COLUMN,
45-
SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, SPAN_STATUS_CODE,
46-
SPAN_STATUS_PREFIX, SPAN_STATUS_UNSET, TIMESTAMP_COLUMN, TRACE_ID_COLUMN,
43+
KEY_OTEL_STATUS_ERROR_KEY, KEY_OTEL_STATUS_MESSAGE, KEY_OTEL_TRACE_STATE, KEY_SERVICE_NAME,
44+
KEY_SPAN_KIND, RESOURCE_ATTRIBUTES_COLUMN, SCOPE_NAME_COLUMN, SCOPE_VERSION_COLUMN,
45+
SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN,
46+
SPAN_KIND_COLUMN, SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, SPAN_STATUS_ERROR,
47+
SPAN_STATUS_MESSAGE_COLUMN, SPAN_STATUS_PREFIX, SPAN_STATUS_UNSET, TIMESTAMP_COLUMN,
48+
TRACE_ID_COLUMN, TRACE_STATE_COLUMN,
4749
};
4850
use crate::query_handler::JaegerQueryHandlerRef;
4951

@@ -654,7 +656,10 @@ async fn covert_to_records(output: Output) -> Result<Option<HttpRecordsOutput>>
654656
.await
655657
.context(CollectRecordbatchSnafu)?,
656658
)?;
657-
debug!("The query records: {:?}", records);
659+
debug!(
660+
"The query records: {}",
661+
serde_json::to_string(&records).unwrap()
662+
);
658663
Ok(Some(records))
659664
}
660665
// It's unlikely to happen. However, if the output is not a stream, return None.
@@ -859,6 +864,38 @@ fn traces_from_records(records: HttpRecordsOutput) -> Result<Vec<Trace>> {
859864
value_type: ValueType::String,
860865
value: Value::String(normalize_status_code(&span_status)),
861866
});
867+
// set error to comply with the Jaeger API
868+
if span_status == SPAN_STATUS_ERROR {
869+
span.tags.push(KeyValue {
870+
key: KEY_OTEL_STATUS_ERROR_KEY.to_string(),
871+
value_type: ValueType::Boolean,
872+
value: Value::Boolean(true),
873+
});
874+
}
875+
}
876+
}
877+
878+
SPAN_STATUS_MESSAGE_COLUMN => {
879+
if let JsonValue::String(span_status_message) = cell
880+
&& !span_status_message.is_empty()
881+
{
882+
span.tags.push(KeyValue {
883+
key: KEY_OTEL_STATUS_MESSAGE.to_string(),
884+
value_type: ValueType::String,
885+
value: Value::String(span_status_message),
886+
});
887+
}
888+
}
889+
890+
TRACE_STATE_COLUMN => {
891+
if let JsonValue::String(trace_state) = cell
892+
&& !trace_state.is_empty()
893+
{
894+
span.tags.push(KeyValue {
895+
key: KEY_OTEL_TRACE_STATE.to_string(),
896+
value_type: ValueType::String,
897+
value: Value::String(trace_state),
898+
});
862899
}
863900
}
864901

‎src/servers/src/otlp/trace.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,13 @@ pub const TIMESTAMP_COLUMN: &str = "timestamp";
3434
pub const DURATION_NANO_COLUMN: &str = "duration_nano";
3535
pub const SPAN_KIND_COLUMN: &str = "span_kind";
3636
pub const SPAN_STATUS_CODE: &str = "span_status_code";
37+
pub const SPAN_STATUS_MESSAGE_COLUMN: &str = "span_status_message";
3738
pub const SPAN_ATTRIBUTES_COLUMN: &str = "span_attributes";
3839
pub const SPAN_EVENTS_COLUMN: &str = "span_events";
3940
pub const SCOPE_NAME_COLUMN: &str = "scope_name";
4041
pub const SCOPE_VERSION_COLUMN: &str = "scope_version";
4142
pub const RESOURCE_ATTRIBUTES_COLUMN: &str = "resource_attributes";
43+
pub const TRACE_STATE_COLUMN: &str = "trace_state";
4244

4345
// const keys
4446
pub const KEY_SERVICE_NAME: &str = "service.name";
@@ -49,6 +51,9 @@ pub const KEY_SPAN_KIND: &str = "span.kind";
4951
pub const KEY_OTEL_SCOPE_NAME: &str = "otel.scope.name";
5052
pub const KEY_OTEL_SCOPE_VERSION: &str = "otel.scope.version";
5153
pub const KEY_OTEL_STATUS_CODE: &str = "otel.status_code";
54+
pub const KEY_OTEL_STATUS_MESSAGE: &str = "otel.status_description";
55+
pub const KEY_OTEL_STATUS_ERROR_KEY: &str = "error";
56+
pub const KEY_OTEL_TRACE_STATE: &str = "w3c.tracestate";
5257

5358
/// The span kind prefix in the database.
5459
/// If the span kind is `server`, it will be stored as `SPAN_KIND_SERVER` in the database.
@@ -57,6 +62,7 @@ pub const SPAN_KIND_PREFIX: &str = "SPAN_KIND_";
5762
// The span status code prefix in the database.
5863
pub const SPAN_STATUS_PREFIX: &str = "STATUS_CODE_";
5964
pub const SPAN_STATUS_UNSET: &str = "STATUS_CODE_UNSET";
65+
pub const SPAN_STATUS_ERROR: &str = "STATUS_CODE_ERROR";
6066

6167
/// Convert SpanTraces to GreptimeDB row insert requests.
6268
/// Returns `InsertRequests` and total number of rows to ingest

‎src/servers/src/otlp/trace/v0.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@ use crate::error::Result;
2626
use crate::otlp::trace::span::{TraceSpan, parse};
2727
use crate::otlp::trace::{
2828
DURATION_NANO_COLUMN, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN,
29-
SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, TIMESTAMP_COLUMN,
30-
TRACE_ID_COLUMN,
29+
SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE,
30+
SPAN_STATUS_MESSAGE_COLUMN, TIMESTAMP_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN,
3131
};
3232
use crate::otlp::utils::{make_column_data, make_string_column_data};
3333
use crate::query_handler::PipelineHandlerRef;
@@ -124,9 +124,9 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()>
124124
make_string_column_data(PARENT_SPAN_ID_COLUMN, span.parent_span_id),
125125
make_string_column_data(SPAN_KIND_COLUMN, Some(span.span_kind)),
126126
make_string_column_data(SPAN_NAME_COLUMN, Some(span.span_name)),
127-
make_string_column_data("span_status_code", Some(span.span_status_code)),
128-
make_string_column_data("span_status_message", Some(span.span_status_message)),
129-
make_string_column_data("trace_state", Some(span.trace_state)),
127+
make_string_column_data(SPAN_STATUS_CODE, Some(span.span_status_code)),
128+
make_string_column_data(SPAN_STATUS_MESSAGE_COLUMN, Some(span.span_status_message)),
129+
make_string_column_data(TRACE_STATE_COLUMN, Some(span.trace_state)),
130130
];
131131
row_writer::write_fields(writer, fields.into_iter(), &mut row)?;
132132

‎src/servers/src/otlp/trace/v1.rs‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,10 @@ use crate::error::Result;
2727
use crate::otlp::trace::attributes::Attributes;
2828
use crate::otlp::trace::span::{TraceSpan, parse};
2929
use crate::otlp::trace::{
30-
DURATION_NANO_COLUMN, KEY_SERVICE_NAME, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN,
31-
SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, TIMESTAMP_COLUMN,
32-
TRACE_ID_COLUMN,
30+
DURATION_NANO_COLUMN, KEY_SERVICE_NAME, PARENT_SPAN_ID_COLUMN, SCOPE_NAME_COLUMN,
31+
SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN,
32+
SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, SPAN_STATUS_MESSAGE_COLUMN,
33+
TIMESTAMP_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN,
3334
};
3435
use crate::otlp::utils::{any_value_to_jsonb, make_column_data, make_string_column_data};
3536
use crate::query_handler::PipelineHandlerRef;
@@ -135,11 +136,11 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()>
135136
make_string_column_data(SPAN_ID_COLUMN, Some(span.span_id)),
136137
make_string_column_data(SPAN_KIND_COLUMN, Some(span.span_kind)),
137138
make_string_column_data(SPAN_NAME_COLUMN, Some(span.span_name)),
138-
make_string_column_data("span_status_code", Some(span.span_status_code)),
139-
make_string_column_data("span_status_message", Some(span.span_status_message)),
140-
make_string_column_data("trace_state", Some(span.trace_state)),
141-
make_string_column_data("scope_name", Some(span.scope_name)),
142-
make_string_column_data("scope_version", Some(span.scope_version)),
139+
make_string_column_data(SPAN_STATUS_CODE, Some(span.span_status_code)),
140+
make_string_column_data(SPAN_STATUS_MESSAGE_COLUMN, Some(span.span_status_message)),
141+
make_string_column_data(TRACE_STATE_COLUMN, Some(span.trace_state)),
142+
make_string_column_data(SCOPE_NAME_COLUMN, Some(span.scope_name)),
143+
make_string_column_data(SCOPE_VERSION_COLUMN, Some(span.scope_version)),
143144
];
144145
row_writer::write_fields(writer, fields.into_iter(), &mut row)?;
145146

0 commit comments

Comments
 (0)