fix: review comments

Author: siddhanta-rath

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)
+		}
+	}
+}

config/publisher.go

@@ -13,14 +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
-	HealthCheckConfig healthcheck
+	FlushInterval     int         // Interval (in seconds) to flush the events during shutdown
+	HealthCheckConfig healthcheck // Configuration for Kafka nroker health check
 }
 
+// healthcheck holds settings used to monitor the health of Kafka broker
 type healthcheck struct {
-	TopicName string
-	TimeOut   int
+	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 {

config/server.go

@@ -38,31 +38,36 @@ 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
-	AuthConfig     auth
-	ConsumerConfig consumer
-	ConnGroup      string
+	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
-	Password string
+	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
-	HealthOnly bool
-	KVKey      string
-	WaitTime   time.Duration
+	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 configuration or metadata
+	WaitTime   time.Duration // Maximum wait time for Consul blocking queries
 }
 
+// consumer contains configuration parameters controlling MQTT message consumption.
 type consumer struct {
-	RetryIntervalInSec time.Duration
-	LogLevel           string
-	WriteTimeoutInSec  time.Duration
-	PoolSize           int
-	TopicFormat        string
+	RetryIntervalInSec time.Duration // Time interval (in seconds) before retrying connection or message processing
+	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 consumer workers or goroutines
+	TopicFormat        string        // Format or pattern for subscribing to MQTT topics (e.g., "topic/{region}/{service}")
 }
 
 func serverConfigLoader() {
@@ -142,7 +147,7 @@ func serverMQTTConfigLoader() {
 		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_POOL_SIZE", time.Second),
+			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"),
 		},

health/registry.go

@@ -1,24 +1,16 @@
 package health
 
-import "sync"
-
 type Checker func() error
 
 var (
-	mu       sync.RWMutex
 	checkers = make(map[string]Checker)
 )
 
 func Register(name string, fn Checker) {
-	mu.Lock()
-	defer mu.Unlock()
 	checkers[name] = fn
 }
 
 func CheckAll() map[string]string {
-	mu.RLock()
-	defer mu.RUnlock()
-
 	results := make(map[string]string)
 	for name, fn := range checkers {
 		if err := fn(); err != nil {

health/registry_test.go

@@ -13,9 +13,6 @@ func TestRegister(t *testing.T) {
 
 	Register("db", mockChecker)
 
-	mu.RLock()
-	defer mu.RUnlock()
-
 	if len(checkers) != 1 {
 		t.Fatalf("expected 1 checker, got %d", len(checkers))
 	}
@@ -80,7 +77,5 @@ func TestCheckAll(t *testing.T) {
 }
 
 func resetCheckers() {
-	mu.Lock()
-	defer mu.Unlock()
 	checkers = make(map[string]Checker)
 }

services/grpc/service.go

@@ -47,6 +47,11 @@ func (s *Service) Shutdown(context.Context) error {
 	return nil
 }
 
+// HealthCheck check for grpc
+func (s *Service) HealthCheck() error {
+	return nil
+}
+
 func newGRPCServer() *grpc.Server {
 	if config.ServerGRPC.TLSEnabled {
 		return grpc.NewServer(grpc.Creds(loadTLSCredentials()))

services/mqtt/handler.go

@@ -41,8 +41,9 @@ func (h *Handler) MQTTHandler(ctx context.Context, c courier.PubSub, message *co
 	}
 	//to be removed post end-to-end test
 	for _, event := range req.Events {
-		log.Infof("MQTT message content post deserialization %v", event)
+		log.Infof("MQTT message content post deserialization event : %v", event)
 	}
+	log.Infof("MQTT message request id %v", req.ReqGuid)
 
 	//instrument the request number
 	metrics.Increment(

services/pprof/service.go

@@ -29,3 +29,8 @@ func (*Service) Name() string {
 func (s *Service) Shutdown(ctx context.Context) error {
 	return s.s.Shutdown(ctx)
 }
+
+// HealthCheck check for pprof
+func (s *Service) HealthCheck() error {
+	return nil
+}

services/rest/service.go

@@ -86,3 +86,8 @@ func (*Service) Name() string {
 func (s *Service) Shutdown(ctx context.Context) error {
 	return s.s.Shutdown(ctx)
 }
+
+// HealthCheck check for rest
+func (s *Service) HealthCheck() error {
+	return nil
+}

services/services.go

@@ -17,15 +17,16 @@ type bootstrapper interface {
 	Init(ctx context.Context) error
 	Shutdown(ctx context.Context) error
 	Name() string
+	HealthCheck() error
 }
 
 type Services struct {
-	b []bootstrapper
+	B []bootstrapper
 }
 
 func (s *Services) Start(ctx context.Context, cancel context.CancelFunc) {
 	logger.Info("starting servers")
-	for _, init := range s.b {
+	for _, init := range s.B {
 		i := init
 		go func() {
 			logger.Infof("%s Server --> startServers", i.Name())
@@ -38,7 +39,7 @@ func (s *Services) Start(ctx context.Context, cancel context.CancelFunc) {
 }
 
 func (s *Services) Shutdown(ctx context.Context) {
-	for _, b := range s.b {
+	for _, b := range s.B {
 		logger.Infof("%s Server --> shutting down", b.Name())
 		b.Shutdown(ctx)
 	}
@@ -47,7 +48,7 @@ func (s *Services) Shutdown(ctx context.Context) {
 func Create(b chan collection.CollectRequest, ctx context.Context) Services {
 	c := collection.NewChannelCollector(b)
 	return Services{
-		b: []bootstrapper{
+		B: []bootstrapper{
 			grpc.NewGRPCService(c),
 			pprof.NewPprofService(),
 			rest.NewRestService(c, ctx),