test(opentelemetry sink): add gRPC integration and e2e tests

Author: thomasqueirozb

Cargo.toml

@@ -1030,7 +1030,7 @@ mongodb_metrics-integration-tests = ["sources-mongodb_metrics"]
 mqtt-integration-tests = ["sinks-mqtt", "sources-mqtt"]
 nats-integration-tests = ["sinks-nats", "sources-nats"]
 nginx-integration-tests = ["sources-nginx_metrics"]
-opentelemetry-integration-tests = ["sources-opentelemetry", "dep:prost"]
+opentelemetry-integration-tests = ["sources-opentelemetry", "sinks-opentelemetry", "dep:prost"]
 postgresql_metrics-integration-tests = ["sources-postgresql_metrics"]
 postgres_sink-integration-tests = ["sinks-postgres"]
 prometheus-integration-tests = ["sinks-prometheus", "sources-prometheus", "sinks-influxdb"]

src/sinks/opentelemetry/integration_tests.rs

@@ -0,0 +1,76 @@
+use futures::stream;
+use prost::Message as _;
+use vector_lib::{
+    codecs::decoding::format::{Deserializer as _, OtlpDeserializer},
+    config::LogNamespace,
+    opentelemetry::proto::{
+        collector::logs::v1::ExportLogsServiceRequest,
+        logs::v1::{LogRecord, ResourceLogs, ScopeLogs},
+        resource::v1::Resource,
+    },
+};
+
+use super::grpc::GrpcSinkConfig;
+use crate::{
+    config::SinkContext,
+    test_util::{
+        components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance},
+        wait_for_tcp,
+    },
+};
+
+fn sink_grpc_address() -> String {
+    std::env::var("OTEL_GRPC_SINK_ADDRESS")
+        .unwrap_or_else(|_| "opentelemetry-collector:4317".to_owned())
+}
+
+fn otlp_log_event() -> vector_lib::event::Event {
+    let req = ExportLogsServiceRequest {
+        resource_logs: vec![ResourceLogs {
+            resource: Some(Resource {
+                attributes: vec![],
+                dropped_attributes_count: 0,
+            }),
+            scope_logs: vec![ScopeLogs {
+                scope: None,
+                log_records: vec![LogRecord {
+                    severity_text: "INFO".to_string(),
+                    body: Some(vector_lib::opentelemetry::proto::common::v1::AnyValue {
+                        value: Some(
+                            vector_lib::opentelemetry::proto::common::v1::any_value::Value::StringValue(
+                                "integration test log message".to_string(),
+                            ),
+                        ),
+                    }),
+                    ..Default::default()
+                }],
+                schema_url: String::new(),
+            }],
+            schema_url: String::new(),
+        }],
+    };
+
+    let bytes = bytes::Bytes::from(req.encode_to_vec());
+    let mut events = OtlpDeserializer::default()
+        .parse(bytes, LogNamespace::Legacy)
+        .expect("failed to deserialize OTLP log event");
+    events.remove(0)
+}
+
+#[tokio::test]
+async fn delivers_logs_via_grpc() {
+    let address = sink_grpc_address();
+    wait_for_tcp(&address).await;
+
+    let config: GrpcSinkConfig = toml::from_str(&format!(
+        r#"
+            endpoint = "http://{address}"
+        "#
+    ))
+    .unwrap();
+
+    let (sink, _healthcheck) = config.build(SinkContext::default()).await.unwrap();
+
+    let events = vec![otlp_log_event()];
+    run_and_assert_sink_compliance(sink, stream::iter(events), &HTTP_SINK_TAGS).await;
+}

src/sinks/opentelemetry/mod.rs

@@ -1,5 +1,8 @@
 mod grpc;
 
+#[cfg(all(test, feature = "opentelemetry-integration-tests"))]
+mod integration_tests;
+
 use indoc::indoc;
 use vector_config::component::GenerateConfig;
 use vector_lib::{

tests/e2e/opentelemetry-logs/config/compose.yaml

@@ -60,6 +60,7 @@ services:
         target: /output
     ports:
       - "${OTEL_COLLECTOR_SINK_HTTP_PORT:-5318}:5318"
+      - "${OTEL_COLLECTOR_SINK_GRPC_PORT:-14317}:4317"
 
   vector:
     container_name: vector-otel-logs-e2e

tests/e2e/opentelemetry-logs/config/test.yaml

@@ -11,11 +11,12 @@ runner:
     OTEL_COLLECTOR_SOURCE_GRPC_PORT: '4317'
     OTEL_COLLECTOR_SOURCE_HTTP_PORT: '4318'
     OTEL_COLLECTOR_SINK_HTTP_PORT: '5318'
+    OTEL_COLLECTOR_SINK_GRPC_PORT: '14317'
 
 matrix:
   # Determines which `otel/opentelemetry-collector-contrib` version to use
   collector_version: [ 'latest' ]
-  vector_config: [ 'vector_default.yaml', 'vector_otlp.yaml' ]
+  vector_config: [ 'vector_default.yaml', 'vector_otlp.yaml', 'vector_grpc.yaml' ]
 
 # Only trigger this integration test if relevant OTEL source/sink files change
 paths:

tests/e2e/opentelemetry-logs/data/collector-sink.yaml

@@ -3,6 +3,8 @@ receivers:
     protocols:
       http:
         endpoint: "0.0.0.0:5318"
+      grpc:
+        endpoint: "0.0.0.0:4317"
 
 processors:
   batch: { }

tests/e2e/opentelemetry-logs/data/vector_grpc.yaml

@@ -0,0 +1,40 @@
+sources:
+  source0:
+    type: opentelemetry
+    grpc:
+      address: 0.0.0.0:4317
+    http:
+      address: 0.0.0.0:4318
+      keepalive:
+        max_connection_age_jitter_factor: 0.1
+        max_connection_age_secs: 300
+    use_otlp_decoding: true
+
+  internal_metrics:
+    type: internal_metrics
+    scrape_interval_secs: 60
+
+sinks:
+  otel_sink:
+    inputs:
+      - source0.logs
+    type: opentelemetry
+    protocol:
+      type: grpc
+      endpoint: http://otel-collector-sink:4317
+
+  otel_file_sink:
+    type: file
+    path: "/output/opentelemetry-logs/vector-file-sink.log"
+    inputs:
+      - source0.logs
+    encoding:
+      codec: json
+
+  metrics_file_sink:
+    type: file
+    path: "/output/opentelemetry-logs/vector-internal-metrics-sink.log"
+    inputs:
+      - internal_metrics
+    encoding:
+      codec: json

tests/integration/opentelemetry/config/test.yaml

@@ -7,6 +7,7 @@ runner:
   env:
     OTEL_HEALTH_URL: http://opentelemetry-collector:13133
     OTEL_OTLPHTTP_URL: http://opentelemetry-collector:9876
+    OTEL_GRPC_SINK_ADDRESS: opentelemetry-collector:4317
 
 matrix:
   version: [0.56.0]
@@ -15,5 +16,6 @@ matrix:
 # expressions are evaluated using https://github.com/micromatch/picomatch
 paths:
 - "src/sources/opentelemetry/**"
+- "src/sinks/opentelemetry/**"
 - "src/sources/util/**"
 - "tests/integration/opentelemetry/**"

tests/integration/opentelemetry/data/config.yaml

@@ -3,6 +3,8 @@ receivers:
     protocols:
       http:
         endpoint: 0.0.0.0:9876
+      grpc:
+        endpoint: 0.0.0.0:4317
 
 exporters:
   otlp: