Merge pull request #73 from gojek/feat/connectionpool-metric

feat: add courier connection pooling metric

Author: dhruvjain99

client.go

@@ -407,3 +407,8 @@ func (c *Client) ConnectTimeout() time.Duration {
 func (c *Client) AckTimeout() time.Duration {
 	return c.options.ackTimeout
 }
+
+// PoolSize returns the size of the connection pool configured for the client
+func (c *Client) PoolSize() int {
+	return c.options.poolSize
+}

docs/docs/sdk/SDK.md

@@ -21,6 +21,7 @@ Package courier contains the client that can be used to interact with the courie
   - [func \(c \*Client\) InfoHandler\(\) http.Handler](#Client.InfoHandler)
   - [func \(c \*Client\) IsConnected\(\) bool](#Client.IsConnected)
   - [func \(c \*Client\) KeepAlive\(\) time.Duration](#Client.KeepAlive)
+  - [func \(c \*Client\) PoolSize\(\) int](#Client.PoolSize)
   - [func \(c \*Client\) Publish\(ctx context.Context, topic string, message interface\{\}, opts ...Option\) error](#Client.Publish)
   - [func \(c \*Client\) Run\(ctx context.Context\) error](#Client.Run)
   - [func \(c \*Client\) Start\(\) error](#Client.Start)
@@ -305,6 +306,15 @@ func (c *Client) KeepAlive() time.Duration
 
 KeepAlive returns the keep alive duration configured for the client
 
+<a name="Client.PoolSize"></a>
+### func \(\*Client\) [PoolSize](https://github.com/gojek/courier-go/blob/main/client.go#L412)
+
+```go
+func (c *Client) PoolSize() int
+```
+
+PoolSize returns the size of the connection pool configured for the client
+
 <a name="Client.Publish"></a>
 ### func \(\*Client\) [Publish](https://github.com/gojek/courier-go/blob/main/client_publish.go#L11)
 

docs/docs/sdk/otelcourier.md

@@ -102,7 +102,7 @@ type BucketBoundaries struct {
 ```
 
 <a name="CourierConfig"></a>
-## type [CourierConfig](https://github.com/gojek/courier-go/blob/main/otelcourier/metric.go#L153-L158)
+## type [CourierConfig](https://github.com/gojek/courier-go/blob/main/otelcourier/metric.go#L153-L159)
 
 
 
@@ -112,6 +112,7 @@ type CourierConfig interface {
     WriteTimeout() time.Duration
     KeepAlive() time.Duration
     AckTimeout() time.Duration
+    PoolSize() int
 }
 ```
 

otelcourier/metric.go

@@ -155,58 +155,65 @@ type CourierConfig interface {
 	WriteTimeout() time.Duration
 	KeepAlive() time.Duration
 	AckTimeout() time.Duration
+	PoolSize() int
 }
 
 func (t *OTel) initCourierConfig(c UseMiddleware) {
-	baseAttrs := append([]attribute.KeyValue{
-		attribute.String("service.name", t.service),
-	}, t.attributes...)
-
-	ctx := context.Background()
-
 	cCfg, ok := c.(CourierConfig)
 	if !ok {
 		return
 	}
 
-	connTimeout := cCfg.ConnectTimeout().Seconds()
-	writeTimeout := cCfg.WriteTimeout().Seconds()
-	keepAlive := cCfg.KeepAlive().Seconds()
-	ackTimeout := cCfg.AckTimeout().Seconds()
+	t.setupCourierMetrics(cCfg)
+}
 
-	connTimeoutGauge, err := t.meter.Float64UpDownCounter(
-		"courier.client.connection_timeout",
-		metric.WithDescription("MQTT connection timeout in seconds"),
-		metric.WithUnit("s"),
-	)
-	if err != nil {
-		panic(err)
-	}
+func (t *OTel) setupCourierMetrics(cCfg CourierConfig) {
+	baseAttrs := append([]attribute.KeyValue{
+		attribute.String("service.name", t.service),
+	}, t.attributes...)
 
-	connTimeoutGauge.Add(ctx, connTimeout, metric.WithAttributes(baseAttrs...))
+	ctx := context.Background()
 
-	writeTimeoutGauge, err := t.meter.Float64UpDownCounter(
-		"courier.client.write_timeout",
-		metric.WithDescription("MQTT write timeout in seconds"),
-		metric.WithUnit("s"),
-	)
-	if err != nil {
-		panic(err)
+	timeoutMetrics := []struct {
+		name        string
+		description string
+		value       float64
+	}{
+		{
+			name:        "courier.client.connection_timeout",
+			description: "MQTT connection timeout in seconds",
+			value:       cCfg.ConnectTimeout().Seconds(),
+		},
+		{
+			name:        "courier.client.write_timeout",
+			description: "MQTT write timeout in seconds",
+			value:       cCfg.WriteTimeout().Seconds(),
+		},
+		{
+			name:        "courier.client.keep_alive",
+			description: "MQTT keep alive in seconds",
+			value:       cCfg.KeepAlive().Seconds(),
+		},
+		{
+			name:        "courier.client.ack_timeout",
+			description: "MQTT ack timeout in seconds",
+			value:       cCfg.AckTimeout().Seconds(),
+		},
 	}
 
-	writeTimeoutGauge.Add(ctx, writeTimeout, metric.WithAttributes(baseAttrs...))
+	for _, tm := range timeoutMetrics {
+		gauge, err := t.meter.Float64UpDownCounter(
+			tm.name,
+			metric.WithDescription(tm.description),
+			metric.WithUnit("s"),
+		)
+		if err != nil {
+			panic(err)
+		}
 
-	keepAliveGauge, err := t.meter.Float64UpDownCounter(
-		"courier.client.keep_alive",
-		metric.WithDescription("MQTT keep alive in seconds"),
-		metric.WithUnit("s"),
-	)
-	if err != nil {
-		panic(err)
+		gauge.Add(ctx, tm.value, metric.WithAttributes(baseAttrs...))
 	}
 
-	keepAliveGauge.Add(ctx, keepAlive, metric.WithAttributes(baseAttrs...))
-
 	versionGauge, err := t.meter.Float64UpDownCounter(
 		"courier.client.library_version",
 		metric.WithDescription("Courier library version"),
@@ -216,17 +223,15 @@ func (t *OTel) initCourierConfig(c UseMiddleware) {
 	}
 
 	versionAttrs := append(baseAttrs, attribute.String("version", courier.Version()))
-
 	versionGauge.Add(ctx, 1.0, metric.WithAttributes(versionAttrs...))
 
-	ackTimeoutGauge, err := t.meter.Float64UpDownCounter(
-		"courier.client.ack_timeout",
-		metric.WithDescription("MQTT ack timeout in seconds"),
-		metric.WithUnit("s"),
+	poolSizeGauge, err := t.meter.Float64UpDownCounter(
+		"courier.client.pool_size",
+		metric.WithDescription("Size of the MQTT connection pool"),
 	)
 	if err != nil {
 		panic(err)
 	}
 
-	ackTimeoutGauge.Add(ctx, ackTimeout, metric.WithAttributes(baseAttrs...))
+	poolSizeGauge.Add(ctx, float64(cCfg.PoolSize()), metric.WithAttributes(baseAttrs...))
 }

otelcourier/metric_test.go

@@ -46,6 +46,7 @@ func Test_courierConfigMetrics(t *testing.T) {
 		courier.WithConnectTimeout(30),
 		courier.WithKeepAlive(60),
 		courier.WithWriteTimeout(30),
+		courier.WithPoolSize(3),
 	)
 	require.NoError(t, err)
 
@@ -68,14 +69,18 @@ courier_client_library_version{otel_scope_name="github.com/gojek/courier-go/otel
 # HELP courier_client_write_timeout_seconds MQTT write timeout in seconds
 # TYPE courier_client_write_timeout_seconds gauge
 courier_client_write_timeout_seconds{otel_scope_name="github.com/gojek/courier-go/otelcourier",otel_scope_version="semver:%s",service_name="test-service"} 3e-08
-`, vsn, vsn, vsn, vsn, vsn, vsn))
+# HELP courier_client_pool_size Size of the MQTT connection pool
+# TYPE courier_client_pool_size gauge
+courier_client_pool_size{otel_scope_name="github.com/gojek/courier-go/otelcourier",otel_scope_version="semver:%s",service_name="test-service"} 3
+`, vsn, vsn, vsn, vsn, vsn, vsn, vsn))
 
 		assert.NoError(t, testutil.GatherAndCompare(reg, buf,
 			"courier_client_connection_timeout_seconds",
 			"courier_client_write_timeout_seconds",
 			"courier_client_keep_alive_seconds",
 			"courier_client_ack_timeout_seconds",
 			"courier_client_library_version",
+			"courier_client_pool_size",
 		))
 	})
 }
@@ -102,6 +107,7 @@ func Test_courierConfigMetrics_NonCourierConfigType(t *testing.T) {
 		assert.NotContains(t, metricName, "courier_client_keep_alive")
 		assert.NotContains(t, metricName, "courier_client_ack_timeout")
 		assert.NotContains(t, metricName, "courier_client_library_version")
+		assert.NotContains(t, metricName, "courier_client_pool_size")
 	}
 }