fix: review comments

Author: siddhanta-rath

constant/healthstatus.go

@@ -0,0 +1,23 @@
+package constant
+
+// HealthStatus represents the possible health states of a service
+// or dependency within the system.
+//
+// It is used to standardize how health information is communicated across
+// services — for example, in HTTP health check endpoints (/ping).
+// Defining these values as constants ensures consistent usage throughout the codebase.
+type HealthStatus string
+
+const (
+	// HealthStatusHealthy indicates that the service is operating
+	// normally, and all dependent subsystems are functioning as expected.
+
+	HealthStatusHealthy HealthStatus = "healthy"
+
+	// HealthStatusUnhealthy indicates that the component or service is
+	// degraded or non-functional, possibly due to dependency failures,
+	// connectivity issues, or internal errors.
+	//
+	// This status will cause readiness and liveness  probes to fail
+	HealthStatusUnhealthy HealthStatus = "unhealthy"
+)

health/registry.go

@@ -1,23 +1,41 @@
+// Package health provides a simple health check framework that allows
+// different components or dependencies to register their own health check functions.
+// Each checker function returns an error to indicate an unhealthy state,
+// or nil to indicate that the component is healthy.
 package health
 
+import "github.com/goto/raccoon/constant"
+
+// Checker represents a function that performs a health check for a component.
+// It returns an error if the check fails (unhealthy), or nil if it passes (healthy).
 type Checker func() error
 
 var (
+	// checkers stores all registered health check functions.
+	// The key is the component name, and the value is its corresponding Checker function.
 	checkers = make(map[string]Checker)
 )
 
+// Register adds a new health check function for a specific component.
+// Each component should have a unique name; calling Register with an existing
+// name will overwrite the previous checker.
 func Register(name string, fn Checker) {
 	checkers[name] = fn
 }
 
