feat(opentelemetry source): support OTLP logs, metrics & traces

changelog.d/otlp_decoding.enhancement.md

@@ -0,0 +1,8 @@
+The `opentelemetry` source now supports a new decoding mode which can be enabled by setting `use_otlp_decoding` to `true`. In this mode,
+all events preserve the [OTLP](https://opentelemetry.io/docs/specs/otel/protocol/) format. These events can be forwarded directly to
+the `opentelemetry` sink without modifications.
+
+**Note:** The OTLP metric format and the Vector metric format differ, so the `opentelemetry` source emits OTLP formatted metrics as Vector log
+events. These events cannot be used with existing metrics transforms. However, they can be ingested by the OTEL collectors as metrics.
+
+authors: pront

lib/codecs/src/decoding/format/mod.rs

@@ -44,11 +44,18 @@ pub trait Deserializer: DynClone + Send + Sync {
     /// frame can potentially hold multiple events, e.g. when parsing a JSON
     /// array. However, we optimize the most common case of emitting one event
     /// by not requiring heap allocations for it.
+    ///
+    /// **Note**: The type of the produced events depends on the implementation.
     fn parse(
         &self,
         bytes: Bytes,
         log_namespace: LogNamespace,
     ) -> vector_common::Result<SmallVec<[Event; 1]>>;
+
+    /// Parses trace events from bytes.
+    fn parse_traces(&self, _bytes: Bytes) -> vector_common::Result<SmallVec<[Event; 1]>> {
+        unimplemented!()
+    }
 }
 
 dyn_clone::clone_trait_object!(Deserializer);

lib/codecs/src/decoding/format/protobuf.rs

@@ -6,17 +6,15 @@ use derivative::Derivative;
 use prost_reflect::{DynamicMessage, MessageDescriptor};
 use smallvec::{SmallVec, smallvec};
 use vector_config::configurable_component;
-use vector_core::event::LogEvent;
+use vector_core::event::{LogEvent, TraceEvent};
 use vector_core::{
     config::{DataType, LogNamespace, log_schema},
     event::Event,
     schema,
 };
-use vrl::protobuf::{
-    descriptor::get_message_descriptor,
-    parse::{Options, proto_to_value},
-};
-use vrl::value::Kind;
+use vrl::protobuf::descriptor::{get_message_descriptor, get_message_descriptor_from_bytes};
+use vrl::protobuf::parse::{Options, proto_to_value};
+use vrl::value::{Kind, Value};
 
 use super::Deserializer;
 
@@ -86,30 +84,56 @@ pub struct ProtobufDeserializerOptions {
 #[derive(Debug, Clone)]
 pub struct ProtobufDeserializer {
     message_descriptor: MessageDescriptor,
+    options: Options,
 }
 
 impl ProtobufDeserializer {
     /// Creates a new `ProtobufDeserializer`.
     pub fn new(message_descriptor: MessageDescriptor) -> Self {
-        Self { message_descriptor }
+        Self {
+            message_descriptor,
+            options: Default::default(),
+        }
+    }
+
+    /// Creates a new deserializer instance using the descriptor bytes directly.
+    pub fn new_from_bytes(
+        desc_bytes: &[u8],
+        message_type: &str,
+        options: Options,
+    ) -> vector_common::Result<Self> {
+        let message_descriptor = get_message_descriptor_from_bytes(desc_bytes, message_type)?;
+        Ok(Self {
+            message_descriptor,
+            options,
+        })
     }
 }
 
+fn extract_vrl_value(
+    bytes: Bytes,
+    message_descriptor: &MessageDescriptor,
+    options: &Options,
+) -> vector_common::Result<Value> {
+    let dynamic_message = DynamicMessage::decode(message_descriptor.clone(), bytes)
+        .map_err(|error| format!("Error parsing protobuf: {error:?}"))?;
+
+    Ok(proto_to_value(
+        &prost_reflect::Value::Message(dynamic_message),
+        None,
+        options,
+    )?)
+}
+
 impl Deserializer for ProtobufDeserializer {
     fn parse(
         &self,
         bytes: Bytes,
         log_namespace: LogNamespace,
     ) -> vector_common::Result<SmallVec<[Event; 1]>> {
-        let dynamic_message = DynamicMessage::decode(self.message_descriptor.clone(), bytes)
-            .map_err(|error| format!("Error parsing protobuf: {error:?}"))?;
-
-        let proto_vrl = proto_to_value(
-            &prost_reflect::Value::Message(dynamic_message),
-            None,
-            &Options::default(),
-        )?;
-        let mut event = Event::Log(LogEvent::from(proto_vrl));
+        let vrl_value = extract_vrl_value(bytes, &self.message_descriptor, &self.options)?;
+        let mut event = Event::Log(LogEvent::from(vrl_value));
+
         let event = match log_namespace {
             LogNamespace::Vector => event,
             LogNamespace::Legacy => {
@@ -126,6 +150,12 @@ impl Deserializer for ProtobufDeserializer {
 
         Ok(smallvec![event])
     }
+
+    fn parse_traces(&self, bytes: Bytes) -> vector_common::Result<SmallVec<[Event; 1]>> {
+        let vrl_value = extract_vrl_value(bytes, &self.message_descriptor, &self.options)?;
+        let trace_event = Event::Trace(TraceEvent::from(vrl_value));
+        Ok(smallvec![trace_event])
+    }
 }
 
 impl TryFrom<&ProtobufDeserializerConfig> for ProtobufDeserializer {

lib/opentelemetry-proto/build.rs

@@ -1,5 +1,7 @@
 use glob::glob;
-use std::{env, io::Result, path::PathBuf};
+use std::fs::{read_to_string, write};
+use std::path::Path;
+use std::{io::Result, path::PathBuf};
 
 fn main() -> Result<()> {
     let proto_root = PathBuf::from("src/proto/opentelemetry-proto");
@@ -10,12 +12,7 @@ fn main() -> Result<()> {
         .filter_map(|result| result.ok())
         .collect();
 
-    // Set up re-run triggers
-    for proto in &proto_paths {
-        println!("cargo:rerun-if-changed={}", proto.display());
-    }
-
-    let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());
+    let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap());
     let descriptor_path = out_dir.join("opentelemetry-proto.desc");
 
     tonic_build::configure()
@@ -24,5 +21,22 @@ fn main() -> Result<()> {
         .file_descriptor_set_path(&descriptor_path)
         .compile(&proto_paths, &[include_path])?;
 
+    write_static_descriptor_reference(&descriptor_path, &out_dir)?;
+
+    Ok(())
+}
+
+fn write_static_descriptor_reference(descriptor_path: &Path, out_dir: &Path) -> Result<()> {
+    let include_line = format!(
+        "pub static DESCRIPTOR_BYTES: &[u8] = include_bytes!(r\"{}\");\n",
+        descriptor_path.display()
+    );
+
+    let include_file = out_dir.join("opentelemetry-proto.rs");
+    let existing = read_to_string(&include_file).ok();
+    if existing.as_deref() != Some(&include_line) {
+        write(&include_file, include_line)?;
+    }
+
     Ok(())
 }

lib/opentelemetry-proto/src/proto.rs

@@ -51,3 +51,6 @@ pub mod resource {
         tonic::include_proto!("opentelemetry.proto.resource.v1");
     }
 }
+
+/// The raw descriptor bytes for all the above.
+include!(concat!(env!("OUT_DIR"), "/opentelemetry-proto.rs"));

lib/vector-core/src/event/trace.rs

@@ -118,6 +118,13 @@ impl TraceEvent {
     }
 }
 
+impl From<Value> for TraceEvent {
+    fn from(value: Value) -> Self {
+        let log_event = LogEvent::from(value);
+        Self(log_event)
+    }
+}
+
 impl From<LogEvent> for TraceEvent {
     fn from(log: LogEvent) -> Self {
         Self(log)

scripts/e2e/opentelemetry-logs/compose.yaml

@@ -2,7 +2,7 @@ name: opentelemetry-vector-e2e
 services:
   otel-collector-source:
     container_name: otel-collector-source
-    image: otel/opentelemetry-collector-contrib:${CONFIG_COLLECTOR_VERSION:-latest}
+    image: otel/opentelemetry-collector-contrib:${CONFIG_COLLECTOR_VERSION}
     init: true
     volumes:
       - type: bind
@@ -33,7 +33,7 @@ services:
 
   otel-collector-sink:
     container_name: otel-collector-sink
-    image: otel/opentelemetry-collector-contrib:${CONFIG_COLLECTOR_VERSION:-latest}
+    image: otel/opentelemetry-collector-contrib:${CONFIG_COLLECTOR_VERSION}
     init: true
     volumes:
       - type: bind
@@ -57,7 +57,7 @@ services:
     init: true
     volumes:
       - type: bind
-        source: ../../../tests/data/e2e/opentelemetry/logs/vector.yaml
+        source: ../../../tests/data/e2e/opentelemetry/logs/${CONFIG_VECTOR_CONFIG}
         target: /etc/vector/vector.yaml
         read_only: true
       - type: bind

scripts/e2e/opentelemetry-logs/test.yaml

@@ -1,5 +1,5 @@
 features:
-- e2e-tests-opentelemetry
+  - e2e-tests-opentelemetry
 
 test: "e2e"
 
@@ -14,6 +14,7 @@ runner:
 matrix:
   # Determines which `otel/opentelemetry-collector-contrib` version to use
   collector_version: [ 'latest' ]
+  vector_config: [ 'vector_default.yaml', 'vector_otlp.yaml' ]
 
 # Only trigger this integration test if relevant OTEL source/sink files change
 paths: