Add Kafka exporter and receiver configuration

cmd/jaeger/collector-with-kafka.yaml

@@ -0,0 +1,28 @@
+service:
+  pipelines:
+    traces:
+      receivers: [otlp, jaeger]
+      processors: [batch]
+      exporters: [kafka]
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_binary:
+      thrift_compact:
+      thrift_http:
+
+processors:
+  batch:
+
+exporters:
+  kafka:
+    brokers:
+      - localhost:9092
+    topic: "jaeger-spans"
+    encoding: otlp_proto

cmd/jaeger/ingester-remote-storage.yaml

@@ -0,0 +1,37 @@
+service:
+  extensions: [jaeger_storage, jaeger_query]
+  pipelines:
+    traces:
+      receivers: [kafka]
+      processors: [batch]
+      exporters: [jaeger_storage_exporter]
+  telemetry:
+    metrics:
+      address: 0.0.0.0:8889
+    logs:
+      level: debug
+
+extensions:
+  jaeger_query:
+    trace_storage: some_storage
+
+  jaeger_storage:
+    backends:
+      some_storage:
+        memory:
+          max_traces: 100000
+
+receivers:  
+  kafka:
+    brokers:
+      - localhost:9092
+    topic: "jaeger-spans"
+    encoding: otlp_proto
+    initial_offset: earliest
+
+processors:
+  batch:
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

cmd/jaeger/internal/integration/README.md

