feat: adding a queue size metric (#1485)

Co-authored-by: Sven Urbanski <sven.urbanski@freiheit.com>

services/cd-service/pkg/repository/queue.go

@@ -28,6 +28,8 @@ This queue does not improve the latency, because each request still waits for th
 import (
 	"context"
 	"fmt"
+	"github.com/freiheit-com/kuberpult/pkg/logger"
+	"go.uber.org/zap"
 )
 
 type queue struct {
@@ -60,6 +62,7 @@ func (q *queue) add(ctx context.Context, transformers []Transformer) <-chan erro
 
 	select {
 	case q.transformerBatches <- e:
+		GaugeQueueSize(ctx, len(q.transformerBatches))
 		return resultChannel
 	default:
 		//Channel is full, we don't want to put anything else there.
@@ -74,3 +77,13 @@ func makeQueueN(size uint) queue {
 		transformerBatches: make(chan transformerBatch, size),
 	}
 }
+
+func GaugeQueueSize(ctx context.Context, queueSize int) {
+	if ddMetrics != nil {
+		err := ddMetrics.Gauge("request_queue_size", float64(queueSize), []string{}, 1)
+
+		if err != nil {
+			logger.FromContext(ctx).Error("Error gauging queue size metric: ", zap.Error(err))
+		}
+	}
+}

services/cd-service/pkg/repository/repository.go

@@ -556,6 +556,7 @@ func (r *repository) drainQueue() []transformerBatch {
 		select {
 		case f := <-r.queue.transformerBatches:
 			// Check that the item is not already cancelled
+			GaugeQueueSize(f.ctx, len(r.queue.transformerBatches))
 			select {
 			case <-f.ctx.Done():
 				f.finish(f.ctx.Err())