chore(loki): Add integration test for loki.source.kafka

integration-tests/docker/docker-compose.yaml

@@ -38,20 +38,23 @@ services:
     container_name: kafka
     ports:
       - "9092:9092"
+      - "29092:29092"
       - "9093:9093"
     environment:
       KAFKA_NODE_ID: 1
       KAFKA_PROCESS_ROLES: broker,controller
-      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093
+      KAFKA_LISTENERS: PLAINTEXT://:9092,PLAINTEXT_HOST://:29092,CONTROLLER://:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092,CONTROLLER://kafka:9093
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 10
       KAFKA_NUM_PARTITIONS: 3
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
     healthcheck:
       test:
         [

integration-tests/docker/tests/loki-kafka/config.alloy

@@ -0,0 +1,40 @@
+loki.relabel "kafka" {
+	forward_to = []
+
+	rule {
+		source_labels = ["__meta_kafka_topic"]
+		target_label  = "topic"
+	}
+}
+
+loki.source.kafka "kafka" {
+	brokers       = ["kafka:9092"]
+	topics        = ["test1", "test2", "test3"]
+	forward_to    = [loki.process.test.receiver]
+	relabel_rules = loki.relabel.kafka.rules
+}
+
+loki.process "test" {
+	forward_to = [loki.relabel.test.receiver]
+
+	stage.static_labels {
+		values = {
+			test_name = "lokikafka",
+		}
+	}
+}
+
+loki.relabel "test" {
+	forward_to = [loki.write.test.receiver]
+
+	rule {
+		source_labels = ["topic"]
+		target_label  = "service_name"
+	}
+}
+
+loki.write "test" {
+	endpoint {
+		url = "http://loki:3100/loki/api/v1/push"
+	}
+}

integration-tests/docker/tests/loki-kafka/loki_kafka_test.go

@@ -0,0 +1,106 @@
+//go:build alloyintegrationtests
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/IBM/sarama"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/integration-tests/docker/common"
+	"github.com/grafana/dskit/backoff"
+)
+
+var topics = []string{"test1", "test2", "test3"}
+
+const (
+	messagesPerTopic = 50
+	broker           = "localhost:29092"
+)
+
+func TestLokiKafka(t *testing.T) {
+	producer, err := newProducer()
+	require.NoError(t, err)
+	defer func() { require.NoError(t, producer.Close()) }()
+
+	var (
+		wg   sync.WaitGroup
+		errs = make([]error, len(topics))
+	)
+
+	for i, t := range topics {
+		wg.Go(func() { errs[i] = sendMessages(producer, t, messagesPerTopic) })
+	}
+
+	wg.Wait()
+	for _, err := range errs {
+		require.NoError(t, err)
+	}
+
+	var (
+		total      = messagesPerTopic * len(topics)
+		assertions = make([]common.ExpectedLogResult, 0, len(topics))
+	)
+
+	for _, t := range topics {
+		assertions = append(assertions, common.ExpectedLogResult{
+			Labels: map[string]string{
+				"service_name": t,
+			},
+			EntryCount: messagesPerTopic,
+		})
+	}
+
+	common.AssertLogsPresent(
+		t,
+		total,
+		assertions...,
+	)
+}
+
+func newProducer() (sarama.SyncProducer, error) {
+	config := sarama.NewConfig()
+	config.Producer.Return.Successes = true
+
+	backoff := backoff.New(context.Background(), backoff.Config{
+		MinBackoff: 50 * time.Millisecond,
+		MaxBackoff: 5 * time.Second,
+		MaxRetries: 10,
+	})
+
+	var (
+		lastErr  error
+		producer sarama.SyncProducer
+	)
+
+	for backoff.Ongoing() {
+		producer, lastErr = sarama.NewSyncProducer([]string{broker}, config)
+		if lastErr == nil {
+			return producer, nil
+		}
+
+		backoff.Wait()
+	}
+
+	return nil, fmt.Errorf("failed to connect to kafka: %w", lastErr)
+}
+
+func sendMessages(producer sarama.SyncProducer, topic string, n int) error {
+	for i := range n {
+		_, _, err := producer.SendMessage(&sarama.ProducerMessage{
+			Topic: topic,
+			Value: sarama.StringEncoder("log line " + strconv.Itoa(i)),
+		})
+		if err != nil {
+			return fmt.Errorf("failed to send message to topic %q: %w", topic, err)
+		}
+	}
+
+	return nil
+}