feat: add mqtt support on raccoon (#24)

Author: siddhanta-rath

.env.sample

@@ -31,3 +31,17 @@ METRIC_STATSD_ADDRESS=":8125"
 METRIC_STATSD_FLUSH_PERIOD_MS=100
 
 LOG_LEVEL="info"
+
+SERVER_MQTT_CONSUL_ADDRESS="consul:8081"
+SERVER_MQTT_CONSUL_KV_KEY="kv/path"
+SERVER_MQTT_CONSUL_HEALTH_ONLY="true"
+SERVER_MQTT_CONSUL_WAIT_TIME="300"
+SERVER_MQTT_AUTH_USERNAME="test"
+SERVER_MQTT_AUTH_PASSWORD="pass"
+SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC=1
+SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC=1
+SERVER_MQTT_CONSUMER_LOG_LEVEL="warn"
+SERVER_MQTT_CONSUMER_POOL_SIZE=1
+SERVER_MQTT_CONSUMER_TOPIC_FORMAT="default-topic"
+SERVER_MQTT_CONNECTION_GROUP="default"
+SERVER_MQTT_CONSUMER_KEEP_ALIVE_IN_SEC=45

.env.test

@@ -32,3 +32,17 @@ METRIC_STATSD_ADDRESS=":8125"
 METRIC_STATSD_FLUSH_PERIOD_MS=100
 
 LOG_LEVEL="info"
+
+SERVER_MQTT_CONSUL_ADDRESS="consul:8081"
+SERVER_MQTT_CONSUL_KV_KEY="kv/path"
+SERVER_MQTT_CONSUL_HEALTH_ONLY="true"
+SERVER_MQTT_CONSUL_WAIT_TIME="300"
+SERVER_MQTT_AUTH_USERNAME="test"
+SERVER_MQTT_AUTH_PASSWORD="pass"
+SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC=1
+SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC=1
+SERVER_MQTT_CONSUMER_LOG_LEVEL="warn"
+SERVER_MQTT_CONSUMER_POOL_SIZE=1
+SERVER_MQTT_CONSUMER_TOPIC_FORMAT="default-topic"
+SERVER_MQTT_CONNECTION_GROUP="default"
+SERVER_MQTT_CONSUMER_KEEP_ALIVE_IN_SEC=45

.github/workflows/build.yaml

@@ -12,7 +12,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v2.1.3
         with:
-          go-version: '1.17'
+          go-version: '1.24'
       - name: Checkout repo
         uses: actions/checkout@v2
       - name: Setup Project
@@ -28,7 +28,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v2.1.3
         with:
-          go-version: '1.17'
+          go-version: '1.24'
       - uses: actions/checkout@v2
       - name: Build
         run: make all

Dockerfile

@@ -1,11 +1,12 @@
-FROM golang:1.17
+FROM golang:1.24
 
 WORKDIR /app
 RUN apt-get update && apt-get install unzip  --no-install-recommends --assume-yes
 COPY . .
 RUN make update-deps && make compile
 
-FROM debian:bullseye
+#bookworm-slim comparible with golang:1.24 and has glibc 2.34
+FROM debian:bookworm-slim
 WORKDIR /app
 COPY --from=0 /app/raccoon ./raccoon
 COPY . .

app/server.go

@@ -3,6 +3,7 @@ package app
 import (
 	"context"
 	"fmt"
+	"github.com/goto/raccoon/health"
 	"os"
 	"os/signal"
 	"runtime"
@@ -31,7 +32,7 @@ func StartServer(ctx context.Context, cancel context.CancelFunc, shutdown chan b
 		logger.Info("Exiting server")
 		os.Exit(0)
 	}
-
+	registerHealthCheck(httpServices, kPublisher)
 	logger.Info("Start worker -->")
 	workerPool := worker.CreateWorkerPool(config.Worker.WorkersPoolSize, bufferChannel, config.Worker.DeliveryChannelSize, kPublisher)
 	workerPool.StartWorkers()
@@ -99,3 +100,12 @@ func reportProcMetrics() {
 		metrics.Gauge("server_mem_gc_pauseTotalNs_current", m.PauseTotalNs, "")
 	}
 }
+
+func registerHealthCheck(svcs services.Services, kafka *publisher.Kafka) {
+	health.Register("kafka-broker", kafka.HealthCheck)
+	for _, svc := range svcs.B {
+		if svc.Name() == "MQTT" {
+			health.Register("mqtt-broker", svc.HealthCheck)
+		}
+	}
+}

app/server_test.go

@@ -69,10 +69,11 @@ func isClosed(ch <-chan collection.CollectRequest) bool {
 // mockKafkaClient is a mock for the Client interface
 type mockKafkaClient struct {
 	// Tracking flags
-	ProduceCalled bool
-	CloseCalled   bool
-	FlushCalled   bool
-	EventsCalled  bool
+	ProduceCalled     bool
+	CloseCalled       bool
+	FlushCalled       bool
+	EventsCalled      bool
+	GetMetadataCalled bool
 
 	ReturnFlushLeft int
 	EventChan       chan kafka.Event
@@ -99,3 +100,8 @@ func (m *mockKafkaClient) Events() chan kafka.Event {
 	}
 	return m.EventChan
 }
+
+func (m *mockKafkaClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error) {
+	m.GetMetadataCalled = true
+	return &kafka.Metadata{}, nil
+}

config/load.go

@@ -29,4 +29,5 @@ func Load() {
 	metricStatsdConfigLoader()
 	eventDistributionConfigLoader()
 	eventConfigLoader()
+	serverMQTTConfigLoader()
 }

config/load_test.go

@@ -52,6 +52,35 @@ func TestGRPCServerConfig(t *testing.T) {
 	assert.Equal(t, "8081", ServerGRPC.Port)
 }
 
+func TestServerMQTTConfig(t *testing.T) {
+	os.Setenv("SERVER_MQTT_CONSUL_ADDRESS", "consul:8081")
+	os.Setenv("SERVER_MQTT_CONSUL_KV_KEY", "kv/path")
+	os.Setenv("SERVER_MQTT_CONSUL_HEALTH_ONLY", "true")
+	os.Setenv("SERVER_MQTT_CONSUL_WAIT_TIME", "300")
+	os.Setenv("SERVER_MQTT_AUTH_USERNAME", "test")
+	os.Setenv("SERVER_MQTT_AUTH_PASSWORD", "pass")
+	os.Setenv("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_LOG_LEVEL", "warn")
+	os.Setenv("SERVER_MQTT_CONSUMER_POOL_SIZE", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_TOPIC_FORMAT", "default-topic")
+	os.Setenv("SERVER_MQTT_CONNECTION_GROUP", "consumer")
+	serverMQTTConfigLoader()
+	assert.Equal(t, "consul:8081", ServerMQTT.ConsulConfig.Address)
+	assert.Equal(t, "kv/path", ServerMQTT.ConsulConfig.KVKey)
+	assert.Equal(t, true, ServerMQTT.ConsulConfig.HealthOnly)
+	assert.Equal(t, 300*time.Second, ServerMQTT.ConsulConfig.WaitTime)
+	assert.Equal(t, "test", ServerMQTT.AuthConfig.Username)
+	assert.Equal(t, "pass", ServerMQTT.AuthConfig.Password)
+	assert.Equal(t, 1*time.Second, ServerMQTT.ConsumerConfig.RetryIntervalInSec)
+	assert.Equal(t, 1*time.Second, ServerMQTT.ConsumerConfig.WriteTimeoutInSec)
+	assert.Equal(t, "warn", ServerMQTT.ConsumerConfig.LogLevel)
+	assert.Equal(t, 1, ServerMQTT.ConsumerConfig.PoolSize)
+	assert.Equal(t, "default-topic", ServerMQTT.ConsumerConfig.TopicFormat)
+	assert.Equal(t, "consumer", ServerMQTT.ConnGroup)
+
+}
+
 func TestDynamicConfigLoad(t *testing.T) {
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_RANDOM", "anything")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_BOOTSTRAP_SERVERS", "localhost:9092")
@@ -67,6 +96,8 @@ func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_ACKS", "1")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "10000")
 	os.Setenv("SOMETHING_PUBLISHER_KAFKA_CLIENT_SOMETHING", "anything")
+	os.Setenv("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME", "test-log")
+	os.Setenv("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS", "5000")
 	publisherKafkaConfigLoader()
 	kafkaConfig := PublisherKafka.ToKafkaConfigMap()
 	bootstrapServer, _ := kafkaConfig.Get("bootstrap.servers", "")
@@ -76,6 +107,8 @@ func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	assert.Equal(t, "", topic)
 	assert.NotEqual(t, something, "anything")
 	assert.Equal(t, 4, len(*kafkaConfig))
+	assert.Equal(t, "test-log", PublisherKafka.HealthCheckConfig.TopicName)
+	assert.Equal(t, 5000, PublisherKafka.HealthCheckConfig.TimeOut)
 }
 
 func TestWorkerConfig(t *testing.T) {

config/publisher.go

@@ -13,8 +13,17 @@ import (
 var PublisherKafka publisherKafka
 var dynamicKafkaClientConfigPrefix = "PUBLISHER_KAFKA_CLIENT_"
 
+// publisherKafka defines configuration parameters for a Kafka-based publisher.
+// It includes flushing behavior and health check settings.
 type publisherKafka struct {
-	FlushInterval int
+	FlushInterval     int         // Interval (in seconds) to flush the events during shutdown
+	HealthCheckConfig healthcheck // Configuration for Kafka broker health check
+}
+
+// healthcheck holds settings used to monitor the health of Kafka broker
+type healthcheck struct {
+	TopicName string // Kafka topic name used for health check
+	TimeOut   int    // Timeout duration (in seconds) for health check operations
 }
 
 func (k publisherKafka) ToKafkaConfigMap() *confluent.ConfigMap {
@@ -43,9 +52,15 @@ func dynamicKafkaClientConfigLoad() []byte {
 func publisherKafkaConfigLoader() {
 	viper.SetDefault("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "100000")
 	viper.SetDefault("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS", "1000")
+	viper.SetDefault("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME", "clickstream-test-log")
+	viper.SetDefault("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS", "5000")
 	viper.MergeConfig(bytes.NewBuffer(dynamicKafkaClientConfigLoad()))
 
 	PublisherKafka = publisherKafka{
 		FlushInterval: util.MustGetInt("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS"),
+		HealthCheckConfig: healthcheck{
+			TopicName: util.MustGetString("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME"),
+			TimeOut:   util.MustGetInt("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS"),
+		},
 	}
 }

config/server.go

@@ -10,6 +10,7 @@ import (
 var Server server
 var ServerWs serverWs
 var ServerGRPC serverGRPC
+var ServerMQTT serverMQTT
 
 type server struct {
 	DedupEnabled bool
@@ -37,6 +38,39 @@ type serverGRPC struct {
 	TLSPublicKey string
 }
 
+// serverMQTT represents the complete configuration for an MQTT server setup.
+// It includes authentication, Consul configuration, and consumer-specific settings.
+type serverMQTT struct {
+	ConsulConfig   consul   // Configuration related to Consul service discovery
+	AuthConfig     auth     // MQTT authentication credentials
+	ConsumerConfig consumer // Consumer behavior and connection settings
+	ConnGroup      string   // Connection group name used for identifying MQTT client group
+}
+
+// auth defines MQTT authentication credentials.
+type auth struct {
+	Username string // Username for authenticating with the MQTT broker
+	Password string // Password for authenticating with the MQTT broker
+}
+
+// consul holds configuration details for connecting and interacting with a Consul agent.
+type consul struct {
+	Address    string        // Address of the Consul agent (e.g., localhost:8500)
+	HealthOnly bool          // When true, only healthy service instances are used
+	KVKey      string        // Key in Consul KV store for retrieving MQTT Broker Address
+	WaitTime   time.Duration // Maximum wait time for Consul blocking queries
+}
+
+// consumer contains configuration parameters controlling MQTT message consumption.
+type consumer struct {
+	RetryIntervalInSec time.Duration // Time interval (in seconds) before retrying connection
+	LogLevel           string        // Log verbosity level (e.g., "info", "debug", "error")
+	WriteTimeoutInSec  time.Duration // Timeout duration (in seconds) for write operations
+	PoolSize           int           // Number of concurrent consumers to consume from MQTT topic
+	TopicFormat        string        // Format or pattern for subscribing to MQTT topics (e.g., "share/raccoon/{service}")
+	KeepAlive          time.Duration // Amount of time that the client should wait before sending a PING request to the broker
+}
+
 func serverConfigLoader() {
 	viper.SetDefault("SERVER_BATCH_DEDUP_IN_CONNECTION_ENABLED", "false")
 	Server = server{
@@ -85,3 +119,41 @@ func serverGRPCConfigLoader() {
 		TLSPublicKey: util.MustGetString("SERVER_GRPC_TLS_PUBLIC_KEY"),
 	}
 }
+
+func serverMQTTConfigLoader() {
+	viper.SetDefault("SERVER_MQTT_CONSUL_ADDRESS", "consul:8081")
+	viper.SetDefault("SERVER_MQTT_CONSUL_KV_KEY", "kv/path")
+	viper.SetDefault("SERVER_MQTT_CONSUL_HEALTH_ONLY", true)
+	viper.SetDefault("SERVER_MQTT_CONSUL_WAIT_TIME", 300)
+	viper.SetDefault("SERVER_MQTT_AUTH_USERNAME", "test")
+	viper.SetDefault("SERVER_MQTT_AUTH_PASSWORD", "pass")
+	viper.SetDefault("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_KEEP_ALIVE_IN_SEC", 45)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_LOG_LEVEL", "warn")
+	viper.SetDefault("SERVER_MQTT_CONSUMER_POOL_SIZE", 1)
+	viper.SetDefault("SERVER_MQTT_CONSUMER_TOPIC_FORMAT", "default-topic")
+	viper.SetDefault("SERVER_MQTT_CONNECTION_GROUP", "default")
+
+	ServerMQTT = serverMQTT{
+		ConsulConfig: consul{
+			Address:    util.MustGetString("SERVER_MQTT_CONSUL_ADDRESS"),
+			HealthOnly: util.MustGetBool("SERVER_MQTT_CONSUL_HEALTH_ONLY"),
+			KVKey:      util.MustGetString("SERVER_MQTT_CONSUL_KV_KEY"),
+			WaitTime:   util.MustGetDuration("SERVER_MQTT_CONSUL_WAIT_TIME", time.Second),
+		},
+		AuthConfig: auth{
+			Username: util.MustGetString("SERVER_MQTT_AUTH_USERNAME"),
+			Password: util.MustGetString("SERVER_MQTT_AUTH_PASSWORD"),
+		},
+		ConsumerConfig: consumer{
+			RetryIntervalInSec: util.MustGetDuration("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", time.Second),
+			LogLevel:           util.MustGetString("SERVER_MQTT_CONSUMER_LOG_LEVEL"),
+			WriteTimeoutInSec:  util.MustGetDuration("SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC", time.Second),
+			PoolSize:           util.MustGetInt("SERVER_MQTT_CONSUMER_POOL_SIZE"),
+			TopicFormat:        util.MustGetString("SERVER_MQTT_CONSUMER_TOPIC_FORMAT"),
+			KeepAlive:          util.MustGetDuration("SERVER_MQTT_CONSUMER_KEEP_ALIVE_IN_SEC", time.Second),
+		},
+		ConnGroup: util.MustGetString("SERVER_MQTT_CONNECTION_GROUP"),
+	}
+}