Add metrics for PubSub and update docs (#349)

Author: srijan-27

docs/quick-start/observability/page.md

@@ -73,11 +73,27 @@ GoFr by default publishes metrics automatically to port: _2121_ on _/metrics_ en
 ---
 * app_sql_stats
 * histogram
-* Observes the response time for SQL queries
+* Response time of SQL queries in microseconds
 ---
 * app_redis_stats
 * histogram
-* Observes the response time for Redis commands
+* Response time of Redis commands in microseconds
+---
+* app_pubsub_publish_total_count
+* counter
+* Number of total publish operations
+---
+* app_pubsub_publish_success_count
+* counter
+* Number of successful publish operations
+---
+* app_pubsub_subscribe_total_count
+* counter
+* Number of total subscribe operations
+---
+* app_pubsub_subscribe_success_count
+* counter
+* Number of successful subscribe operations
 
 {% /table %}
 

pkg/gofr/container/container.go

@@ -87,13 +87,13 @@ func (c *Container) Create(conf config.Config) {
 				Partition:       partition,
 				ConsumerGroupID: conf.Get("CONSUMER_ID"),
 				OffSet:          offSet,
-			}, c.Logger)
+			}, c.Logger, c.metricsManager)
 		}
 	case "GOOGLE":
 		c.pubsub = google.New(google.Config{
 			ProjectID:        conf.Get("GOOGLE_PROJECT_ID"),
 			SubscriptionName: conf.Get("GOOGLE_SUBSCRIPTION_NAME"),
-		}, c.Logger)
+		}, c.Logger, c.metricsManager)
 	}
 }
 
