From 751cb62de9d8cabd85bb796e56cdd63db6829e31 Mon Sep 17 00:00:00 2001 From: Thomas Date: Thu, 8 Oct 2026 17:36:49 -0400 Subject: [PATCH 01/13] feat(codecs): convert native Vector logs to OTLP in otlp serializer --- .../otlp_native_log_encoding.feature.md | 6 + lib/codecs/src/encoding/format/otlp.rs | 50 ++- lib/opentelemetry-proto/src/common.rs | 51 ++- lib/opentelemetry-proto/src/logs.rs | 419 +++++++++++++++++- .../components/sinks/opentelemetry.cue | 39 +- 5 files changed, 550 insertions(+), 15 deletions(-) create mode 100644 changelog.d/otlp_native_log_encoding.feature.md diff --git a/changelog.d/otlp_native_log_encoding.feature.md b/changelog.d/otlp_native_log_encoding.feature.md new file mode 100644 index 0000000000000..549efd18ad518 --- /dev/null +++ b/changelog.d/otlp_native_log_encoding.feature.md @@ -0,0 +1,6 @@ +The `otlp` codec now converts native Vector log events to OTLP log records, so logs from any source can +be sent with the `opentelemetry` sink without a `remap` transform that builds the `resourceLogs` structure. +The conversion is the inverse of the `opentelemetry` source decoding. Fields with no OTLP equivalent are +sent as log record attributes. Events that already have the OTLP structure are sent as they are. + +authors: thomasqueirozb diff --git a/lib/codecs/src/encoding/format/otlp.rs b/lib/codecs/src/encoding/format/otlp.rs index 1065eacee1957..ebbe30417acc3 100644 --- a/lib/codecs/src/encoding/format/otlp.rs +++ b/lib/codecs/src/encoding/format/otlp.rs @@ -1,5 +1,6 @@ use crate::encoding::ProtobufSerializer; use bytes::BytesMut; +use opentelemetry_proto::logs::log_event_to_export_request; use opentelemetry_proto::metrics::metric_event_to_export_request; use opentelemetry_proto::proto::{ DESCRIPTOR_BYTES, LOGS_REQUEST_MESSAGE_TYPE, METRICS_REQUEST_MESSAGE_TYPE, @@ -51,14 +52,15 @@ impl OtlpSerializerConfig { /// /// # Implementation approach /// -/// This serializer converts Vector's internal event representation to the appropriate OTLP message type -/// based on the top-level field in the event: +/// This serializer converts Vector's internal event representation to the appropriate OTLP message type. +/// Events that already have the OTLP structure are encoded as they are, based on the top-level field: /// - `resourceLogs` → `ExportLogsServiceRequest` /// - `resourceMetrics` → `ExportMetricsServiceRequest` /// - `resourceSpans` → `ExportTraceServiceRequest` /// -/// The implementation is the inverse of what the `opentelemetry` source does when decoding, -/// ensuring round-trip compatibility. +/// Native Vector logs and metrics are converted to `ExportLogsServiceRequest` and +/// `ExportMetricsServiceRequest`. The conversions are the inverse of what the `opentelemetry` +/// source does when decoding, ensuring round-trip compatibility. #[derive(Debug, Clone)] #[allow(dead_code)] // Fields will be used once encoding is implemented pub struct OtlpSerializer { @@ -119,11 +121,12 @@ impl Encoder for OtlpSerializer { } else if log.contains(event_path!(RESOURCE_METRICS_JSON_FIELD)) { // Currently the OTLP metrics are Vector logs (not metrics). self.metrics_descriptor.encode(Event::Log(log), buffer) + } else if log.contains(event_path!(RESOURCE_SPANS_JSON_FIELD)) { + // OTLP-structured traces can be log events, for example when read as JSON. + self.traces_descriptor.encode(Event::Log(log), buffer) } else { - Err(format!( - "Log event does not contain OTLP top-level fields ({RESOURCE_LOGS_JSON_FIELD} or {RESOURCE_METRICS_JSON_FIELD})", - ) - .into()) + let request = log_event_to_export_request(log); + request.encode(buffer).map_err(Into::into) } } Event::Trace(trace) => { @@ -194,6 +197,37 @@ mod tests { assert_eq!(metric.clone(), round_trip_metric(metric)); } + #[test] + fn native_log_round_trips_through_otlp_source_decoding() { + use opentelemetry_proto::proto::collector::logs::v1::ExportLogsServiceRequest; + use vector_core::{config::LogNamespace, event::LogEvent}; + + let mut log = LogEvent::from("disk full"); + log.insert(event_path!("host"), "web-1"); + log.insert(event_path!("timestamp"), Utc.timestamp_nanos(1_000_000_000)); + + let mut buffer = BytesMut::new(); + OtlpSerializer::new() + .unwrap() + .encode(Event::Log(log), &mut buffer) + .expect("native log must be converted, not rejected"); + + let request = ExportLogsServiceRequest::decode(buffer.freeze()).unwrap(); + let mut events: Vec = request + .resource_logs + .into_iter() + .flat_map(|logs| logs.into_event_iter(LogNamespace::Legacy)) + .collect(); + assert_eq!(events.len(), 1); + let decoded = events.remove(0).into_log(); + assert_eq!(decoded["message"], "disk full".into()); + assert_eq!(decoded["attributes.host"], "web-1".into()); + assert_eq!( + decoded["timestamp"], + Utc.timestamp_nanos(1_000_000_000).into() + ); + } + #[test] fn round_trip_gauge() { let metric = with_empty_tags( diff --git a/lib/opentelemetry-proto/src/common.rs b/lib/opentelemetry-proto/src/common.rs index e948aea127ab7..6c9ef37942e34 100644 --- a/lib/opentelemetry-proto/src/common.rs +++ b/lib/opentelemetry-proto/src/common.rs @@ -1,9 +1,12 @@ use bytes::Bytes; +use chrono::SecondsFormat; use ordered_float::NotNan; use vector_core::event::metric::{TagValue, TagValueSet}; use vrl::value::{ObjectMap, Value}; -use super::proto::common::v1::{AnyValue, ArrayValue, KeyValue, any_value::Value as PBValue}; +use super::proto::common::v1::{ + AnyValue, ArrayValue, KeyValue, KeyValueList, any_value::Value as PBValue, +}; impl From for Value { fn from(av: PBValue) -> Self { @@ -24,6 +27,40 @@ impl From for Value { } } +/// Inverse of `From for Value`. Strings and byte strings that are valid UTF-8 become +/// `string_value`; other byte strings become `bytes_value`. Timestamps and regexes have no +/// OTLP equivalent and are encoded as strings. `Null` becomes an empty `AnyValue`. +impl From for AnyValue { + fn from(value: Value) -> Self { + let value = match value { + Value::Bytes(bytes) => string_or_bytes(bytes), + Value::String(string) => string_or_bytes(string.into_bytes()), + Value::Regex(regex) => PBValue::StringValue(regex.as_str().to_owned()), + Value::Integer(int) => PBValue::IntValue(int), + Value::Float(float) => PBValue::DoubleValue(float.into_inner()), + Value::Boolean(boolean) => PBValue::BoolValue(boolean), + Value::Timestamp(timestamp) => { + PBValue::StringValue(timestamp.to_rfc3339_opts(SecondsFormat::AutoSi, true)) + } + Value::Object(object) => PBValue::KvlistValue(KeyValueList { + values: object_into_kv_list(object), + }), + Value::Array(array) => PBValue::ArrayValue(ArrayValue { + values: array.into_iter().map(Into::into).collect(), + }), + Value::Null => return Self { value: None }, + }; + Self { value: Some(value) } + } +} + +fn string_or_bytes(bytes: Bytes) -> PBValue { + match String::from_utf8(Vec::from(bytes)) { + Ok(string) => PBValue::StringValue(string), + Err(error) => PBValue::BytesValue(error.into_bytes()), + } +} + impl From for TagValue { fn from(pb: PBValue) -> Self { match pb { @@ -80,6 +117,18 @@ pub fn kv_list_into_value(arr: Vec) -> Value { ) } +/// Inverse of [`kv_list_into_value`]. +#[must_use] +pub fn object_into_kv_list(object: ObjectMap) -> Vec { + object + .into_iter() + .map(|(key, value)| KeyValue { + key: key.into(), + value: Some(value.into()), + }) + .collect() +} + #[must_use] pub fn to_hex(d: &[u8]) -> String { if d.is_empty() { diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index d3e2b9b44d6da..c34014b6a462e 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -4,12 +4,13 @@ use vector_core::{ config::{LegacyKey, LogNamespace, log_schema}, event::{Event, LogEvent}, }; -use vrl::{core::Value, path}; +use vrl::{core::Value, path, path::ValuePath, value::ObjectMap}; -use super::common::{kv_list_into_value, to_hex}; +use super::common::{kv_list_into_value, object_into_kv_list, to_hex}; use crate::proto::{ - common::v1::{InstrumentationScope, any_value::Value as PBValue}, - logs::v1::{LogRecord, ResourceLogs, SeverityNumber}, + collector::logs::v1::ExportLogsServiceRequest, + common::v1::{AnyValue, InstrumentationScope, KeyValue, any_value::Value as PBValue}, + logs::v1::{LogRecord, ResourceLogs, ScopeLogs, SeverityNumber}, resource::v1::Resource, }; @@ -26,6 +27,7 @@ pub const SEVERITY_NUMBER_KEY: &str = "severity_number"; pub const OBSERVED_TIMESTAMP_KEY: &str = "observed_timestamp"; pub const DROPPED_ATTRIBUTES_COUNT_KEY: &str = "dropped_attributes_count"; pub const FLAGS_KEY: &str = "flags"; +pub const TIMESTAMP_KEY: &str = "timestamp"; impl ResourceLogs { pub fn into_event_iter(self, log_namespace: LogNamespace) -> impl Iterator { @@ -232,7 +234,7 @@ impl ResourceLog { SOURCE_NAME, &mut log, log_schema().timestamp_key().map(LegacyKey::Overwrite), - path!("timestamp"), + path!(TIMESTAMP_KEY), timestamp, ); @@ -251,3 +253,410 @@ impl ResourceLog { log.into() } } + +/// Convert a native Vector log event into an OTLP export request. +/// +/// This is the inverse of the `opentelemetry` source decoding, so a log record decoded by +/// that source encodes back to an equivalent OTLP log record. +/// +/// - Legacy namespace: the message field becomes the body and the timestamp field becomes +/// `time_unix_nano`. The fields written by the source (`attributes`, `resources`, `scope`, +/// `trace_id`, `span_id`, `severity_text`, `severity_number`, `flags`, +/// `dropped_attributes_count`, and `observed_timestamp`) fill the matching OTLP fields. +/// - Vector namespace: the event value becomes the body. The same fields are read from the +/// `opentelemetry` source metadata. When they are missing, the timestamp falls back to the +/// field with the `timestamp` meaning and the observed timestamp to the Vector ingest +/// timestamp. +/// +/// A field that does not have the type OTLP requires (for example a `trace_id` that is not +/// 32 hex characters) is not dropped: it is sent as a log record attribute. All other event +/// fields are also sent as log record attributes; when a key is in both, the value from +/// `attributes` is used. The `source_type` field is internal to Vector and is not sent. +#[must_use] +pub fn log_event_to_export_request(log: LogEvent) -> ExportLogsServiceRequest { + let mut record = LogRecord::default(); + + let mut fields = match log.namespace() { + LogNamespace::Vector => { + let meaning_time = log + .get_timestamp() + .and_then(Value::as_timestamp) + .and_then(timestamp_nanos); + let (body, mut metadata) = log.into_parts(); + record.body = into_body(body); + let ingest_time = metadata + .value() + .get(path!("vector", "ingest_timestamp")) + .and_then(Value::as_timestamp) + .and_then(timestamp_nanos); + let mut fields = match metadata.value_mut().remove(path!(SOURCE_NAME), false) { + Some(fields @ Value::Object(_)) => fields, + _ => Value::Object(ObjectMap::new()), + }; + record.time_unix_nano = take(&mut fields, path!(TIMESTAMP_KEY), into_timestamp_nanos) + .or(meaning_time) + .unwrap_or_default(); + record.observed_time_unix_nano = take( + &mut fields, + path!(OBSERVED_TIMESTAMP_KEY), + into_timestamp_nanos, + ) + .or(ingest_time) + .unwrap_or_default(); + fields + } + LogNamespace::Legacy => { + let schema = log_schema(); + let (mut fields, _) = log.into_parts(); + if let Some(path) = schema.message_key() { + record.body = fields.remove(path, true).and_then(into_body); + } + if let Some(path) = schema.source_type_key() { + fields.remove(path, true); + } + if let Some(path) = schema.timestamp_key() { + record.time_unix_nano = + take(&mut fields, path, into_timestamp_nanos).unwrap_or_default(); + } + record.observed_time_unix_nano = take( + &mut fields, + path!(OBSERVED_TIMESTAMP_KEY), + into_timestamp_nanos, + ) + .unwrap_or_default(); + fields + } + }; + + let (resource, scope) = take_record_fields(&mut fields, &mut record); + ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(resource), + scope_logs: vec![ScopeLogs { + scope: Some(scope), + log_records: vec![record], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } +} + +/// Move the namespace-independent OTLP fields from `fields` into `record` and return the +/// resource and scope. All fields that remain become log record attributes. +fn take_record_fields( + fields: &mut Value, + record: &mut LogRecord, +) -> (Resource, InstrumentationScope) { + record.severity_text = take(fields, path!(SEVERITY_TEXT_KEY), into_string).unwrap_or_default(); + record.severity_number = + take(fields, path!(SEVERITY_NUMBER_KEY), into_severity_number).unwrap_or_default(); + record.flags = take(fields, path!(FLAGS_KEY), into_u32).unwrap_or_default(); + record.dropped_attributes_count = + take(fields, path!(DROPPED_ATTRIBUTES_COUNT_KEY), into_u32).unwrap_or_default(); + record.trace_id = take(fields, path!(TRACE_ID_KEY), into_id::<16>).unwrap_or_default(); + record.span_id = take(fields, path!(SPAN_ID_KEY), into_id::<8>).unwrap_or_default(); + record.attributes = take(fields, path!(ATTRIBUTES_KEY), into_kv_list).unwrap_or_default(); + + let resource = Resource { + attributes: take(fields, path!(RESOURCE_KEY), into_kv_list).unwrap_or_default(), + ..Default::default() + }; + let scope = InstrumentationScope { + name: take(fields, path!(SCOPE_KEY, NAME_KEY), into_string).unwrap_or_default(), + version: take(fields, path!(SCOPE_KEY, VERSION_KEY), into_string).unwrap_or_default(), + attributes: take(fields, path!(SCOPE_KEY, ATTRIBUTES_KEY), into_kv_list) + .unwrap_or_default(), + dropped_attributes_count: take( + fields, + path!(SCOPE_KEY, DROPPED_ATTRIBUTES_COUNT_KEY), + into_u32, + ) + .unwrap_or_default(), + }; + + match std::mem::replace(fields, Value::Null) { + Value::Object(rest) => { + for (key, value) in rest { + if !record.attributes.iter().any(|kv| kv.key == key.as_str()) { + record.attributes.push(KeyValue { + key: key.into(), + value: Some(value.into()), + }); + } + } + } + // A Legacy event whose root is not an object has no message field; the root is the body. + other if record.body.is_none() => record.body = into_body(other), + _ => {} + } + + (resource, scope) +} + +/// Remove the value at `path` and convert it. If the conversion fails, the value is put back +/// so that it is sent as an attribute instead of being lost. +fn take<'a, T>( + fields: &mut Value, + path: impl ValuePath<'a> + Copy, + convert: fn(Value) -> Result, +) -> Option { + let value = fields.remove(path, true)?; + match convert(value) { + Ok(converted) => Some(converted), + Err(value) => { + fields.insert(path, value); + None + } + } +} + +fn into_body(value: Value) -> Option { + (!value.is_null()).then(|| value.into()) +} + +fn timestamp_nanos(timestamp: &DateTime) -> Option { + timestamp + .timestamp_nanos_opt() + .and_then(|nanos| u64::try_from(nanos).ok()) +} + +fn into_timestamp_nanos(value: Value) -> Result { + match value { + Value::Timestamp(timestamp) => { + timestamp_nanos(×tamp).ok_or(Value::Timestamp(timestamp)) + } + other => Err(other), + } +} + +/// The raw bytes of a string value, or the value itself if it is not a string. +fn into_string_bytes(value: Value) -> Result { + match value { + Value::Bytes(bytes) => Ok(bytes), + Value::String(string) => Ok(string.into_bytes()), + other => Err(other), + } +} + +fn into_string(value: Value) -> Result { + String::from_utf8(Vec::from(into_string_bytes(value)?)) + .map_err(|error| Value::Bytes(error.into_bytes().into())) +} + +fn into_u32(value: Value) -> Result { + match value { + Value::Integer(int) => u32::try_from(int).map_err(|_| Value::Integer(int)), + other => Err(other), + } +} + +fn into_severity_number(value: Value) -> Result { + match value { + Value::Integer(int) => i32::try_from(int) + .ok() + .filter(|number| SeverityNumber::try_from(*number).is_ok()) + .ok_or(Value::Integer(int)), + other => Err(other), + } +} + +/// Decode a hex ID of `LEN` bytes, as written by the source. An empty string means "no ID". +fn into_id(value: Value) -> Result, Value> { + let hex = into_string_bytes(value)?; + match hex::decode(&hex) { + Ok(id) if id.is_empty() || id.len() == LEN => Ok(id), + _ => Err(Value::Bytes(hex)), + } +} + +fn into_kv_list(value: Value) -> Result, Value> { + match value { + Value::Object(object) => Ok(object_into_kv_list(object)), + other => Err(other), + } +} + +#[cfg(test)] +mod tests { + use chrono::{TimeZone, Utc}; + use vrl::event_path; + + use super::*; + use crate::proto::common::v1::{ArrayValue, KeyValueList}; + + fn kv(key: &str, value: PBValue) -> KeyValue { + KeyValue { + key: key.to_owned(), + value: Some(AnyValue { value: Some(value) }), + } + } + + fn string(value: &str) -> PBValue { + PBValue::StringValue(value.to_owned()) + } + + fn request( + resource: Resource, + scope: InstrumentationScope, + record: LogRecord, + ) -> ExportLogsServiceRequest { + ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(resource), + scope_logs: vec![ScopeLogs { + scope: Some(scope), + log_records: vec![record], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } + } + + /// Every field the source decodes, with attribute keys in sorted order because the + /// decoded event stores attributes in a sorted map. + fn full_request() -> ExportLogsServiceRequest { + request( + Resource { + attributes: vec![kv("service.name", string("api"))], + ..Default::default() + }, + InstrumentationScope { + name: "lib".to_owned(), + version: "1.2.3".to_owned(), + attributes: vec![kv("scope.attr", PBValue::BoolValue(true))], + dropped_attributes_count: 1, + }, + LogRecord { + time_unix_nano: 1_700_000_000_000_000_001, + observed_time_unix_nano: 1_700_000_000_000_000_002, + severity_number: SeverityNumber::Warn as i32, + severity_text: "WARN".to_owned(), + body: Some(AnyValue { + value: Some(string("hello")), + }), + attributes: vec![ + kv("count", PBValue::IntValue(3)), + kv( + "list", + PBValue::ArrayValue(ArrayValue { + values: vec![AnyValue { + value: Some(string("a")), + }], + }), + ), + kv( + "nested", + PBValue::KvlistValue(KeyValueList { + values: vec![kv("inner", PBValue::DoubleValue(0.5))], + }), + ), + ], + dropped_attributes_count: 2, + flags: 1, + trace_id: (1..=16).collect(), + span_id: (1..=8).collect(), + }, + ) + } + + fn round_trip(namespace: LogNamespace) { + let expected = full_request(); + let mut events: Vec = expected.resource_logs[0] + .clone() + .into_event_iter(namespace) + .collect(); + assert_eq!(events.len(), 1); + let log = events.pop().unwrap().into_log(); + assert_eq!(log.namespace(), namespace); + assert_eq!(log_event_to_export_request(log), expected); + } + + #[test] + fn decoded_legacy_log_round_trips() { + round_trip(LogNamespace::Legacy); + } + + #[test] + fn decoded_vector_namespace_log_round_trips() { + round_trip(LogNamespace::Vector); + } + + #[test] + fn native_legacy_log_maps_message_timestamp_and_extra_fields() { + let timestamp = Utc.timestamp_nanos(1_700_000_000_000_000_000); + let mut log = LogEvent::from("disk full"); + log.insert(event_path!("timestamp"), timestamp); + log.insert(event_path!("host"), "web-1"); + log.insert(event_path!("source_type"), "file"); + + let expected = request( + Resource::default(), + InstrumentationScope::default(), + LogRecord { + time_unix_nano: 1_700_000_000_000_000_000, + body: Some(AnyValue { + value: Some(string("disk full")), + }), + attributes: vec![kv("host", string("web-1"))], + ..Default::default() + }, + ); + assert_eq!(log_event_to_export_request(log), expected); + } + + #[test] + fn fields_with_invalid_otlp_values_are_kept_as_attributes() { + let mut log = LogEvent::from("message"); + log.insert(event_path!("trace_id"), "not-hex"); + log.insert(event_path!("span_id"), "0102"); + log.insert(event_path!("severity_number"), 99); + log.insert(event_path!("attributes", "host"), "from-attributes"); + log.insert(event_path!("host"), "from-root"); + + let request = log_event_to_export_request(log); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert!(record.trace_id.is_empty()); + assert!(record.span_id.is_empty()); + assert_eq!(record.severity_number, 0); + // Keys from `attributes` come first and win over top-level fields with the same key. + assert_eq!( + record.attributes, + vec![ + kv("host", string("from-attributes")), + kv("severity_number", PBValue::IntValue(99)), + kv("span_id", string("0102")), + kv("trace_id", string("not-hex")), + ] + ); + } + + #[test] + fn native_vector_namespace_log_uses_event_as_body_and_ingest_time_as_observed() { + let mut log = LogEvent::from(Value::from(ObjectMap::from([( + "msg".into(), + Value::from("hi"), + )]))); + let metadata = log.metadata_mut().value_mut(); + metadata.insert(path!("vector", "ingest_timestamp"), Utc.timestamp_nanos(5)); + metadata.insert(path!("vector", "source_type"), "demo_logs"); + metadata.insert(path!("demo_logs", "service"), "vector"); + assert_eq!(log.namespace(), LogNamespace::Vector); + + let expected = request( + Resource::default(), + InstrumentationScope::default(), + LogRecord { + observed_time_unix_nano: 5, + body: Some(AnyValue { + value: Some(PBValue::KvlistValue(KeyValueList { + values: vec![kv("msg", string("hi"))], + })), + }), + ..Default::default() + }, + ); + assert_eq!(log_event_to_export_request(log), expected); + } +} diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index 507d984002b4b..982eb0c23e3e3 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -30,7 +30,13 @@ components: sinks: opentelemetry: { } support: { - requirements: ["This sink accepts events conforming to the [OTEL proto format](\(urls.opentelemetry_proto)). You can use [Remap](\(urls.vector_remap_transform)) to prepare events for ingestion."] + requirements: [""" + With `encoding.codec: otlp`, native Vector logs and metrics are converted to the + [OTEL proto format](\(urls.opentelemetry_proto)), and events that already have the OTLP + structure (for example from the `opentelemetry` source with `use_otlp_decoding`) are sent as + they are. Trace events must already have the OTLP structure. With other codecs, you can use + [Remap](\(urls.vector_remap_transform)) to prepare events for ingestion. + """] warnings: [ """ Batching only works with `encoding.codec: otlp`, which encodes events as protobuf @@ -45,6 +51,37 @@ components: sinks: opentelemetry: { configuration: generated.components.sinks.opentelemetry.configuration how_it_works: { + native_log_conversion: { + title: "Native log conversion" + body: """ + With `encoding.codec: otlp`, a log event without a `resourceLogs` field is converted + to one OTLP log record. The conversion is the inverse of the `opentelemetry` source + decoding, so logs that the source decodes without `use_otlp_decoding` are sent back + unchanged. + + | Event field (Legacy namespace) | OTLP field | + | --- | --- | + | `message` | `body` | + | `timestamp` | `timeUnixNano` | + | `observed_timestamp` | `observedTimeUnixNano` | + | `attributes` | `attributes` | + | `resources` | `resource.attributes` | + | `scope.name`, `scope.version`, `scope.attributes`, `scope.dropped_attributes_count` | `scope` | + | `trace_id`, `span_id` (hex strings) | `traceId`, `spanId` | + | `severity_text`, `severity_number` | `severityText`, `severityNumber` | + | `flags`, `dropped_attributes_count` | `flags`, `droppedAttributesCount` | + + With the Vector log namespace, the event becomes the `body` and the other fields are + read from the `opentelemetry` source metadata. If that metadata is missing, the field + with the `timestamp` meaning gives `timeUnixNano`, and the Vector ingest timestamp + gives `observedTimeUnixNano`. + + All other event fields are sent as log record attributes. This includes a mapped field + that does not have the type OTLP requires, for example a `trace_id` that is not 32 hex + characters. If a key is both in `attributes` and at the top level, the value from + `attributes` is used. The `source_type` field is not sent. + """ + } quickstart: { title: "Quickstart" body: """ From e5749f74c5d9b4d1a4120554afe280f1be37708a Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 11:17:36 -0400 Subject: [PATCH 02/13] fix(codecs): read Legacy OTLP log body from metadata-prefixed message key --- lib/opentelemetry-proto/src/logs.rs | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index c34014b6a462e..786fd40c3b4c0 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -273,7 +273,7 @@ impl ResourceLog { /// fields are also sent as log record attributes; when a key is in both, the value from /// `attributes` is used. The `source_type` field is internal to Vector and is not sent. #[must_use] -pub fn log_event_to_export_request(log: LogEvent) -> ExportLogsServiceRequest { +pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceRequest { let mut record = LogRecord::default(); let mut fields = match log.namespace() { @@ -307,10 +307,12 @@ pub fn log_event_to_export_request(log: LogEvent) -> ExportLogsServiceRequest { } LogNamespace::Legacy => { let schema = log_schema(); - let (mut fields, _) = log.into_parts(); - if let Some(path) = schema.message_key() { - record.body = fields.remove(path, true).and_then(into_body); + // The message key can point into metadata (for example `%message`), so remove it + // with its full target path before the metadata is discarded. + if let Some(path) = schema.message_key_target_path() { + record.body = log.remove_prune(path, true).and_then(into_body); } + let (mut fields, _) = log.into_parts(); if let Some(path) = schema.source_type_key() { fields.remove(path, true); } @@ -480,7 +482,8 @@ fn into_kv_list(value: Value) -> Result, Value> { #[cfg(test)] mod tests { use chrono::{TimeZone, Utc}; - use vrl::event_path; + use vector_core::config::{LogSchema, init_log_schema}; + use vrl::{event_path, owned_value_path, path::OwnedTargetPath}; use super::*; use crate::proto::common::v1::{ArrayValue, KeyValueList}; @@ -583,6 +586,17 @@ mod tests { round_trip(LogNamespace::Vector); } + #[test] + fn decoded_legacy_log_with_metadata_message_key_round_trips() { + let mut schema = LogSchema::default(); + schema.set_message_key(Some(OwnedTargetPath::metadata(owned_value_path!( + "message" + )))); + init_log_schema(schema, true); + + round_trip(LogNamespace::Legacy); + } + #[test] fn native_legacy_log_maps_message_timestamp_and_extra_fields() { let timestamp = Utc.timestamp_nanos(1_700_000_000_000_000_000); From 48f4e0c219ddf0eaa7ab86599e6b8157e3ff69b6 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 11:18:47 -0400 Subject: [PATCH 03/13] docs(opentelemetry sink): note schema URLs are not kept in native log round trips --- lib/opentelemetry-proto/src/logs.rs | 4 +++- website/cue/reference/components/sinks/opentelemetry.cue | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 786fd40c3b4c0..2caa6ab49f6d7 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -257,7 +257,9 @@ impl ResourceLog { /// Convert a native Vector log event into an OTLP export request. /// /// This is the inverse of the `opentelemetry` source decoding, so a log record decoded by -/// that source encodes back to an equivalent OTLP log record. +/// that source encodes back to an equivalent OTLP log record. The source does not keep the +/// resource and scope `schema_url` or the resource `dropped_attributes_count`, so these +/// fields are empty. /// /// - Legacy namespace: the message field becomes the body and the timestamp field becomes /// `time_unix_nano`. The fields written by the source (`attributes`, `resources`, `scope`, diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index 982eb0c23e3e3..f7cbdd4342f9e 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -57,7 +57,9 @@ components: sinks: opentelemetry: { With `encoding.codec: otlp`, a log event without a `resourceLogs` field is converted to one OTLP log record. The conversion is the inverse of the `opentelemetry` source decoding, so logs that the source decodes without `use_otlp_decoding` are sent back - unchanged. + with the same log record, resource attributes, and scope. The source does not keep the + resource and scope `schemaUrl` or the resource `droppedAttributesCount`, so these fields + are empty. To keep all OTLP fields, use `use_otlp_decoding` on the source. | Event field (Legacy namespace) | OTLP field | | --- | --- | From 39eae8a924e8c3704ba44a97a10bd083cecdbdb7 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 13:26:24 -0400 Subject: [PATCH 04/13] docs(opentelemetry sink): document bytes and missing timestamp changes in native log round trips --- lib/opentelemetry-proto/src/logs.rs | 11 ++++++++--- .../reference/components/sinks/opentelemetry.cue | 13 ++++++++++--- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 2caa6ab49f6d7..65fb31ed0f3a6 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -257,9 +257,14 @@ impl ResourceLog { /// Convert a native Vector log event into an OTLP export request. /// /// This is the inverse of the `opentelemetry` source decoding, so a log record decoded by -/// that source encodes back to an equivalent OTLP log record. The source does not keep the -/// resource and scope `schema_url` or the resource `dropped_attributes_count`, so these -/// fields are empty. +/// that source encodes back to an equivalent OTLP log record, with these exceptions: +/// +/// - The source does not keep the resource and scope `schema_url` or the resource +/// `dropped_attributes_count`, so these fields are empty. +/// - A `bytes_value` that is valid UTF-8 becomes a `string_value`, because the source decodes +/// both to the same Vector value. +/// - A record without `time_unix_nano` gets the observed time, because the source uses the +/// observed time as the event timestamp. /// /// - Legacy namespace: the message field becomes the body and the timestamp field becomes /// `time_unix_nano`. The fields written by the source (`attributes`, `resources`, `scope`, diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index f7cbdd4342f9e..aaabdabd9cae7 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -57,9 +57,16 @@ components: sinks: opentelemetry: { With `encoding.codec: otlp`, a log event without a `resourceLogs` field is converted to one OTLP log record. The conversion is the inverse of the `opentelemetry` source decoding, so logs that the source decodes without `use_otlp_decoding` are sent back - with the same log record, resource attributes, and scope. The source does not keep the - resource and scope `schemaUrl` or the resource `droppedAttributesCount`, so these fields - are empty. To keep all OTLP fields, use `use_otlp_decoding` on the source. + with the same log record, resource attributes, and scope, with these exceptions: + + - The source does not keep the resource and scope `schemaUrl` or the resource + `droppedAttributesCount`, so these fields are empty. + - A `bytesValue` body or attribute that is valid UTF-8 is sent as a `stringValue`. + - A record without `timeUnixNano` is sent with `timeUnixNano` set to the observed + time, because the source uses the observed time as the event timestamp. + + To send OTLP logs from the source exactly as received, use `use_otlp_decoding` on the + source. | Event field (Legacy namespace) | OTLP field | | --- | --- | From 05493b575a142f0b542a67e28f7cf060da920e5d Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 13:29:53 -0400 Subject: [PATCH 05/13] docs(codecs): add native log example to otlp changelog --- .../otlp_native_log_encoding.feature.md | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/changelog.d/otlp_native_log_encoding.feature.md b/changelog.d/otlp_native_log_encoding.feature.md index 549efd18ad518..5066127630db1 100644 --- a/changelog.d/otlp_native_log_encoding.feature.md +++ b/changelog.d/otlp_native_log_encoding.feature.md @@ -1,6 +1,31 @@ The `otlp` codec now converts native Vector log events to OTLP log records, so logs from any source can be sent with the `opentelemetry` sink without a `remap` transform that builds the `resourceLogs` structure. +For example, this log event from the `file` source: + +```yaml +message: disk full +host: web-1 +timestamp: 2026-10-09T12:00:00Z +source_type: file +``` + +is sent as this OTLP log record (shown as YAML, without default fields): + +```yaml +resourceLogs: + - scopeLogs: + - logRecords: + - timeUnixNano: "1791547200000000000" + body: + stringValue: disk full + attributes: + - key: host + value: + stringValue: web-1 +``` + The conversion is the inverse of the `opentelemetry` source decoding. Fields with no OTLP equivalent are -sent as log record attributes. Events that already have the OTLP structure are sent as they are. +sent as log record attributes. A log event with a `resourceLogs`, `resourceMetrics`, or `resourceSpans` +root field is treated as an OTLP request that is already built, and is sent as it is. authors: thomasqueirozb From d74d78f69a407fcaa10d905fc27c9ae729d8b695 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 13:29:54 -0400 Subject: [PATCH 06/13] docs(opentelemetry sink): document reserved OTLP root fields --- lib/codecs/src/encoding/format/otlp.rs | 3 ++ .../components/sinks/opentelemetry.cue | 38 +++++++++++++++++-- 2 files changed, 37 insertions(+), 4 deletions(-) diff --git a/lib/codecs/src/encoding/format/otlp.rs b/lib/codecs/src/encoding/format/otlp.rs index ebbe30417acc3..c8c8874b68de2 100644 --- a/lib/codecs/src/encoding/format/otlp.rs +++ b/lib/codecs/src/encoding/format/otlp.rs @@ -58,6 +58,9 @@ impl OtlpSerializerConfig { /// - `resourceMetrics` → `ExportMetricsServiceRequest` /// - `resourceSpans` → `ExportTraceServiceRequest` /// +/// The first field in this list that the event has is used, and all other event fields are +/// not encoded. +/// /// Native Vector logs and metrics are converted to `ExportLogsServiceRequest` and /// `ExportMetricsServiceRequest`. The conversions are the inverse of what the `opentelemetry` /// source does when decoding, ensuring round-trip compatibility. diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index aaabdabd9cae7..c46292cfd0936 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -51,13 +51,43 @@ components: sinks: opentelemetry: { configuration: generated.components.sinks.opentelemetry.configuration how_it_works: { + otlp_root_fields: { + title: "Events with the OTLP structure" + body: """ + With `encoding.codec: otlp`, the root fields `resourceLogs`, `resourceMetrics`, and + `resourceSpans` are reserved. A log event with one of these fields is treated as an + OTLP request that is already built: + + | Root field | Encoded as | + | --- | --- | + | `resourceLogs` | `ExportLogsServiceRequest` | + | `resourceMetrics` | `ExportMetricsServiceRequest` | + | `resourceSpans` | `ExportTraceServiceRequest` | + + If an event has more than one of these fields, the first field in the table is used. + Only the fields of that request are encoded, and all other event fields are not sent. + If the value does not have the OTLP structure, the encoding fails. + + Vector does not examine the value to find if it is OTLP data. For example, this event + is sent as an empty OTLP logs request, and `message` is not sent: + + ```yaml + message: finished checking resources + resourceLogs: [] + ``` + + To send an event like this as a log record, rename the field before the sink, for + example with a `remap` transform. + """ + } native_log_conversion: { title: "Native log conversion" body: """ - With `encoding.codec: otlp`, a log event without a `resourceLogs` field is converted - to one OTLP log record. The conversion is the inverse of the `opentelemetry` source - decoding, so logs that the source decodes without `use_otlp_decoding` are sent back - with the same log record, resource attributes, and scope, with these exceptions: + With `encoding.codec: otlp`, a log event without a `resourceLogs`, `resourceMetrics`, + or `resourceSpans` root field is converted to one OTLP log record. The conversion is + the inverse of the `opentelemetry` source decoding, so logs that the source decodes + without `use_otlp_decoding` are sent back with the same log record, resource + attributes, and scope, with these exceptions: - The source does not keep the resource and scope `schemaUrl` or the resource `droppedAttributesCount`, so these fields are empty. From 984a1b62bb9f809d8b08a2c4ba7f673f8eb1fbcd Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 15:04:09 -0400 Subject: [PATCH 07/13] docs(opentelemetry sink): note native log records are not regrouped by resource and scope --- lib/opentelemetry-proto/src/logs.rs | 2 ++ website/cue/reference/components/sinks/opentelemetry.cue | 2 ++ 2 files changed, 4 insertions(+) diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 65fb31ed0f3a6..0bd2eac0f1154 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -265,6 +265,8 @@ impl ResourceLog { /// both to the same Vector value. /// - A record without `time_unix_nano` gets the observed time, because the source uses the /// observed time as the event timestamp. +/// - The request has one `ResourceLogs` and one `ScopeLogs` for the one record, so records +/// that shared a resource and scope in the original request are not grouped again. /// /// - Legacy namespace: the message field becomes the body and the timestamp field becomes /// `time_unix_nano`. The fields written by the source (`attributes`, `resources`, `scope`, diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index c46292cfd0936..2da0b3d4fffbd 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -94,6 +94,8 @@ components: sinks: opentelemetry: { - A `bytesValue` body or attribute that is valid UTF-8 is sent as a `stringValue`. - A record without `timeUnixNano` is sent with `timeUnixNano` set to the observed time, because the source uses the observed time as the event timestamp. + - Each log record is sent in its own `resourceLogs` and `scopeLogs` entry. Records + that had the same resource and scope in the original request are not grouped again. To send OTLP logs from the source exactly as received, use `use_otlp_decoding` on the source. From 2fae247fde84177598aa12ee81c643f7fa8ce262 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 15:04:47 -0400 Subject: [PATCH 08/13] docs(opentelemetry sink): point to use_otlp_decoding to keep OTLP log grouping --- website/cue/reference/components/sinks/opentelemetry.cue | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index 2da0b3d4fffbd..721c46ded3f90 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -95,10 +95,12 @@ components: sinks: opentelemetry: { - A record without `timeUnixNano` is sent with `timeUnixNano` set to the observed time, because the source uses the observed time as the event timestamp. - Each log record is sent in its own `resourceLogs` and `scopeLogs` entry. Records - that had the same resource and scope in the original request are not grouped again. + that had the same resource and scope in the original request are not grouped again, + so the resource and scope are repeated for each record and the request is larger. - To send OTLP logs from the source exactly as received, use `use_otlp_decoding` on the - source. + To send OTLP logs from the source exactly as received, with the original grouping, use + `use_otlp_decoding` on the source. With this option, the events have the OTLP structure + instead of the fields in the following table. | Event field (Legacy namespace) | OTLP field | | --- | --- | From f7411498c2ea8c9eb2ea0beb1cd6a42af9b4705a Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 15:33:53 -0400 Subject: [PATCH 09/13] fix(codecs): read Legacy OTLP log timestamp from metadata-prefixed timestamp key --- lib/opentelemetry-proto/src/logs.rs | 61 ++++++++++++++++++++++++++--- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 0bd2eac0f1154..c5140c2a9a3c8 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -4,7 +4,12 @@ use vector_core::{ config::{LegacyKey, LogNamespace, log_schema}, event::{Event, LogEvent}, }; -use vrl::{core::Value, path, path::ValuePath, value::ObjectMap}; +use vrl::{ + core::Value, + path, + path::{PathPrefix, ValuePath}, + value::ObjectMap, +}; use super::common::{kv_list_into_value, object_into_kv_list, to_hex}; use crate::proto::{ @@ -321,14 +326,25 @@ pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceReques if let Some(path) = schema.message_key_target_path() { record.body = log.remove_prune(path, true).and_then(into_body); } + // The timestamp key can also point into metadata. Many sources and decoders write it + // there with its full target path, but the `opentelemetry` source writes it to the + // event root, so read the metadata first and then the event root. + let metadata_time = schema + .timestamp_key_target_path() + .filter(|path| path.prefix == PathPrefix::Metadata) + .and_then(|path| log.remove(path)) + .and_then(|value| into_timestamp_nanos(value).ok()); let (mut fields, _) = log.into_parts(); if let Some(path) = schema.source_type_key() { fields.remove(path, true); } - if let Some(path) = schema.timestamp_key() { - record.time_unix_nano = - take(&mut fields, path, into_timestamp_nanos).unwrap_or_default(); - } + record.time_unix_nano = metadata_time + .or_else(|| { + schema + .timestamp_key() + .and_then(|path| take(&mut fields, path, into_timestamp_nanos)) + }) + .unwrap_or_default(); record.observed_time_unix_nano = take( &mut fields, path!(OBSERVED_TIMESTAMP_KEY), @@ -606,6 +622,41 @@ mod tests { round_trip(LogNamespace::Legacy); } + fn init_metadata_timestamp_key() { + let mut schema = LogSchema::default(); + schema.set_timestamp_key(Some(OwnedTargetPath::metadata(owned_value_path!( + "timestamp" + )))); + init_log_schema(schema, true); + } + + #[test] + fn native_legacy_log_reads_metadata_timestamp_key() { + init_metadata_timestamp_key(); + let mut log = LogEvent::from("disk full"); + log.insert( + log_schema().timestamp_key_target_path().unwrap(), + Utc.timestamp_nanos(1_700_000_000_000_000_000), + ); + log.insert(event_path!("timestamp"), "root field"); + + let request = log_event_to_export_request(log); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert_eq!(record.time_unix_nano, 1_700_000_000_000_000_000); + // With a metadata timestamp key, a `timestamp` event field is an ordinary field. + assert_eq!( + record.attributes, + vec![kv("timestamp", string("root field"))] + ); + } + + #[test] + fn decoded_legacy_log_with_metadata_timestamp_key_round_trips() { + // The source writes the timestamp to the event root even with a metadata key. + init_metadata_timestamp_key(); + round_trip(LogNamespace::Legacy); + } + #[test] fn native_legacy_log_maps_message_timestamp_and_extra_fields() { let timestamp = Utc.timestamp_nanos(1_700_000_000_000_000_000); From 6b94fec7c543008e998ab4fda5e4d31b01510a46 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 16:59:41 -0400 Subject: [PATCH 10/13] fix(codecs): preserve payload source type with metadata markers --- .../otlp_native_log_encoding.feature.md | 20 +++++++-- lib/opentelemetry-proto/src/logs.rs | 42 +++++++++++++++++-- .../components/sinks/opentelemetry.cue | 26 +++++++++--- 3 files changed, 76 insertions(+), 12 deletions(-) diff --git a/changelog.d/otlp_native_log_encoding.feature.md b/changelog.d/otlp_native_log_encoding.feature.md index 5066127630db1..5b2c45be52c25 100644 --- a/changelog.d/otlp_native_log_encoding.feature.md +++ b/changelog.d/otlp_native_log_encoding.feature.md @@ -24,8 +24,22 @@ resourceLogs: stringValue: web-1 ``` -The conversion is the inverse of the `opentelemetry` source decoding. Fields with no OTLP equivalent are -sent as log record attributes. A log event with a `resourceLogs`, `resourceMetrics`, or `resourceSpans` -root field is treated as an OTLP request that is already built, and is sent as it is. +The conversion is the inverse of the `opentelemetry` source decoding. A log event with a `resourceLogs`, +`resourceMetrics`, or `resourceSpans` root field is treated as an OTLP request that is already built, +and is sent as it is. + +The source type marker identifies the Vector source component type that produced the event, for example +`source_type: file`. + +With `log_namespace: false` (Legacy namespace), fields with no OTLP equivalent are sent as log record +attributes. The marker's location is configured with `log_schema.source_type_key` (default: +`.source_type`). The codec omits this internal field. If the configured path points to metadata and +that field exists (for example, `%source_type`), the matching payload field (`.source_type`) is kept as +an attribute. Otherwise, the matching event field is omitted, because some Legacy sources still write +the marker there. + +With `log_namespace: true` (Vector namespace), the entire event payload becomes the OTLP `body`. +A payload field named `.source_type` is preserved in the body, not sent as an attribute. The internal +marker is stored separately in `%vector.source_type` metadata and is not sent. authors: thomasqueirozb diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index c5140c2a9a3c8..0a98f53273495 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -285,7 +285,10 @@ impl ResourceLog { /// A field that does not have the type OTLP requires (for example a `trace_id` that is not /// 32 hex characters) is not dropped: it is sent as a log record attribute. All other event /// fields are also sent as log record attributes; when a key is in both, the value from -/// `attributes` is used. The `source_type` field is internal to Vector and is not sent. +/// `attributes` is used. Vector's source type marker is not sent. In the Legacy namespace, +/// a marker at the configured metadata path takes precedence over the matching event field. +/// If that metadata is absent, the event field is treated as the marker for sources that +/// write it there. #[must_use] pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceRequest { let mut record = LogRecord::default(); @@ -321,6 +324,11 @@ pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceReques } LogNamespace::Legacy => { let schema = log_schema(); + // A marker at the configured metadata path leaves the event field as user data. + // Otherwise, remove the root marker for Legacy sources that ignore the prefix. + let source_type_key = schema + .source_type_key_target_path() + .filter(|path| path.prefix == PathPrefix::Event || !log.contains(*path)); // The message key can point into metadata (for example `%message`), so remove it // with its full target path before the metadata is discarded. if let Some(path) = schema.message_key_target_path() { @@ -335,8 +343,8 @@ pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceReques .and_then(|path| log.remove(path)) .and_then(|value| into_timestamp_nanos(value).ok()); let (mut fields, _) = log.into_parts(); - if let Some(path) = schema.source_type_key() { - fields.remove(path, true); + if let Some(path) = source_type_key { + fields.remove(&path.path, true); } record.time_unix_nano = metadata_time .or_else(|| { @@ -657,6 +665,34 @@ mod tests { round_trip(LogNamespace::Legacy); } + fn init_metadata_source_type_key() { + let mut schema = LogSchema::default(); + schema.set_source_type_key(Some(OwnedTargetPath::metadata(owned_value_path!( + "source_type" + )))); + init_log_schema(schema, true); + } + + #[test] + fn native_legacy_log_preserves_payload_with_metadata_source_type() { + init_metadata_source_type_key(); + let mut log = LogEvent::from("disk full"); + log.insert(log_schema().source_type_key_target_path().unwrap(), "kafka"); + // Even when the values match, the event field is not the metadata marker. + log.insert(event_path!("source_type"), "kafka"); + + let request = log_event_to_export_request(log); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert_eq!(record.attributes, vec![kv("source_type", string("kafka"))]); + } + + #[test] + fn decoded_legacy_log_with_metadata_source_type_key_round_trips() { + // The source writes the marker to the event root even with a metadata key. + init_metadata_source_type_key(); + round_trip(LogNamespace::Legacy); + } + #[test] fn native_legacy_log_maps_message_timestamp_and_extra_fields() { let timestamp = Utc.timestamp_nanos(1_700_000_000_000_000_000); diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index 721c46ded3f90..430b9195ec2b4 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -102,6 +102,8 @@ components: sinks: opentelemetry: { `use_otlp_decoding` on the source. With this option, the events have the OTLP structure instead of the fields in the following table. + With `log_namespace: false` (Legacy namespace), the conversion uses these event fields: + | Event field (Legacy namespace) | OTLP field | | --- | --- | | `message` | `body` | @@ -114,15 +116,27 @@ components: sinks: opentelemetry: { | `severity_text`, `severity_number` | `severityText`, `severityNumber` | | `flags`, `dropped_attributes_count` | `flags`, `droppedAttributesCount` | - With the Vector log namespace, the event becomes the `body` and the other fields are - read from the `opentelemetry` source metadata. If that metadata is missing, the field - with the `timestamp` meaning gives `timeUnixNano`, and the Vector ingest timestamp - gives `observedTimeUnixNano`. - All other event fields are sent as log record attributes. This includes a mapped field that does not have the type OTLP requires, for example a `trace_id` that is not 32 hex characters. If a key is both in `attributes` and at the top level, the value from - `attributes` is used. The `source_type` field is not sent. + `attributes` is used. + + Vector's source type marker identifies the source component type that produced the + event, for example `source_type: file`. Its location is configured with + `log_schema.source_type_key` (default: `.source_type`). The codec omits this internal + field. If the configured path points to metadata and that field exists (for example, + `%source_type`), the matching payload field (`.source_type`) is kept as an attribute. + Otherwise, the matching event field is omitted, because some Legacy sources still + write the marker there. + + With `log_namespace: true` (Vector namespace), the entire event payload becomes the + OTLP `body`. A payload field named `.source_type` is preserved in the body, not sent + as an attribute. The internal marker is stored separately in `%vector.source_type` + metadata and is not sent. + + The other OTLP fields are read from the `opentelemetry` source metadata. If that + metadata is missing, the field with the `timestamp` meaning gives `timeUnixNano`, + and the Vector ingest timestamp gives `observedTimeUnixNano`. """ } quickstart: { From 3c57c6a1fb1aad3f51820059f5d7a410abde4810 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 17:50:12 -0400 Subject: [PATCH 11/13] refactor(opentelemetry): use Value::to_str_lossy when encoding OTLP strings --- lib/opentelemetry-proto/src/common.rs | 20 +++++++++----------- lib/opentelemetry-proto/src/logs.rs | 4 ++-- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/lib/opentelemetry-proto/src/common.rs b/lib/opentelemetry-proto/src/common.rs index 6c9ef37942e34..1d670f83d1e8d 100644 --- a/lib/opentelemetry-proto/src/common.rs +++ b/lib/opentelemetry-proto/src/common.rs @@ -27,14 +27,19 @@ impl From for Value { } } -/// Inverse of `From for Value`. Strings and byte strings that are valid UTF-8 become -/// `string_value`; other byte strings become `bytes_value`. Timestamps and regexes have no +/// Inverse of `From for Value`. Strings become `string_value`; byte strings become +/// `string_value` with a lossy UTF-8 decode. Timestamps and regexes have no /// OTLP equivalent and are encoded as strings. `Null` becomes an empty `AnyValue`. impl From for AnyValue { fn from(value: Value) -> Self { let value = match value { - Value::Bytes(bytes) => string_or_bytes(bytes), - Value::String(string) => string_or_bytes(string.into_bytes()), + Value::Bytes(_) => PBValue::StringValue( + value + .to_str_lossy() + .expect("`Value::Bytes` always converts to a string") + .into_owned(), + ), + Value::String(string) => PBValue::StringValue(string.into()), Value::Regex(regex) => PBValue::StringValue(regex.as_str().to_owned()), Value::Integer(int) => PBValue::IntValue(int), Value::Float(float) => PBValue::DoubleValue(float.into_inner()), @@ -54,13 +59,6 @@ impl From for AnyValue { } } -fn string_or_bytes(bytes: Bytes) -> PBValue { - match String::from_utf8(Vec::from(bytes)) { - Ok(string) => PBValue::StringValue(string), - Err(error) => PBValue::BytesValue(error.into_bytes()), - } -} - impl From for TagValue { fn from(pb: PBValue) -> Self { match pb { diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 0a98f53273495..9142315f18bb7 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -266,8 +266,8 @@ impl ResourceLog { /// /// - The source does not keep the resource and scope `schema_url` or the resource /// `dropped_attributes_count`, so these fields are empty. -/// - A `bytes_value` that is valid UTF-8 becomes a `string_value`, because the source decodes -/// both to the same Vector value. +/// - A `bytes_value` becomes a `string_value`, because the source decodes both to the same +/// Vector value. Bytes that are not valid UTF-8 are replaced with U+FFFD. /// - A record without `time_unix_nano` gets the observed time, because the source uses the /// observed time as the event timestamp. /// - The request has one `ResourceLogs` and one `ScopeLogs` for the one record, so records From dafb63d435ed3b55da6da2aa7c0c09141abbafeb Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 18:12:03 -0400 Subject: [PATCH 12/13] test(opentelemetry): enable helpers for isolated crate tests --- lib/opentelemetry-proto/Cargo.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/lib/opentelemetry-proto/Cargo.toml b/lib/opentelemetry-proto/Cargo.toml index b0fe616913726..71fc070796f3c 100644 --- a/lib/opentelemetry-proto/Cargo.toml +++ b/lib/opentelemetry-proto/Cargo.toml @@ -23,3 +23,6 @@ tonic.workspace = true vrl.workspace = true vector-common = { path = "../vector-common", default-features = false } vector-core = { path = "../vector-core", default-features = false } + +[dev-dependencies] +vector-core = { path = "../vector-core", default-features = false, features = ["test"] } From 47b9ab03d4d3dea8443a6b23dd93e361d6e52a96 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 9 Oct 2026 18:25:59 -0400 Subject: [PATCH 13/13] fix(codecs): preserve timestamps nested in log messages --- .../otlp_native_log_encoding.feature.md | 3 +- lib/opentelemetry-proto/src/logs.rs | 101 ++++++++++++++++-- .../components/sinks/opentelemetry.cue | 5 + 3 files changed, 98 insertions(+), 11 deletions(-) diff --git a/changelog.d/otlp_native_log_encoding.feature.md b/changelog.d/otlp_native_log_encoding.feature.md index 5b2c45be52c25..e4063c03384fa 100644 --- a/changelog.d/otlp_native_log_encoding.feature.md +++ b/changelog.d/otlp_native_log_encoding.feature.md @@ -32,7 +32,8 @@ The source type marker identifies the Vector source component type that produced `source_type: file`. With `log_namespace: false` (Legacy namespace), fields with no OTLP equivalent are sent as log record -attributes. The marker's location is configured with `log_schema.source_type_key` (default: +attributes. A timestamp inside the configured message field still sets `timeUnixNano` and remains in +the body. The marker's location is configured with `log_schema.source_type_key` (default: `.source_type`). The codec omits this internal field. If the configured path points to metadata and that field exists (for example, `%source_type`), the matching payload field (`.source_type`) is kept as an attribute. Otherwise, the matching event field is omitted, because some Legacy sources still write diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index 9142315f18bb7..85dd87456cee0 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -329,28 +329,36 @@ pub fn log_event_to_export_request(mut log: LogEvent) -> ExportLogsServiceReques let source_type_key = schema .source_type_key_target_path() .filter(|path| path.prefix == PathPrefix::Event || !log.contains(*path)); + // Read timestamps before removing a message that may contain them. The + // opentelemetry source ignores the timestamp key's metadata prefix, so + // prefer metadata but retain the event-root fallback. + let metadata_time = schema + .timestamp_key_target_path() + .filter(|path| path.prefix == PathPrefix::Metadata) + .and_then(|path| log.get(path)) + .and_then(Value::as_timestamp) + .and_then(timestamp_nanos); + let event_time = schema + .timestamp_key() + .filter(|_| metadata_time.is_none()) + .and_then(|path| log.get((PathPrefix::Event, path))) + .and_then(Value::as_timestamp) + .and_then(timestamp_nanos); // The message key can point into metadata (for example `%message`), so remove it // with its full target path before the metadata is discarded. if let Some(path) = schema.message_key_target_path() { record.body = log.remove_prune(path, true).and_then(into_body); } - // The timestamp key can also point into metadata. Many sources and decoders write it - // there with its full target path, but the `opentelemetry` source writes it to the - // event root, so read the metadata first and then the event root. - let metadata_time = schema - .timestamp_key_target_path() - .filter(|path| path.prefix == PathPrefix::Metadata) - .and_then(|path| log.remove(path)) - .and_then(|value| into_timestamp_nanos(value).ok()); let (mut fields, _) = log.into_parts(); if let Some(path) = source_type_key { fields.remove(&path.path, true); } record.time_unix_nano = metadata_time .or_else(|| { - schema + let extracted_time = schema .timestamp_key() - .and_then(|path| take(&mut fields, path, into_timestamp_nanos)) + .and_then(|path| take(&mut fields, path, into_timestamp_nanos)); + event_time.or(extracted_time) }) .unwrap_or_default(); record.observed_time_unix_nano = take( @@ -630,6 +638,79 @@ mod tests { round_trip(LogNamespace::Legacy); } + #[test] + fn native_legacy_log_reads_timestamp_inside_event_message() { + let mut schema = LogSchema::default(); + schema.set_message_key(Some(OwnedTargetPath::event(owned_value_path!("payload")))); + schema.set_timestamp_key(Some(OwnedTargetPath::event(owned_value_path!( + "payload", + "timestamp" + )))); + init_log_schema(schema, true); + let mut log = LogEvent::default(); + log.insert(event_path!("payload", "message"), "hello"); + log.insert(event_path!("payload", "timestamp"), Utc.timestamp_nanos(5)); + + let request = log_event_to_export_request(log); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert_eq!(record.time_unix_nano, 5); + assert_eq!( + record.body, + Some(AnyValue { + value: Some(PBValue::KvlistValue(KeyValueList { + values: vec![ + kv("message", string("hello")), + kv("timestamp", string("1970-01-01T00:00:00.000000005Z")), + ], + })), + }) + ); + } + + #[test] + fn native_legacy_log_reads_timestamp_inside_metadata_message() { + let mut schema = LogSchema::default(); + schema.set_message_key(Some(OwnedTargetPath::metadata(owned_value_path!( + "payload" + )))); + schema.set_timestamp_key(Some(OwnedTargetPath::metadata(owned_value_path!( + "payload", + "timestamp" + )))); + init_log_schema(schema, true); + let mut log = LogEvent::default(); + log.insert((PathPrefix::Metadata, path!("payload", "message")), "hello"); + log.insert( + log_schema().timestamp_key_target_path().unwrap(), + Utc.timestamp_nanos(5), + ); + log.insert(event_path!("payload", "timestamp"), Utc.timestamp_nanos(9)); + + let request = log_event_to_export_request(log); + let record = &request.resource_logs[0].scope_logs[0].log_records[0]; + assert_eq!(record.time_unix_nano, 5); + assert_eq!( + record.body, + Some(AnyValue { + value: Some(PBValue::KvlistValue(KeyValueList { + values: vec![ + kv("message", string("hello")), + kv("timestamp", string("1970-01-01T00:00:00.000000005Z")), + ], + })), + }) + ); + assert_eq!( + record.attributes, + vec![kv( + "payload", + PBValue::KvlistValue(KeyValueList { + values: vec![kv("timestamp", string("1970-01-01T00:00:00.000000009Z"))], + }), + )] + ); + } + fn init_metadata_timestamp_key() { let mut schema = LogSchema::default(); schema.set_timestamp_key(Some(OwnedTargetPath::metadata(owned_value_path!( diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index 430b9195ec2b4..11744232e171a 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -116,6 +116,11 @@ components: sinks: opentelemetry: { | `severity_text`, `severity_number` | `severityText`, `severityNumber` | | `flags`, `dropped_attributes_count` | `flags`, `droppedAttributesCount` | + If `log_schema.timestamp_key` is inside `log_schema.message_key`, the timestamp sets + `timeUnixNano` and also remains in the body. For example, this applies to `.payload` + as the message key and `.payload.timestamp` as the timestamp key. It also applies + when both keys point to metadata. + All other event fields are sent as log record attributes. This includes a mapped field that does not have the type OTLP requires, for example a `trace_id` that is not 32 hex characters. If a key is both in `attributes` and at the top level, the value from