feat(codecs): convert native Vector logs to OTLP in otlp serializer

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

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<Event> 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<Event> = 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(

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<PBValue> for Value {
     fn from(av: PBValue) -> Self {
@@ -24,6 +27,40 @@ impl From<PBValue> for Value {
     }
 }
 
+/// Inverse of `From<PBValue> 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<Value> 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<PBValue> for TagValue {
     fn from(pb: PBValue) -> Self {
         match pb {
@@ -80,6 +117,18 @@ pub fn kv_list_into_value(arr: Vec<KeyValue>) -> Value {
     )
 }
 
+/// Inverse of [`kv_list_into_value`].
+#[must_use]
+pub fn object_into_kv_list(object: ObjectMap) -> Vec<KeyValue> {
+    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() {