create kafka exporter addon

Author: spdeepak

addons/confluent-kafka/go.mod

@@ -0,0 +1,8 @@
+module github.com/spdeepak/flowtracker/confluent-kafka
+
+go 1.24
+
+require (
+	github.com/confluentinc/confluent-kafka-go v1.9.2
+	github.com/spdeepak/flowtracker v0.0.3
+)

addons/confluent-kafka/go.sum

@@ -0,0 +1,114 @@
+package confluent_kafka
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"github.com/spdeepak/flowtracker"
+)
+
+// Config holds the setup parameters.
+// You must provide EITHER Producer OR KafkaConfigMap.
+type Config struct {
+	// Topic is the destination Kafka topic name (Required).
+	Topic string
+
+	// Producer allows you to pass an existing Kafka Producer.
+	// If this is set, KafkaConfigMap is ignored.
+	Producer *kafka.Producer
+
+	// KafkaConfigMap allows you to configure a new Producer.
+	// Use this if you don't have an existing client.
+	// Example: &kafka.ConfigMap{"bootstrap.servers": "localhost:9092"}
+	KafkaConfigMap *kafka.ConfigMap
+}
+
+// KafkaExporter implements the flowtracker.Exporter interface.
+type KafkaExporter struct {
+	producer *kafka.Producer
+	topic    string
+	// isOwned tracks if this exporter created the producer (and thus should close it).
+	isOwned bool
+}
+
+// New creates a new KafkaExporter.
+func New(cfg Config) (*KafkaExporter, error) {
+	if cfg.Topic == "" {
+		return nil, fmt.Errorf("kafka exporter: topic is required")
+	}
+
+	var p *kafka.Producer
+	var err error
+	isOwned := false
+
+	if cfg.Producer != nil {
+		// Use the user-provided producer
+		p = cfg.Producer
+	} else if cfg.KafkaConfigMap != nil {
+		// Initialize a new producer
+		p, err = kafka.NewProducer(cfg.KafkaConfigMap)
+		if err != nil {
+			return nil, fmt.Errorf("kafka exporter: failed to create producer: %w", err)
+		}
+		isOwned = true
+
+		// Background goroutine to handle delivery reports.
+		// Essential for confluent-kafka-go to prevent local queue filling up.
+		go func() {
+			for e := range p.Events() {
+				switch ev := e.(type) {
+				case *kafka.Message:
+					if ev.TopicPartition.Error != nil {
+						log.Printf("FlowTracker Kafka Error: %v\n", ev.TopicPartition.Error)
+					}
+				}
+			}
+		}()
+	} else {
+		return nil, fmt.Errorf("kafka exporter: must provide either Producer or KafkaConfigMap")
+	}
+
+	return &KafkaExporter{
+		producer: p,
+		topic:    cfg.Topic,
+		isOwned:  isOwned,
+	}, nil
+}
+
+// Export sends the trace to Kafka.
+func (k *KafkaExporter) Export(tr *flowtracker.Trace) {
+	// Serialize Trace to JSON
+	payload, err := json.Marshal(tr)
+	if err != nil {
+		log.Printf("FlowTracker: failed to marshal trace: %v", err)
+		return
+	}
+
+	// Construct the Kafka Message
+	msg := &kafka.Message{
+		TopicPartition: kafka.TopicPartition{Topic: &k.topic, Partition: kafka.PartitionAny},
+		Value:          payload,
+		// We use the TraceID as the Key. This ensures that if you update this logic
+		// to stream updates, all spans for the same trace go to the same partition.
+		Key: []byte(tr.TraceID),
+	}
+
+	// Produce is asynchronous. We rely on the background event loop (started in New) to handle errors.
+	err = k.producer.Produce(msg, nil)
+	if err != nil {
+		log.Printf("FlowTracker: failed to produce message: %v", err)
+	}
+}
+
+// Close flushes the producer if we own it.
+// Note: The main flowtracker library doesn't call Close(), but you can call this
+// manually in your main.go shutdown hook.
+func (k *KafkaExporter) Close() {
+	if k.isOwned && k.producer != nil {
+		// Wait up to 5 seconds for outstanding messages to be delivered
+		k.producer.Flush(5000)
+		k.producer.Close()
+	}
+}

addons/confluent-kafka/kafka.go

@@ -0,0 +1,30 @@
+version: "3"
+services:
+  zookeeper:
+    image: "zookeeper:latest"
+    environment:
+      - ZOO_CFG_EXTRA=autopurge.snapRetainCount=3
+      - ZOO_CFG_EXTRA=autopurge.purgeInterval=24
+    ports:
+      - "2181:2181"
+    restart: unless-stopped
+
+  kafka:
+    image: wurstmeister/kafka
+    ports:
+      - "9092:9092"
+      - "19092:19092"
+    volumes:
+      - ./test.json:/test.json
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: localhost
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:19092
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_HOST://localhost:19092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_CREATE_TOPICS: "flowtracker.logs:1:1"
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+    depends_on:
+      - zookeeper
+    restart: unless-stopped

examples/confluent-kafka/docker-compose.yaml

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"log/slog"
+	"net/http"
+	"os"
+
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"github.com/spdeepak/flowtracker"
+	confluentkafka "github.com/spdeepak/flowtracker/confluent-kafka"
+	"github.com/spdeepak/flowtracker/examples"
+)
+
+func main() {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", examples.Handler)
+
+	config := confluentkafka.Config{
+		Topic: "flowtracker.logs",
+		KafkaConfigMap: &kafka.ConfigMap{
+			"bootstrap.servers": "localhost:9092",
+			"client.id":         "flowtracker-client",
+			"acks":              "all",
+		},
+	}
+	exporter, err := confluentkafka.New(config)
+	if err != nil {
+		slog.Error("Error during kafka exporter creation", slog.Any("error", err))
+		os.Exit(1)
+	}
+	mw := flowtracker.NewMiddleware(flowtracker.WithExporter(exporter))
+
+	http.ListenAndServe(":8080", mw(mux))
+}

examples/confluent-kafka/kafka.go

@@ -2,6 +2,14 @@ module github.com/spdeepak/flowtracker/examples
 
 go 1.24
 
-require github.com/spdeepak/flowtracker v0.0.3
+require (
+	github.com/spdeepak/flowtracker v0.0.3
+	github.com/spdeepak/flowtracker/confluent-kafka v0.0.1
+)
 
-replace github.com/spdeepak/flowtracker => ../
+require github.com/confluentinc/confluent-kafka-go v1.9.2 // indirect
+
+replace (
+	github.com/spdeepak/flowtracker => ../
+	github.com/spdeepak/flowtracker/confluent-kafka => ../addons/confluent-kafka/
+)