diff --git a/src/frontend/src/instance/otlp/README.md b/src/frontend/src/instance/otlp/README.md index 8db2738418..4489ab0c3f 100644 --- a/src/frontend/src/instance/otlp/README.md +++ b/src/frontend/src/instance/otlp/README.md @@ -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; diff --git a/src/frontend/src/instance/otlp/trace_ingest.rs b/src/frontend/src/instance/otlp/trace_ingest.rs index f6ce60b2d1..671f6c6f7c 100644 --- a/src/frontend/src/instance/otlp/trace_ingest.rs +++ b/src/frontend/src/instance/otlp/trace_ingest.rs @@ -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, + 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, + 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::(); 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, 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, + messages: TraceFailureMessages, ) -> Option { 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::>() + .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) diff --git a/src/frontend/src/instance/otlp/trace_ingest/tests.rs b/src/frontend/src/instance/otlp/trace_ingest/tests.rs index b3a0570db5..ec5bad8975 100644 --- a/src/frontend/src/instance/otlp/trace_ingest/tests.rs +++ b/src/frontend/src/instance/otlp/trace_ingest/tests.rs @@ -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!["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}" ); } diff --git a/src/frontend/src/instance/otlp/trace_types.rs b/src/frontend/src/instance/otlp/trace_types.rs index 0e2df25053..603cc66c0d 100644 --- a/src/frontend/src/instance/otlp/trace_types.rs +++ b/src/frontend/src/instance/otlp/trace_types.rs @@ -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] diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 079439bf78..90cbb8c93f 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -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",