rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-05 21:18:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(otlp): report the cause of rejected trace spans (#8897)

Browse Source 
* feat(otlp): report the cause of rejected trace spans

When trace-v1 ingestion cannot coerce an attribute value, it falls back to
single-span writes and rejects the bad span. That behavior is correct, but the
OTLP partial-success message only carried `Rejected span <trace_id>:<span_id>
(InvalidArguments)`: the column, the source value, the source type and the
target type were all dropped, so locating the bad attribute required adding a
detailed exporter on the collector side and replaying traffic.

Two places lost the information. `prepare_trace_column_rewrites` built a message
without the failing value, and the span rejection path kept only the status code
from the error.

Coercion errors now name the failing value, e.g.

    failed to coerce trace column 'span_attributes.http.response.body.size'
    in table 'opentelemetry_traces' from String("") to Int64

and the rejection detail carries that cause. Values are user data, so a string
keeps at most 16 characters, is escaped, and binary payloads report only their
length; the cause itself is bounded at 256 characters. Both truncations cut on a
char boundary.

Failure details now deduplicate: repeats of the same (site, cause) collapse into
one entry with an occurrence count, keyed on the untruncated cause so two
failures that differ past the display limit stay separate. Only four distinct
entries are retained and the rest are counted, which keeps the state bounded no
matter how many distinct bad values a request carries. A fully rejected request
is logged at warn level and a partial success at debug level, since the latter
repeats every export interval; the detail goes out as a Debug field so a newline
in an attribute key cannot forge log lines.

Rejection semantics are unchanged: partial success, same accepted and rejected
counts, same HTTP status mapping.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(otlp): compare failure keys directly instead of hashing

The dedup identity was a DefaultHasher fingerprint of `(label, key)`, which
bought a fixed 8 bytes per entry at the cost of an import, four lines, and a
collision argument the reader has to make. Entries are capped at four and a
cause runs a couple of hundred characters, so the saving is about a kilobyte
per in-flight request while the column name it avoids retaining is already held
several times over by the request itself.

Compare the strings instead, keeping the untruncated cause as the key so
failures differing past the display limit still stay apart. Labels are metric
label values and always static, so the entry borrows them.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
dennis zhuang
2026-08-19 04:32:23 +00:00
committed by GitHub
parent 8dbfbee611
commit 3de4feddd5
5 changed files with 436 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/frontend/src/instance/otlp/README.md
+12
View File
@@ -158,6 +158,18 @@ Accounting follows the main-table write:
accepted or rejected span counts; and
- failure details are bounded before they are folded into `TraceIngestOutcome`.
A rejection detail carries the failing cause, not just a status code, so an
unusable attribute value names its column, source value, and target type. Repeats
of the same `(site, cause)` collapse into one entry with an occurrence count, and
only a fixed number of distinct entries is retained; failures past that limit are
counted but their causes are dropped, which keeps the state bounded regardless of
how many distinct bad values a request carries.
A request that rejects every span is logged at warn level. A partial success is
logged at debug level only: it repeats every export interval while the sender
keeps emitting the value, and its detail already reaches the sender through the
OTLP partial-success message.
Finally, the HTTP handler returns:
- full success when there are no rejected spans or failure details;
src/frontend/src/instance/otlp/trace_ingest.rs
+152 -20
View File
@@ -25,7 +25,7 @@ use client::Output;
use common_error::ext::{BoxedError, ErrorExt};
use common_error::status_code::StatusCode;
use common_meta::rpc::ddl::TriggerReason;
use common_telemetry::warn;
use common_telemetry::{debug, warn};
use datatypes::prelude::ConcreteDataType;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use pipeline::{GreptimePipelineParams, PipelineWay};
@@ -50,11 +50,15 @@ use crate::instance::otlp::trace_types::{
PendingTraceColumnRewrite, PreparedTraceColumnRewrites, TraceColumnRewriteError,
choose_trace_reconcile_decision, enrich_trace_reconcile_error,
is_trace_reconcile_candidate_type, prepare_trace_column_rewrites, push_observed_trace_type,
truncate_for_diagnostics,
};
use crate::metrics::{OTLP_TRACES_FAILURE_COUNT, OTLP_TRACES_ROWS};
const TRACE_FAILURE_MESSAGE_LIMIT: usize = 4;
/// Maximum characters of a failure cause echoed to the client and the log.
const TRACE_FAILURE_CAUSE_LIMIT: usize = 256;
/// Determines how trace ingestion responds to a failure before a write is dispatched.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ChunkFailureReaction {
@@ -86,7 +90,34 @@ struct TraceChunkIngestContext<'a> {
struct TraceIngestState {
aux_data: TraceAuxData,
outcome: TraceIngestOutcome,
failure_messages: Vec<String>,
failure_messages: TraceFailureMessages,
}
/// Bounded, deduplicated failure details for one trace request.
///
/// Occurrences past [`TRACE_FAILURE_MESSAGE_LIMIT`] distinct failures are
/// counted but their keys are dropped: retaining them would let this state grow
/// with the number of distinct bad values in a request.
#[derive(Debug, Default)]
struct TraceFailureMessages {
entries: Vec<TraceFailureEntry>,
suppressed_occurrences: usize,
}
#[derive(Debug)]
struct TraceFailureEntry {
label: &'static str,
/// Untruncated: two causes can agree on a truncated prefix and differ
/// exactly where the actionable detail is.
key: String,
message: String,
occurrences: usize,
}
impl TraceFailureMessages {
fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
/// How a v1 chunk should be reconciled when it is written.
@@ -863,7 +894,7 @@ impl Instance {
let mut ingest_state = TraceIngestState {
aux_data: TraceAuxData::default(),
outcome: TraceIngestOutcome::default(),
failure_messages: Vec::new(),
failure_messages: TraceFailureMessages::default(),
};
let main_result: ServerResult<()> = async {
@@ -954,6 +985,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"aux_table_update_failed",
err.status_code().as_ref(),
format!(
"Auxiliary trace tables were not fully updated ({})",
err.status_code().as_ref()
@@ -968,6 +1000,7 @@ impl Instance {
Err(err) => Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"aux_table_update_failed",
err.status_code().as_ref(),
format!(
"Auxiliary trace tables were not fully updated ({})",
err.status_code().as_ref()
@@ -985,6 +1018,35 @@ impl Instance {
ingest_state.failure_messages,
);
if let Some(error_message) = &ingest_state.outcome.error_message {
let accepted_spans = ingest_state.outcome.accepted_spans;
let rejected_spans = ingest_state.outcome.rejected_spans;
// A partial success repeats every export interval while the sender
// keeps emitting the bad value, so only a fully rejected request is
// worth a warning. Either way the detail reaches the sender: as an
// OTLP partial success, or as the status message of a 400.
//
// The detail embeds attribute keys verbatim, so it goes out as a
// Debug field; interpolating it would let a newline in an attribute
// key forge log lines.
if accepted_spans == 0 && rejected_spans > 0 {
warn!(
table_name = ingest_ctx.table_name,
rejected_spans,
error_message = ?error_message,
"OTLP trace ingest rejected every span"
);
} else {
debug!(
table_name = ingest_ctx.table_name,
accepted_spans,
rejected_spans,
error_message = ?error_message,
"OTLP trace ingest reported failures"
);
}
}
Ok(ingest_state.outcome)
}
@@ -1019,6 +1081,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating chunk write failure ({})",
err.status_code().as_ref()
@@ -1069,13 +1132,18 @@ impl Instance {
// recover valid data.
let span_count = chunks.iter().map(|chunk| chunk.rows.len()).sum::<usize>();
ingest_state.outcome.rejected_spans += span_count;
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::DiscardChunk.as_metric_label(),
// Merging discards of different sizes would make the reported
// count times its occurrences wrong.
&format!("{span_count}:{cause}"),
format!(
"Discarded {} spans after pre-write request failure ({})",
"Discarded {} spans after pre-write request failure ({}): {}",
span_count,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
return Ok(());
@@ -1110,6 +1178,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::RetryPerSpan.as_metric_label(),
"incompatible_binary_and_json",
"Chunk fallback triggered by incompatible binary and JSON values".to_string(),
);
return self
@@ -1136,6 +1205,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::RetryPerSpan.as_metric_label(),
err.status_code().as_ref(),
format!("Chunk fallback triggered by {}", err.status_code().as_ref()),
);
return self
@@ -1144,13 +1214,19 @@ impl Instance {
}
ChunkFailureReaction::DiscardChunk => {
ingest_state.outcome.rejected_spans += span_count;
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::DiscardChunk.as_metric_label(),
// Chunks discarded for one cause differ in size; merging
// them would make the reported count times its
// occurrences wrong.
&format!("{span_count}:{cause}"),
format!(
"Discarded {} spans after pre-write chunk failure ({})",
"Discarded {} spans after pre-write chunk failure ({}): {}",
span_count,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
return Ok(());
@@ -1159,6 +1235,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating pre-write chunk failure ({})",
err.status_code().as_ref()
@@ -1182,6 +1259,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating chunk write failure ({})",
err.status_code().as_ref()
@@ -1221,6 +1299,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating pre-write span failure for {}:{} ({})",
span.trace_id,
@@ -1232,14 +1311,19 @@ impl Instance {
}
ingest_state.outcome.rejected_spans += 1;
// Dedup on the cause, not the span id: one bad column rejects
// every span and would otherwise fill the entry list.
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"span_rejected",
&cause,
format!(
"Rejected span {}:{} ({})",
"Rejected span {}:{} ({}): {}",
span.trace_id,
span.span_id,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
continue;
@@ -1256,6 +1340,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating span write failure for {}:{} ({})",
span.trace_id,
@@ -1619,24 +1704,52 @@ impl Instance {
outcome.write_cost += cost;
}
fn push_trace_failure_message(messages: &mut Vec<String>, label: &str, message: String) {
/// Returns the full cause of a pre-write failure and its display form.
///
/// `output_msg` masks internal errors and unwraps the root cause. Dedup must
/// key on the full text: two causes can agree on a truncated prefix and
/// differ exactly where the actionable detail is.
fn trace_failure_cause(err: &error::Error) -> (String, String) {
let cause = err.output_msg();
let display = truncate_for_diagnostics(&cause, TRACE_FAILURE_CAUSE_LIMIT);
(cause, display)
}
/// Records one failure, merging repeats of `(label, key)` into a count.
fn push_trace_failure_message(
messages: &mut TraceFailureMessages,
label: &'static str,
key: &str,
message: String,
) {
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).inc();
if messages.len() < TRACE_FAILURE_MESSAGE_LIMIT {
messages.push(message);
} else if messages.len() == TRACE_FAILURE_MESSAGE_LIMIT {
tracing::debug!(
label,
limit = TRACE_FAILURE_MESSAGE_LIMIT,
"Trace ingest failure message limit reached; suppressing additional failure details"
);
if let Some(entry) = messages
.entries
.iter_mut()
.find(|entry| entry.label == label && entry.key == key)
{
entry.occurrences += 1;
return;
}
if messages.entries.len() >= TRACE_FAILURE_MESSAGE_LIMIT {
messages.suppressed_occurrences += 1;
return;
}
messages.entries.push(TraceFailureEntry {
label,
key: key.to_string(),
message,
occurrences: 1,
});
}
fn finish_trace_failure_message(
accepted_spans: usize,
rejected_spans: usize,
messages: Vec<String>,
messages: TraceFailureMessages,
) -> Option<String> {
if rejected_spans == 0 && messages.is_empty() {
return None;
@@ -1648,8 +1761,27 @@ impl Instance {
);
if !messages.is_empty() {
let details = messages
.entries
.into_iter()
.map(|entry| {
if entry.occurrences > 1 {
format!("{} (x{})", entry.message, entry.occurrences)
} else {
entry.message
}
})
.collect::<Vec<_>>()
.join("; ");
summary.push_str(": ");
summary.push_str(&messages.join("; "));
summary.push_str(&details);
}
if messages.suppressed_occurrences > 0 {
summary.push_str(&format!(
"; {} additional failures suppressed",
messages.suppressed_occurrences
));
}
Some(summary)
src/frontend/src/instance/otlp/trace_ingest/tests.rs
+128 -23
View File
@@ -34,8 +34,8 @@ use servers::otlp::trace::span::{SpanEvents, SpanLinks, TraceSpan};
use servers::otlp::trace::v1::{TraceBinaryType, TraceRetryColumn};
use super::{
ChunkFailureReaction, Instance, TraceChunkRetry, TraceChunkSchemaState, TraceRequestSchema,
TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
ChunkFailureReaction, Instance, TraceChunkRetry, TraceChunkSchemaState, TraceFailureMessages,
TraceRequestSchema, TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
wrap_trace_alter_failure,
};
use crate::metrics::OTLP_TRACES_FAILURE_COUNT;
@@ -102,22 +102,28 @@ fn test_classify_trace_prewrite_failure() {
#[test]
fn test_finish_trace_failure_message() {
let message = Instance::finish_trace_failure_message(
3,
2,
vec!["Rejected span trace:span (InvalidArguments)".to_string()],
)
.unwrap();
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
"coercion",
"Rejected span trace:span (InvalidArguments)".to_string(),
);
let message = Instance::finish_trace_failure_message(3, 2, messages).unwrap();
assert!(message.contains("Accepted 3 spans, rejected 2 spans"));
assert!(message.contains("Rejected span trace:span"));
assert_eq!(Instance::finish_trace_failure_message(2, 0, vec![]), None);
assert_eq!(
Instance::finish_trace_failure_message(2, 0, TraceFailureMessages::default()),
None
);
}
#[test]
fn test_finish_trace_failure_message_without_detail_messages() {
assert_eq!(
Instance::finish_trace_failure_message(0, 2, vec![]),
Instance::finish_trace_failure_message(0, 2, TraceFailureMessages::default()),
Some("Accepted 0 spans, rejected 2 spans".to_string())
);
}
@@ -126,39 +132,138 @@ fn test_finish_trace_failure_message_without_detail_messages() {
fn test_push_trace_failure_message_increments_labeled_counter() {
let label = "retry_per_span_counter_test";
let initial = OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get();
let mut messages = Vec::new();
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
label,
"InvalidArguments",
"Chunk fallback triggered by InvalidArguments".to_string(),
);
assert_eq!(messages.len(), 1);
assert_eq!(messages.entries.len(), 1);
assert_eq!(
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get(),
initial + 1
);
}
/// One bad column rejects every span, so repeats must collapse instead of
/// filling the bounded entry list with the same cause.
#[test]
fn test_push_trace_failure_message_collapses_repeated_cause() {
let label = "span_rejected_dedup_test";
let initial = OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get();
let mut messages = TraceFailureMessages::default();
for idx in 0..3 {
Instance::push_trace_failure_message(
&mut messages,
label,
"failed to coerce",
format!("Rejected span trace:span-{idx} (InvalidArguments): failed to coerce"),
);
}
assert_eq!(messages.entries.len(), 1);
assert_eq!(messages.entries[0].occurrences, 3);
assert_eq!(messages.suppressed_occurrences, 0);
// Every occurrence is still metered even though the details collapse.
assert_eq!(
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get(),
initial + 3
);
let summary = Instance::finish_trace_failure_message(0, 3, messages).unwrap();
assert!(
summary.contains("Rejected span trace:span-0 (InvalidArguments): failed to coerce (x3)"),
"unexpected summary: {summary}"
);
}
/// Dedup identity comes from the full cause, so two failures that only differ
/// past the display limit must stay separate instead of one silently vanishing.
#[test]
fn test_trace_failure_cause_keeps_identity_beyond_the_display_limit() {
let shared_prefix = "x".repeat(400);
let (first_cause, first_shown) = Instance::trace_failure_cause(
&servers::error::InvalidParameterSnafu {
reason: format!("{shared_prefix}-int64"),
}
.build(),
);
let (second_cause, second_shown) = Instance::trace_failure_cause(
&servers::error::InvalidParameterSnafu {
reason: format!("{shared_prefix}-float64"),
}
.build(),
);
assert_eq!(first_shown, second_shown);
assert_ne!(first_cause, second_cause);
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(&mut messages, "span_rejected", &first_cause, first_shown);
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
&second_cause,
second_shown,
);
assert_eq!(messages.entries.len(), 2);
}
/// The same cause reported by two different sites must stay distinguishable.
#[test]
fn test_push_trace_failure_message_separates_labels_sharing_a_key() {
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
"failed to coerce",
"Rejected span trace:span (InvalidArguments): failed to coerce".to_string(),
);
Instance::push_trace_failure_message(
&mut messages,
"discard_chunk",
"failed to coerce",
"Discarded 7 spans after pre-write chunk failure (InvalidArguments): failed to coerce"
.to_string(),
);
assert_eq!(messages.entries.len(), 2);
}
#[test]
fn test_push_trace_failure_message_caps_recorded_messages() {
let label = "retry_per_span_limit_test";
let mut messages = Vec::new();
let mut messages = TraceFailureMessages::default();
for idx in 0..=4 {
Instance::push_trace_failure_message(&mut messages, label, format!("failure-{idx}"));
for idx in 0..=5 {
Instance::push_trace_failure_message(
&mut messages,
label,
&format!("cause-{idx}"),
format!("failure-{idx}"),
);
}
assert_eq!(messages.len(), 4);
assert_eq!(
messages,
vec![
"failure-0".to_string(),
"failure-1".to_string(),
"failure-2".to_string(),
"failure-3".to_string()
]
messages
.entries
.iter()
.map(|entry| entry.message.as_str())
.collect::<Vec<_>>(),
vec!["failure-0", "failure-1", "failure-2", "failure-3"]
);
assert_eq!(messages.suppressed_occurrences, 2);
let summary = Instance::finish_trace_failure_message(0, 6, messages).unwrap();
assert!(
summary.ends_with("; 2 additional failures suppressed"),
"unexpected summary: {summary}"
);
}
src/frontend/src/instance/otlp/trace_types.rs
+135 -18
View File
@@ -22,6 +22,39 @@ use servers::otlp::coerce::{
use crate::instance::otlp::trace_semconv::trace_semconv_fixed_type;
/// Attribute values are user data echoed back in the OTLP partial-success
/// message and the server log, so diagnostics keep at most this many characters.
const TRACE_VALUE_DIAGNOSTIC_LIMIT: usize = 16;
/// Truncates to `limit` characters, marking the cut with `...`.
///
/// Diagnostics carry user-controlled text; slicing by byte offset would panic in
/// the middle of a multi-byte sequence.
pub(super) fn truncate_for_diagnostics(text: &str, limit: usize) -> String {
match text.char_indices().nth(limit) {
Some((offset, _)) => format!("{}...", &text[..offset]),
None => text.to_string(),
}
}
/// Renders a failing trace value as `Type(value)`, e.g. `String("")`.
fn describe_trace_value(value: &ValueData, request_type: ColumnDataType) -> String {
let payload = match value {
ValueData::StringValue(string_value) => format!(
"{:?}",
truncate_for_diagnostics(string_value, TRACE_VALUE_DIAGNOSTIC_LIMIT)
),
ValueData::BoolValue(bool_value) => bool_value.to_string(),
ValueData::I64Value(int_value) => int_value.to_string(),
ValueData::F64Value(float_value) => float_value.to_string(),
ValueData::BinaryValue(bytes) => format!("{} bytes", bytes.len()),
// Other value kinds never reach trace coercion, so report the type alone.
_ => return format!("{request_type:?}"),
};
format!("{request_type:?}({payload})")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum TraceReconcileDecision {
UseExisting(ColumnDataType),
@@ -191,8 +224,10 @@ pub(super) fn prepare_trace_column_rewrites(
let Some(value) = row.values.get(pending_rewrite.col_idx) else {
continue;
};
let Some(request_type) = value.value_data.as_ref().and_then(trace_value_datatype)
else {
let Some(request_value) = value.value_data.as_ref() else {
continue;
};
let Some(request_type) = trace_value_datatype(request_value) else {
continue;
};
if request_type == pending_rewrite.target_type {
@@ -201,20 +236,18 @@ pub(super) fn prepare_trace_column_rewrites(
let value_data =
coerce_value_data(&value.value_data, pending_rewrite.target_type, request_type)
.map_err(|_| {
TraceColumnRewriteError {
error: error::InvalidParameterSnafu {
reason: format!(
"failed to coerce trace column '{}' in table '{}' from {:?} to {:?}",
pending_rewrite.column_name,
table_name,
request_type,
pending_rewrite.target_type
),
}
.build(),
column_name: pending_rewrite.column_name.clone(),
}
.map_err(|_| TraceColumnRewriteError {
error: error::InvalidParameterSnafu {
reason: format!(
"failed to coerce trace column '{}' in table '{}' from {} to {:?}",
pending_rewrite.column_name,
table_name,
describe_trace_value(request_value, request_type),
pending_rewrite.target_type
),
}
.build(),
column_name: pending_rewrite.column_name.clone(),
})?;
values.push(PreparedTraceValueRewrite {
row_idx,
@@ -301,8 +334,8 @@ mod tests {
use super::{
PendingTraceColumnRewrite, TraceReconcileDecision, choose_trace_reconcile_decision,
enrich_trace_reconcile_error, is_trace_reconcile_candidate_type,
prepare_trace_column_rewrites, push_observed_trace_type,
describe_trace_value, enrich_trace_reconcile_error, is_trace_reconcile_candidate_type,
prepare_trace_column_rewrites, push_observed_trace_type, truncate_for_diagnostics,
};
#[test]
@@ -458,6 +491,90 @@ mod tests {
.unwrap_err();
assert_eq!(err.error.status_code(), StatusCode::InvalidArguments);
assert_eq!(err.column_name, "span_attributes.attr_int");
assert!(
err.error.to_string().contains(
"failed to coerce trace column 'span_attributes.attr_int' in table \
'trace_type_atomicity' from String(\"not_a_number\") to Int64"
),
"unexpected error message: {}",
err.error
);
}
/// The PHP instrumentation case: an empty string must be distinguishable
/// from any other unparsable value in the reported diagnostics.
#[test]
fn test_prepare_trace_column_rewrites_reports_empty_string_value() {
let rows = vec![Row {
values: vec![Value {
value_data: Some(ValueData::StringValue(String::new())),
}],
}];
let pending_rewrites = vec![PendingTraceColumnRewrite {
col_idx: 0,
target_type: ColumnDataType::Int64,
column_name: "span_attributes.http.response.body.size".to_string(),
}];
let err = prepare_trace_column_rewrites(&rows, pending_rewrites, "opentelemetry_traces")
.unwrap_err();
assert!(
err.error.to_string().contains(
"'span_attributes.http.response.body.size' in table 'opentelemetry_traces' \
from String(\"\") to Int64"
),
"unexpected error message: {}",
err.error
);
}
#[test]
fn test_describe_trace_value_bounds_and_escapes_strings() {
assert_eq!(
describe_trace_value(
&ValueData::StringValue(String::new()),
ColumnDataType::String
),
r#"String("")"#
);
assert_eq!(
describe_trace_value(
&ValueData::StringValue("a\tb\"c".to_string()),
ColumnDataType::String
),
r#"String("a\tb\"c")"#
);
assert_eq!(
describe_trace_value(
&ValueData::StringValue("0123456789abcdefghij".to_string()),
ColumnDataType::String
),
r#"String("0123456789abcdef...")"#
);
}
#[test]
fn test_describe_trace_value_omits_binary_content() {
assert_eq!(
describe_trace_value(
&ValueData::BinaryValue(vec![1_u8, 2, 3]),
ColumnDataType::Binary
),
"Binary(3 bytes)"
);
assert_eq!(
describe_trace_value(&ValueData::F64Value(1.5), ColumnDataType::Float64),
"Float64(1.5)"
);
}
/// Truncation runs over user-supplied text, so it must not split a
/// multi-byte character.
#[test]
fn test_truncate_for_diagnostics_cuts_on_char_boundary() {
assert_eq!(truncate_for_diagnostics("日本語テキスト", 3), "日本語...");
assert_eq!(truncate_for_diagnostics("short", 16), "short");
assert_eq!(truncate_for_diagnostics("exact", 5), "exact");
}
#[test]
tests-integration/tests/http.rs
+9
View File
@@ -7782,6 +7782,15 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) {
),
"unexpected partial success body: {body:?}"
);
// The rejection must name the column, the failing value and the target type,
// otherwise locating the bad attribute needs a collector-side capture.
assert!(
partial_success.error_message.contains(
"failed to coerce trace column 'span_attributes.attr_int' in table \
'trace_type_abort' from String(\"not_a_number\") to Int64"
),
"unexpected partial success body: {body:?}"
);
validate_data(
"otlp_traces_v1_type_abort_rows",