Add support for OTLP trace data in any source

quickwit-oss · Oct 23, 2023 · be1c795 · be1c795
1 parent 7414956
commit be1c795
Show file tree

Hide file tree

Showing 13 changed files with 615 additions and 249 deletions.
diff --git a/...ig/tutorials/otel-trace/index-config.yaml → ...g/tutorials/otel-traces/index-config.yaml b/...ig/tutorials/otel-trace/index-config.yaml → ...g/tutorials/otel-traces/index-config.yaml
diff --git a/config/tutorials/otel-traces/kafka-source.yaml b/config/tutorials/otel-traces/kafka-source.yaml
@@ -0,0 +1,8 @@
+version: 0.6
+source_id: kafka-source
+source_type: kafka
+input_format: otlp_trace_proto
+params:
+  topic: otlp_spans
+  client_params:
+    bootstrap.servers: localhost:9092
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -93,14 +93,15 @@ services:
     ports:
       - "${MAP_HOST_KAFKA:-127.0.0.1}:9092:9092"
       - "${MAP_HOST_KAFKA:-127.0.0.1}:9101:9101"
+      - "${MAP_HOST_KAFKA:-127.0.0.1}:29092:29092"
     profiles:
       - all
       - kafka
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT_HOST://localhost:9092,PLAINTEXT://kafka-broker:29092
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
@@ -167,8 +168,6 @@ services:
     image: jaegertracing/all-in-one:${JAEGER_VERSION:-1.48.0}
     container_name: jaeger
     ports:
-      - "${MAP_HOST_JAEGER:-127.0.0.1}:4317:4317"   # OTLP over gRPC
-      - "${MAP_HOST_JAEGER:-127.0.0.1}:4318:4318"   # OTLP over HTTP
       - "${MAP_HOST_JAEGER:-127.0.0.1}:16686:16686" # Frontend
     profiles:
       - jaeger
@@ -206,18 +205,19 @@ services:
       - "host.docker.internal:host-gateway"
 
   gcp-pubsub-emulator:
-    profiles:
-      - gcp-pubsub
-      - all
     # It is not an official docker image
     # if we prefer we can build a docker from the official docker image (gcloud cli)
     # and install the pubsub emulator https://cloud.google.com/pubsub/docs/emulator
     image: thekevjames/gcloud-pubsub-emulator:${GCLOUD_EMULATOR:-7555256f2c}
+    container_name: gcp-pubsub-emulator
     ports:
       - "${MAP_HOST_GCLOUD_EMULATOR:-127.0.0.1}:8681:8681"
     environment:
       # create a fake gcp project and a topic / subscription
       - PUBSUB_PROJECT1=quickwit-emulator,emulator_topic:emulator_subscription
+    profiles:
+      - all
+      - gcp-pubsub
 
 volumes:
   localstack_data:

diff --git a/monitoring/otel-collector-config.yaml b/monitoring/otel-collector-config.yaml
@@ -20,6 +20,10 @@ exporters:
     tls:
       insecure: true
 
+  kafka:
+    brokers:
+      - kafka-broker:29092
+
   otlp/qw:
     endpoint: host.docker.internal:7281
     tls:
@@ -36,7 +40,7 @@ service:
     traces:
       receivers: [jaeger, otlp]
       processors: [batch]
-      exporters: [jaeger, otlp/qw]
+      exporters: [jaeger, kafka, otlp/qw]
     # metrics:
     #   receivers: [otlp]
     #   processors: [batch]

diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/quickwit-config/src/source_config/mod.rs b/quickwit/quickwit-config/src/source_config/mod.rs
@@ -201,11 +201,14 @@ impl TestableForRegression for SourceConfig {
     }
 }
 
-#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize, utoipa::ToSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum SourceInputFormat {
     #[default]
     Json,
+    OtlpTraceJson,
+    #[serde(alias = "otlp_trace_proto")]
+    OtlpTraceProtobuf,
     #[serde(alias = "plain")]
     PlainText,
 }

diff --git a/quickwit/quickwit-config/src/source_config/serialize.rs b/quickwit/quickwit-config/src/source_config/serialize.rs
@@ -108,6 +108,12 @@ impl SourceConfigForSerialization {
         }
 
         if let Some(transform_config) = &self.transform {
+            if matches!(
+                self.input_format,
+                SourceInputFormat::OtlpTraceJson | SourceInputFormat::OtlpTraceProtobuf
+            ) {
+                bail!("VRL transforms are not supported for OTLP input formats");
+            }
             transform_config.validate_vrl_script()?;
         }
 

diff --git a/quickwit/quickwit-indexing/Cargo.toml b/quickwit/quickwit-indexing/Cargo.toml
@@ -61,6 +61,7 @@ quickwit-directories = { workspace = true }
 quickwit-doc-mapper = { workspace = true }
 quickwit-ingest = { workspace = true }
 quickwit-metastore = { workspace = true }
+quickwit-opentelemetry = { workspace = true }
 quickwit-proto = { workspace = true }
 quickwit-storage = { workspace = true }
 
@@ -87,9 +88,10 @@ bytes = { workspace = true }
 criterion = { workspace = true, features = ["async_tokio"] }
 mockall = { workspace = true }
 proptest = { workspace = true }
+prost = { workspace = true }
 rand = { workspace = true }
-tempfile = { workspace = true }
 reqwest = { workspace = true }
+tempfile = { workspace = true }
 
 quickwit-actors = { workspace = true, features = ["testsuite"] }
 quickwit-cluster = { workspace = true, features = ["testsuite"] }