OTLP logs direct ingestion fails after auto-created table due to JSONB schema alignment mismatch
## Bug
Direct OTLP log ingestion can fail after the log table already exists. The first request can auto-create `opentelemetry_logs`, but a later request is rejected while aligning incoming rows with the existing table schema.
The observed error is:
```text
Invalid request parameter: failed to align log column `log_attributes` in table `opentelemetry_logs` from Binary to Json
```
## Impact
This can effectively break the default direct OTLP logs flow for sustained ingestion:
- First request to a missing `opentelemetry_logs` table can succeed and create the table.
- Second and later requests to the same table can fail on `log_attributes`.
- Pre-created OTLP log tables with the built-in JSONB columns can fail on the first ingest.
- Pipeline-based OTLP ingestion is likely unaffected; this is in the direct OTLP logs path.
## Root Cause
Recent OTLP logs schema-alignment changes in #8229 made `OtlpLogDirect` fetch the existing table schema and align incoming rows against it.
The built-in OTLP JSON columns such as `log_attributes`, `scope_attributes`, and `resource_attributes` are sent by the direct path as pre-encoded JSONB bytes:
```rust
ValueData::BinaryValue(...)
```
Their request schema uses:
```rust
ColumnDataType::Binary
JsonType(JsonBinary)
```
When the table schema is read back from metadata, these same columns round-trip as logical JSON columns:
```rust
ColumnDataType::Json
JsonType(JsonBinary)
```
The new alignment logic treats this as an incompatible `Binary -> Json` conversion even though both forms represent the same legacy JSONB column on the insert path. The request is rejected before it reaches storage.
## Reproduction
Add this test to an OSS integration test module that has access to `setup_cluster`, then run it against the OSS workspace. The important part is sending two direct OTLP log POSTs to `/v1/otlp/v1/logs` against a fresh instance with no existing `opentelemetry_logs` table.
```rust
use std::time::Duration;
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
use opentelemetry_proto::tonic::common::v1::{AnyValue, InstrumentationScope, KeyValue, any_value};
use opentelemetry_proto::tonic::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
use opentelemetry_proto::tonic::resource::v1::Resource;
use prost::Message;
fn create_otlp_logs_request(
num_logs: usize,
base_timestamp_nanos: u64,
) -> ExportLogsServiceRequest {
let log_records = (0..num_logs)
.map(|idx| LogRecord {
time_unix_nano: base_timestamp_nanos + idx as u64 * 1_000_000_000,
severity_number: 9,
severity_text: "INFO".to_string(),
body: Some(AnyValue {
value: Some(any_value::Value::StringValue(format!(
"integration test log {}",
idx
))),
}),
attributes: vec![KeyValue {
key: "test_index".to_string(),
value: Some(AnyValue {
value: Some(any_value::Value::IntValue(idx as i64)),
}),
}],
..Default::default()
})
.collect();
ExportLogsServiceRequest {
resource_logs: vec![ResourceLogs {
resource: Some(Resource {
attributes: vec![KeyValue {
key: "service.name".to_string(),
value: Some(AnyValue {
value: Some(any_value::Value::StringValue(
"otlp-repro-test".to_string(),
)),
}),
}],
..Default::default()
}),
scope_logs: vec![ScopeLogs {
scope: Some(InstrumentationScope {
name: "test-scope".to_string(),
version: "1.0.0".to_string(),
..Default::default()
}),
log_records,
..Default::default()
}],
..Default::default()
}],
}
}
#[tokio::test]
async fn test_otlp_logs_multiple_batches_repro() {
common_telemetry::init_default_ut_logging();
let cluster = setup_cluster().await;
let fe_http_addr = cluster.frontend.http_addr().await;
tokio::time::sleep(Duration::from_secs(3)).await;
let client = reqwest::Client::new();
let request1 = create_otlp_logs_request(3, 1_704_067_200_000_000_000);
let response1 = client
.post(format!("http://{}/v1/otlp/v1/logs", fe_http_addr))
.header("Content-Type", "application/x-protobuf")
.body(request1.encode_to_vec())
.send()
.await
.unwrap();
assert!(
response1.status().is_success(),
"first batch should create the table and be accepted, got {}",
response1.status()
);
let request2 = create_otlp_logs_request(4, 1_704_153_600_000_000_000);
let response2 = client
.post(format!("http://{}/v1/otlp/v1/logs", fe_http_addr))
.header("Content-Type", "application/x-protobuf")
.body(request2.encode_to_vec())
.send()
.await
.unwrap();
assert!(
response2.status().is_success(),
"second batch should be accepted, got {}",
response2.status()
);
}
```
On the regressed code, the second request returns `400 Bad Request` with:
```text
failed to align log column `log_attributes` in table `opentelemetry_logs` from Binary to Json
```
## Expected Behavior
Repeated direct OTLP log batches should be accepted after the table is auto-created.
Existing `Json + JsonBinary` table columns should be considered compatible with incoming `Binary + JsonBinary` OTLP direct values that are already encoded as JSONB bytes.
## Fix Direction
Do not simply change the first-batch schema to `ColumnDataType::Json` without changing values, because operator preprocessing expects JSON columns to carry `StringValue` JSON text and would try to re-encode them.
A minimal fix is to teach the OTLP logs existing-schema alignment that `Json + JsonBinary` existing columns are compatible with incoming `Binary + JsonBinary` request values, while keeping the insert wire schema as `Binary + JsonBinary` for pre-encoded JSONB bytes.
0 条评论