@@ -129,6 +129,12 @@ func (c *Container) registerFrameworkMetrics() {
 	c.Metrics().NewHistogram("app_sql_stats", "Response time of SQL queries in microseconds.", sqlBuckets...)
 	c.Metrics().NewGauge("app_sql_open_connections", "Number of open SQL connections.")
 	c.Metrics().NewGauge("app_sql_inUse_connections", "Number of inUse SQL connections.")
+
+	// pubsub metrics
+	c.Metrics().NewCounter("app_pubsub_publish_total_count", "Number of total publish operations.")
+	c.Metrics().NewCounter("app_pubsub_publish_success_count", "Number of successful publish operations.")
+	c.Metrics().NewCounter("app_pubsub_subscribe_total_count", "Number of total subscribe operations.")
+	c.Metrics().NewCounter("app_pubsub_subscribe_success_count", "Number of successful subscribe operations.")
 }
 
 func (c *Container) GetAppName() string {

pkg/gofr/datasource/pubsub/google/google.go

@@ -23,12 +23,13 @@ type Config struct {
 type googleClient struct {
 	Config
 
-	client *gcPubSub.Client
-	logger pubsub.Logger
+	client  *gcPubSub.Client
+	logger  pubsub.Logger
+	metrics Metrics
 }
 
 //nolint:revive // We do not want anyone using the client without initialization steps.
-func New(conf Config, logger pubsub.Logger) *googleClient {
+func New(conf Config, logger pubsub.Logger, metrics Metrics) *googleClient {
 	err := validateConfigs(&conf)
 	if err != nil {
 		logger.Errorf("google pubsub could not be configured, err : %v", err)
@@ -44,9 +45,10 @@ func New(conf Config, logger pubsub.Logger) *googleClient {
 	}
 
 	return &googleClient{
-		Config: conf,
-		client: client,
-		logger: logger,
+		Config:  conf,
+		client:  client,
+		logger:  logger,
+		metrics: metrics,
 	}
 }
 
@@ -63,6 +65,8 @@ func validateConfigs(conf *Config) error {
 }
 
 func (g *googleClient) Publish(ctx context.Context, topic string, message []byte) error {
+	g.metrics.IncrementCounter(ctx, "app_pubsub_publish_total_count", "topic", topic)
+
 	t, err := g.getTopic(ctx, topic)
 	if err != nil {
 		return err
@@ -80,10 +84,14 @@ func (g *googleClient) Publish(ctx context.Context, topic string, message []byte
 
 	g.logger.Debugf("published google message %v on topic %v", string(message), topic)
 
+	g.metrics.IncrementCounter(ctx, "app_pubsub_publish_success_count", "topic", topic)
+
 	return nil
 }
 
 func (g *googleClient) Subscribe(ctx context.Context, topic string) (*pubsub.Message, error) {
+	g.metrics.IncrementCounter(ctx, "app_pubsub_subscribe_total_count", "topic", topic)
+
 	var m = pubsub.NewMessage(ctx)
 
 	t, err := g.getTopic(ctx, topic)
@@ -118,6 +126,8 @@ func (g *googleClient) Subscribe(ctx context.Context, topic string) (*pubsub.Mes
 
 	g.logger.Debugf("received google message %v on topic %v", string(m.Value), m.Topic)
 
+	g.metrics.IncrementCounter(ctx, "app_pubsub_subscribe_success_count", "topic", topic)
+
 	return m, nil
 }
 

pkg/gofr/datasource/pubsub/google/google_test.go

@@ -7,6 +7,7 @@ import (
 	gcPubSub "cloud.google.com/go/pubsub"
 	"cloud.google.com/go/pubsub/pstest"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/mock/gomock"
 	"google.golang.org/api/option"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -35,10 +36,13 @@ func TestGoogleClient_New_Error(t *testing.T) {
 		g *googleClient
 	)
 
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
 	out := testutil.StderrOutputForFunc(func() {
 		logger := testutil.NewMockLogger(testutil.ERRORLOG)
 
-		g = New(Config{}, logger)
+		g = New(Config{}, logger, NewMockMetrics(ctrl))
 	})
 
 	assert.Nil(t, g)
@@ -49,6 +53,11 @@ func TestGoogleClient_Publish_Success(t *testing.T) {
 	client := getGoogleClient(t)
 	defer client.Close()
 
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockMetrics := NewMockMetrics(ctrl)
+
 	topic := "test-topic"
 	message := []byte("test message")
 	expectedLog := "published google message test message on topic test-topic\n"
@@ -61,8 +70,12 @@ func TestGoogleClient_Publish_Success(t *testing.T) {
 				ProjectID:        "test",
 				SubscriptionName: "sub",
 			},
+			metrics: mockMetrics,
 		}
 
+		mockMetrics.EXPECT().IncrementCounter(context.Background(), "app_pubsub_publish_total_count", "topic", topic)
+		mockMetrics.EXPECT().IncrementCounter(context.Background(), "app_pubsub_publish_success_count", "topic", topic)
+
 		err := g.Publish(context.Background(), topic, message)
 
 		assert.Nil(t, err)
@@ -72,16 +85,23 @@ func TestGoogleClient_Publish_Success(t *testing.T) {
 }
 
 func TestGoogleClient_PublishTopic_Error(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockMetrics := NewMockMetrics(ctrl)
+
 	g := &googleClient{client: getGoogleClient(t), Config: Config{
 		ProjectID:        "test",
 		SubscriptionName: "sub",
-	}}
+	}, metrics: mockMetrics}
 	defer g.client.Close()
 
 	ctx, cancel := context.WithCancel(context.Background())
 
 	cancel()
 
+	mockMetrics.EXPECT().IncrementCounter(ctx, "app_pubsub_publish_total_count", "topic", "test-topic")
+
 	err := g.Publish(ctx, "test-topic", []byte(""))
 	if assert.Error(t, err) {
 		assert.Contains(t, err.Error(), "context canceled")

pkg/gofr/datasource/pubsub/google/metrics.go

@@ -0,0 +1,7 @@
+package google
+
+import "context"
+
+type Metrics interface {
+	IncrementCounter(ctx context.Context, name string, labels ...string)
+}

pkg/gofr/datasource/pubsub/google/mock_metrics.go

@@ -29,12 +29,13 @@ type kafkaClient struct {
 	reader map[string]Reader
 	mu     *sync.RWMutex
 
-	logger pubsub.Logger
-	config Config
+	logger  pubsub.Logger
+	config  Config
+	metrics Metrics
 }
 
 //nolint:revive // We do not want anyone using the client without initialization steps.
-func New(conf Config, logger pubsub.Logger) *kafkaClient {
+func New(conf Config, logger pubsub.Logger, metrics Metrics) *kafkaClient {
 	err := validateConfigs(conf)
 	if err != nil {
 		logger.Errorf("could not initialize kafka, err : %v", err)
@@ -55,12 +56,13 @@ func New(conf Config, logger pubsub.Logger) *kafkaClient {
 	reader := make(map[string]Reader)
 
 	return &kafkaClient{
-		config: conf,
-		dialer: dialer,
-		reader: reader,
-		logger: logger,
-		writer: writer,
-		mu:     &sync.RWMutex{},
+		config:  conf,
+		dialer:  dialer,
+		reader:  reader,
+		logger:  logger,
+		writer:  writer,
+		mu:      &sync.RWMutex{},
+		metrics: metrics,
 	}
 }
 
@@ -77,6 +79,8 @@ func validateConfigs(conf Config) error {
 }
 
 func (k *kafkaClient) Publish(ctx context.Context, topic string, message []byte) error {
+	k.metrics.IncrementCounter(ctx, "app_pubsub_publish_total_count", "topic", topic)
+
 	if k.writer == nil || topic == "" {
 		return errPublisherNotConfigured
 	}
@@ -96,10 +100,14 @@ func (k *kafkaClient) Publish(ctx context.Context, topic string, message []byte)
 
 	k.logger.Debugf("published kafka message %v on topic %v", string(message), topic)
 
+	k.metrics.IncrementCounter(ctx, "app_pubsub_publish_success_count", "topic", topic)
+
 	return nil
 }
 
 func (k *kafkaClient) Subscribe(ctx context.Context, topic string) (*pubsub.Message, error) {
+	k.metrics.IncrementCounter(ctx, "app_pubsub_subscribe_total_count", "topic", topic)
+
 	var reader Reader
 	// Lock the reader map to ensure only one subscriber access the reader at a time
 	k.mu.Lock()
@@ -130,6 +138,8 @@ func (k *kafkaClient) Subscribe(ctx context.Context, topic string) (*pubsub.Mess
 
 	k.logger.Debugf("received kafka message %v on topic %v", string(msg.Value), msg.Topic)
 
+	k.metrics.IncrementCounter(ctx, "app_pubsub_subscribe_success_count", "topic", topic)
+
 	return m, err
 }