Add support for OTLP trace data in any source (#3988)

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

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:

docs/reference/metrics.md

@@ -33,8 +33,8 @@ Currently Quickwit exposes metrics for three caches: `fastfields`, `shortlived`,
 
 | Namespace | Metric Name | Description | Labels | Type |
 | --------- | ----------- | ----------- | ------ | ---- |
-| `quickwit_indexing` | `processed_docs_total`| Number of processed docs by index, source and processed status in [`valid`, `missing_field`, `parsing_error`, `transform_error`] | [`index`, `source`, `docs_processed_status`] | `counter` |
-| `quickwit_indexing` | `processed_docs_total`| Number of processed bytes by index, source and processed status in [`valid`, `missing_field`, `parsing_error`, `transform_error`] | [`index`, `source`, `docs_processed_status`] | `counter` |
+| `quickwit_indexing` | `processed_docs_total`| Number of processed docs by index, source and processed status in [`valid`, `schema_error`, `parse_error`, `transform_error`] | [`index`, `source`, `docs_processed_status`] | `counter` |
+| `quickwit_indexing` | `processed_docs_total`| Number of processed bytes by index, source and processed status in [`valid`, `schema_error`, `parse_error`, `transform_error`] | [`index`, `source`, `docs_processed_status`] | `counter` |
 | `quickwit_indexing` | `available_concurrent_upload_permits`| Number of available concurrent upload permits by component in [`merger`, `indexer`] | [`component`] | `gauge` |
 | `quickwit_indexing` | `ongoing_merge_operations`| Number of available concurrent upload permits by component in [`merger`, `indexer`]. | [`index`, `source`] | `gauge` |
 

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]

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,
 }

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()?;
         }
 

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"] }