@@ -63,6 +63,56 @@ flowchart LR
     end
 ```
 
+## Kafka Integration
+
+The primary difference between the Kafka integration tests and other integration tests lies in the flow of data. In the standard tests, spans are written by the SpanWriter, sent through an RPC_client directly to a receiver, then to an exporter, and written to a storage backend. Spans are read by the SpanReader, which queries the jaeger_query process accessing the storage backend. In contrast, the Kafka tests introduce Kafka as an intermediary. Spans go from the SpanWriter through an RPC_client to an OTLP receiver in the Jaeger Collector, exported to Kafka, received by the Jaeger Ingester, and then stored. For details, see the [Architecture](#KafkaArchitecture) section below.
+
+
+## Kafka Architecture
+
+``` mermaid
+flowchart LR
+        Test -->|writeSpan| SpanWriter
+        SpanWriter --> RPCW[RPC_client]
+        RPCW --> OTLP_Receiver[Receiver]
+        OTLP_Receiver --> CollectorExporter[Kafka Exporter]
+        CollectorExporter --> Kafka[Kafka]
+        Kafka --> IngesterReceiver[Kafka Receiver]
+        IngesterReceiver --> IngesterExporter[Exporter]
+        IngesterExporter --> StorageBackend[(In-Memory Store)]
+        Test -->|readSpan| SpanReader
+        SpanReader --> RPCR[RPC_client]
+        RPCR --> QueryProcess[jaeger_query]
+        StorageCleaner -->|purge| StorageBackend
+        QueryProcess --> StorageBackend
+
+
+        subgraph Integration_Test_Executable
+            Test
+            SpanWriter
+            SpanReader
+            RPCW
+            RPCR
+        end
+
+        subgraph Jaeger Collector
+            OTLP_Receiver
+            CollectorExporter
+        end
+
+        subgraph Jaeger Ingester
+            IngesterReceiver
+            IngesterExporter
+            QueryProcess
+            StorageBackend
+            StorageCleaner[Storage Cleaner Extension]
+        end
+
+        subgraph Kafka
+            Topic
+        end
+```
+
 ## Running tests locally
 
 All integration tests can be run locally.

cmd/jaeger/internal/integration/badger_test.go

@@ -23,7 +23,7 @@ func TestBadgerStorage(t *testing.T) {
 			GetOperationsMissingSpanKind: true,
 		},
 	}
-	s.e2eInitialize(t, "badger")
+	s.e2eInitialize(t, "badger", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 	})

cmd/jaeger/internal/integration/cassandra_test.go

@@ -21,7 +21,7 @@ func TestCassandraStorage(t *testing.T) {
 			SkipList: integration.CassandraSkippedTests,
 		},
 	}
-	s.e2eInitialize(t, "cassandra")
+	s.e2eInitialize(t, "cassandra", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 	})

cmd/jaeger/internal/integration/e2e_integration.go

@@ -44,9 +44,9 @@ type E2EStorageIntegration struct {
 // e2eInitialize starts the Jaeger-v2 collector with the provided config file,
 // it also initialize the SpanWriter and SpanReader below.
 // This function should be called before any of the tests start.
-func (s *E2EStorageIntegration) e2eInitialize(t *testing.T, storage string) {
+func (s *E2EStorageIntegration) e2eInitialize(t *testing.T, storage string, injectStorageCleaner bool) {
 	logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
-	configFile := createStorageCleanerConfig(t, s.ConfigFile, storage)
+	configFile := createStorageCleanerConfig(t, s.ConfigFile, storage, injectStorageCleaner)
 	t.Logf("Starting Jaeger-v2 in the background with config file %s", configFile)
 
 	outFile, err := os.OpenFile(
@@ -124,28 +124,44 @@ func (s *E2EStorageIntegration) e2eCleanUp(t *testing.T) {
 	require.NoError(t, s.SpanWriter.(io.Closer).Close())
 }
 
-func createStorageCleanerConfig(t *testing.T, configFile string, storage string) string {
+func createStorageCleanerConfig(t *testing.T, configFile string, storage string, injectStorageCleaner bool) string {
 	data, err := os.ReadFile(configFile)
 	require.NoError(t, err)
+
 	var config map[string]any
 	err = yaml.Unmarshal(data, &config)
 	require.NoError(t, err)
 
-	serviceAny, ok := config["service"]
-	require.True(t, ok)
-	service := serviceAny.(map[string]any)
-	service["extensions"] = append(service["extensions"].([]any), "storage_cleaner")
-
+	// Ensure extensions are correctly parsed before any modifications
 	extensionsAny, ok := config["extensions"]
 	require.True(t, ok)
 	extensions := extensionsAny.(map[string]any)
-	queryAny, ok := extensions["jaeger_query"]
-	require.True(t, ok)
-	traceStorageAny, ok := queryAny.(map[string]any)["trace_storage"]
-	require.True(t, ok)
-	traceStorage := traceStorageAny.(string)
-	extensions["storage_cleaner"] = map[string]string{"trace_storage": traceStorage}
 
+
+	if injectStorageCleaner{	
+		// Add storage_cleaner to the service extensions
+		serviceAny, ok := config["service"]
+		require.True(t, ok)
+		service := serviceAny.(map[string]any)
+
+		// Ensure the extensions key is present and is a slice
+		if extList, ok := service["extensions"].([]any); ok {
+			service["extensions"] = append(extList, "storage_cleaner")
+		} else {
+			service["extensions"] = []any{"storage_cleaner"}
+		}
+
+		// Ensure jaeger_query is correctly parsed
+		queryAny, ok := extensions["jaeger_query"]
+		require.True(t, ok)
+		query := queryAny.(map[string]any)
+		trace_storage, ok := query["trace_storage"].(string)
+		require.True(t, ok)
+
+		extensions["storage_cleaner"] = map[string]string{"trace_storage": trace_storage}
+	}
+
+    // Ensure jaeger_storage is correctly parsed
 	jaegerStorageAny, ok := extensions["jaeger_storage"]
 	require.True(t, ok)
 	jaegerStorage := jaegerStorageAny.(map[string]any)

cmd/jaeger/internal/integration/e2e_integration_test.go

@@ -8,6 +8,6 @@ import "testing"
 func TestCreateStorageCleanerConfig(t *testing.T) {
 	// Ensure that we can parse the existing configs correctly.
 	// This is faster to run than the full integration test.
-	createStorageCleanerConfig(t, "../../config-elasticsearch.yaml", "elasticsearch")
-	createStorageCleanerConfig(t, "../../config-opensearch.yaml", "opensearch")
+	createStorageCleanerConfig(t, "../../config-elasticsearch.yaml", "elasticsearch", true)
+	createStorageCleanerConfig(t, "../../config-opensearch.yaml", "opensearch", true)
 }

cmd/jaeger/internal/integration/elasticsearch_test.go

@@ -20,7 +20,7 @@ func TestElasticsearchStorage(t *testing.T) {
 			GetOperationsMissingSpanKind: true,
 		},
 	}
-	s.e2eInitialize(t, "elasticsearch")
+	s.e2eInitialize(t, "elasticsearch", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 	})

cmd/jaeger/internal/integration/grpc_test.go

@@ -34,7 +34,7 @@ func TestGRPCStorage(t *testing.T) {
 	}
 	s.CleanUp = s.cleanUp
 	s.initialize(t)
-	s.e2eInitialize(t, "grpc")
+	s.e2eInitialize(t, "grpc", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 		s.remoteStorage.Close(t)

cmd/jaeger/internal/integration/kafka_test.go

@@ -0,0 +1,50 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+)
+
+func TestKafkaStorage(t *testing.T) {
+    integration.SkipUnlessEnv(t, "kafka")
+
+    collectorConfig := "../../collector-with-kafka.yaml"
+    ingesterConfig := "../../ingester-remote-storage.yaml"
+
+    collector := &E2EStorageIntegration{
+        ConfigFile: collectorConfig,
+        StorageIntegration: integration.StorageIntegration{
+            GetDependenciesReturnsSource: true,
+            SkipArchiveTest: true,
+        },
+    }
+
+    // Initialize and start the collector
+    collector.e2eInitialize(t, "kafka-collector", false)
+
+
+    ingester := &E2EStorageIntegration{
+        ConfigFile: ingesterConfig,
+        StorageIntegration: integration.StorageIntegration{
+            CleanUp: purge,
+            GetDependenciesReturnsSource: true,
+            SkipArchiveTest: true,
+        },
+    }
+
+	// Initialize and start the ingester
+    ingester.e2eInitialize(t, "kafka-ingester", true)
+
+    // Set up cleanup for both collector and ingester
+    t.Cleanup(func() {
+        collector.e2eCleanUp(t)
+        ingester.e2eCleanUp(t)
+    })
+
+    // Run the span store tests
+    ingester.RunSpanStoreTests(t)
+}

cmd/jaeger/internal/integration/memory_test.go

@@ -19,7 +19,7 @@ func TestMemoryStorage(t *testing.T) {
 			CleanUp:         purge,
 		},
 	}
-	s.e2eInitialize(t, "memory")
+	s.e2eInitialize(t, "memory", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 	})

cmd/jaeger/internal/integration/opensearch_test.go

@@ -19,7 +19,7 @@ func TestOpenSearchStorage(t *testing.T) {
 			GetOperationsMissingSpanKind: true,
 		},
 	}
-	s.e2eInitialize(t, "opensearch")
+	s.e2eInitialize(t, "opensearch", true)
 	t.Cleanup(func() {
 		s.e2eCleanUp(t)
 	})

plugin/storage/integration/integration.go

@@ -49,7 +49,7 @@ var fixtures embed.FS
 // - in those functions it instantiates and populates this struct
 // - it then calls RunAll.
 //
-// Some implementations may declate multuple tests, with different settings,
+// Some implementations may declare multiple tests, with different settings,
 // and RunAll() under different conditions.
 type StorageIntegration struct {
 	SpanWriter        spanstore.Writer
@@ -541,4 +541,4 @@ func (s *StorageIntegration) RunSpanStoreTests(t *testing.T) {
 	t.Run("GetTrace", s.testGetTrace)
 	t.Run("GetLargeSpans", s.testGetLargeSpan)
 	t.Run("FindTraces", s.testFindTraces)
-}
+}