-func CheckAll() map[string]string {
-	results := make(map[string]string)
+// CheckAll runs all registered health check functions and returns their results.
+// It returns a map where the key is the component name and the value is its health status.
+// Components that return an error are marked as Unhealthy; others are marked as Healthy.
+func CheckAll() map[string]constant.HealthStatus {
+	results := make(map[string]constant.HealthStatus)
+
 	for name, fn := range checkers {
 		if err := fn(); err != nil {
-			results[name] = "unhealthy: " + err.Error()
+			results[name] = constant.HealthStatusUnhealthy
 		} else {
-			results[name] = "healthy"
+			results[name] = constant.HealthStatusHealthy
 		}
 	}
+
 	return results
 }

health/registry_test.go

@@ -4,6 +4,8 @@ import (
 	"errors"
 	"reflect"
 	"testing"
+
+	"github.com/goto/raccoon/constant"
 )
 
 func TestRegister(t *testing.T) {
@@ -26,24 +28,24 @@ func TestCheckAll(t *testing.T) {
 	tests := []struct {
 		name     string
 		register map[string]Checker
-		expected map[string]string
+		expected map[string]constant.HealthStatus
 	}{
 		{
 			name: "single healthy service on register map",
 			register: map[string]Checker{
 				"serviceA": func() error { return nil },
 			},
-			expected: map[string]string{
-				"serviceA": "healthy",
+			expected: map[string]constant.HealthStatus{
+				"serviceA": constant.HealthStatusHealthy,
 			},
 		},
 		{
 			name: "single unhealthy service on register map",
 			register: map[string]Checker{
 				"serviceB": func() error { return errors.New("timeout") },
 			},
-			expected: map[string]string{
-				"serviceB": "unhealthy: timeout",
+			expected: map[string]constant.HealthStatus{
+				"serviceB": constant.HealthStatusUnhealthy,
 			},
 		},
 		{
@@ -52,16 +54,16 @@ func TestCheckAll(t *testing.T) {
 				"ok":   func() error { return nil },
 				"fail": func() error { return errors.New("db down") },
 			},
-			expected: map[string]string{
-				"ok":   "healthy",
-				"fail": "unhealthy: db down",
+			expected: map[string]constant.HealthStatus{
+				"ok":   constant.HealthStatusHealthy,
+				"fail": constant.HealthStatusUnhealthy,
 			},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			//rest the register map for every test scenario
+			// reset the register map for every test scenario
 			resetCheckers()
 
 			for name, fn := range tt.register {
@@ -76,6 +78,7 @@ func TestCheckAll(t *testing.T) {
 	}
 }
 
+// resetCheckers clears all registered checkers to ensure isolation between tests.
 func resetCheckers() {
 	checkers = make(map[string]Checker)
 }

services/mqtt/client/pubsub.go

@@ -8,6 +8,7 @@ import (
 	"github.com/gojekfarm/xtools/xproto"
 	"github.com/goto/raccoon/config"
 	"github.com/goto/raccoon/logger"
+	"github.com/goto/raccoon/metrics"
 	"io"
 )
 
@@ -17,13 +18,30 @@ type MqttPubSubClient struct {
 	resolver resolver
 }
 
+// client defines the minimal interface for a courier MQTT client instance.
+// It abstracts basic lifecycle operations and connection state management.
 type client interface {
+	// Start begins the client’s operation and establishes connections
+	// to the MQTT broker or messaging backend.
 	Start() error
+
+	// Stop terminates the client’s operation gracefully, releasing
+	// any resources and closing network connections.
 	Stop()
+
+	// IsConnected reports whether the client is currently connected
+	// to the MQTT broker.
 	IsConnected() bool
 }
 
+// resolver defines the interface for a service discovery resolver
+// responsible for discovering and updating MQTT broker endpoints dynamically.
+//
+// Implementations like ConsulResolver use this interface to periodically
+// fetch and update broker addresses for high availability and fault tolerance.
 type resolver interface {
+	// Start initiates the resolver’s background process to watch for
+	// endpoint changes.
 	Start()
 }
 
@@ -78,6 +96,10 @@ func registerHandler(ctx context.Context, handler courier.MessageHandler) func(c
 	return func(ps courier.PubSub) {
 		topic := config.ServerMQTT.ConsumerConfig.TopicFormat
 		if err := ps.Subscribe(ctx, topic, handler, courier.QOSZero); err != nil {
+			metrics.Increment(
+				"mqtt_error",
+				fmt.Sprintf("reason=subscribe_failed"),
+			)
 			logger.Errorf("failed to register MQTT handler for topic %q: %v", topic, err)
 		} else {
 			logger.Infof("successfully registered MQTT handler for topic %q", topic)
@@ -89,6 +111,10 @@ func registerHandler(ctx context.Context, handler courier.MessageHandler) func(c
 func (m *MqttPubSubClient) Start() error {
 	go m.resolver.Start()
 	if err := m.client.Start(); err != nil {
+		metrics.Increment(
+			"mqtt_error",
+			fmt.Sprintf("reason=start_failed"),
+		)
 		logger.Infof("MQTT client start failed due to %v", err)
 		return fmt.Errorf("failed to start MQTT client: %w", err)
 	}

services/mqtt/handler.go

@@ -4,15 +4,15 @@ import (
 	pb "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/raccoon/v1beta1"
 	"context"
 	"fmt"
-	"github.com/goto/raccoon/clients/go/log"
-	"github.com/goto/raccoon/serialization"
 	"time"
 
 	"github.com/gojek/courier-go"
+	"github.com/goto/raccoon/clients/go/log"
 	"github.com/goto/raccoon/collection"
 	"github.com/goto/raccoon/config"
 	"github.com/goto/raccoon/identification"
 	"github.com/goto/raccoon/metrics"
+	"github.com/goto/raccoon/serialization"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -25,69 +25,81 @@ type Handler struct {
 // and sends them to the Collector.
 func (h *Handler) MQTTHandler(ctx context.Context, c courier.PubSub, message *courier.Message) {
 	start := time.Now()
-
-	identifier := identification.Identifier{
-		Group: config.ServerMQTT.ConnGroup,
-	}
+	group := config.ServerMQTT.ConnGroup
 	log.Infof("MQTT message received with content %v", message)
 
 	var req pb.SendEventRequest
 	if err := message.DecodePayload(&req); err != nil {
-		metrics.Increment(
-			"batches_read_total",
-			fmt.Sprintf("status=failed,conn_group=%s,reason=serde", identifier.Group),
-		)
-		log.Errorf("mqtt message decoding failed due to : %v", err)
+		h.recordMetrics("request", fmt.Sprintf("status=failed,conn_group=%s,reason=serde", group), nil)
+		log.Errorf("mqtt message decoding failed: %v", err)
 		return
 	}
 
 	if proto.Equal(&req, &pb.SendEventRequest{}) {
-		metrics.Increment(
-			"batches_read_total",
-			fmt.Sprintf("status=failed,conn_group=%s,reason=empty", identifier.Group),
-		)
+		h.recordMetrics("request", fmt.Sprintf("status=failed,conn_group=%s,reason=empty", group), nil)
 		log.Errorf("mqtt request message according proto format is empty")
 		return
 	}
 
-	//to be removed post end-to-end test
+	// Debug — can be removed after E2E verification
 	for _, event := range req.Events {
-		log.Infof("MQTT message content post deserialization event : %v", event)
+		log.Infof("MQTT message content post deserialization event: %v", event)
 	}
-	log.Infof("MQTT message request id %v", req.ReqGuid)
+	log.Infof("MQTT message request id: %v", req.ReqGuid)
 
-	//instrument the request number
-	metrics.Increment(
-		"batches_read_total",
-		fmt.Sprintf("status=success,conn_group=%s", identifier.Group),
-	)
-	//instrument the request size
+	// Serialize to compute request size
 	reqBytes, err := serialization.SerializeProto(&req)
 	if err != nil {
-		log.Errorf("mqtt message serialization failed : %v", err)
-	} else {
-		metrics.Count("request_bytes_total", len(reqBytes),
-			fmt.Sprintf("conn_group=%s", identifier.Group))
+		log.Errorf("mqtt message serialization failed: %v", err)
 	}
 
-	h.recordEventMetrics(req.Events, identifier.Group)
+	// Record all metrics via generic function
+	h.recordMetrics("request", fmt.Sprintf("status=success,conn_group=%s", group), reqBytes)
+	h.recordMetrics("event", fmt.Sprintf("conn_group=%s", group), req.Events)
 
 	h.Collector.Collect(ctx, &collection.CollectRequest{
-		ConnectionIdentifier: identifier,
+		ConnectionIdentifier: identification.Identifier{Group: group},
 		TimeConsumed:         start,
 		SendEventRequest:     &req,
 		AckFunc:              nil,
 	})
 }
 
-// recordEventMetrics updates per-event metrics like byte size and event count.
-func (h *Handler) recordEventMetrics(events []*pb.Event, group string) {
+// recordMetrics is a generic entry function that routes metric recording
+// based on metricName.
+func (h *Handler) recordMetrics(metricName string, tags string, data any) {
+	switch metricName {
+	case "request":
+		h.recordRequestMetrics(tags, data)
+	case "event":
+		h.recordEventMetrics(tags, data)
+	default:
+		log.Errorf("unknown metricName=%s ignored", metricName)
+	}
+}
+
+// recordRequestMetrics captures request-level metrics (success/failure, bytes).
+func (h *Handler) recordRequestMetrics(tags string, data any) {
+	metrics.Increment("batches_read_total", tags)
+
+	if reqBytes, ok := data.([]byte); ok && len(reqBytes) > 0 {
+		metrics.Count("request_bytes_total", len(reqBytes), tags)
+	}
+}
+
+// recordEventMetrics captures per-event metrics like count and size.
+func (h *Handler) recordEventMetrics(tags string, data any) {
+	events, ok := data.([]*pb.Event)
+	if !ok {
+		return
+	}
+
 	for _, e := range events {
 		if e == nil {
 			continue
 		}
-		tags := fmt.Sprintf("conn_group=%s,event_type=%s", group, e.Type)
-		metrics.Count("events_rx_bytes_total", len(e.EventBytes), tags)
-		metrics.Increment("events_rx_total", tags)
+		eventTags := fmt.Sprintf("%s,event_type=%s", tags, e.Type)
+		metrics.Increment("events_rx_total", eventTags)
+		metrics.Count("events_rx_bytes_total", len(e.EventBytes), eventTags)
 	}
 }