enhance(metrics): gather queued builds

api/metrics.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/go-vela/server/database"
+	"github.com/go-vela/server/queue"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -29,6 +30,8 @@ type MetricsQueryParameters struct {
 	RunningBuildCount bool `form:"running_build_count"`
 	// PendingBuildCount represents total number of builds with status==pending
 	PendingBuildCount bool `form:"pending_build_count"`
+	// QueuedBuildCount represents total number of builds currently in the queue
+	QueuedBuildCount bool `form:"queued_build_count"`
 	// FailureBuildCount represents total number of builds with status==failure
 	FailureBuildCount bool `form:"failure_build_count"`
 	// KilledBuildCount represents total number of builds with status==killed
@@ -258,6 +261,17 @@ func recordGauges(c *gin.Context) {
 		totals.WithLabelValues("build", "status", "pending").Set(float64(bPen))
 	}
 
+	// queued_build_count
+	if q.QueuedBuildCount {
+		// send API call to capture the total number of queued builds
+		t, err := queue.FromContext(c).Length(c)
+		if err != nil {
+			logrus.Errorf("unable to get count of all queued builds: %v", err)
+		}
+
+		totals.WithLabelValues("build", "status", "queued").Set(float64(t))
+	}
+
 	// failure_build_count
 	if q.FailureBuildCount {
 		// send API call to capture the total number of failure builds

queue/redis/length.go

@@ -0,0 +1,27 @@
+// Copyright (c) 2023 Target Brands, Inc. All rights reserved.
+//
+// Use of this source code is governed by the LICENSE file in this repository.
+
+package redis
+
+import (
+	"context"
+)
+
+// Length tallies all items present in the configured channels in the queue.
+func (c *client) Length(ctx context.Context) (int64, error) {
+	c.Logger.Tracef("reading length of all configured channels in queue")
+
+	total := int64(0)
+
+	for _, channel := range c.config.Channels {
+		items, err := c.Redis.LLen(ctx, channel).Result()
+		if err != nil {
+			return 0, err
+		}
+
+		total += items
+	}
+
+	return total, nil
+}

queue/redis/length_test.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2023 Target Brands, Inc. All rights reserved.
+//
+// Use of this source code is governed by the LICENSE file in this repository.
+
+package redis
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"github.com/go-vela/types"
+	"gopkg.in/square/go-jose.v2/json"
+)
+
+func TestRedis_Length(t *testing.T) {
+	// setup types
+	// use global variables in redis_test.go
+	_item := &types.Item{
+		Build:    _build,
+		Pipeline: _steps,
+		Repo:     _repo,
+		User:     _user,
+	}
+
+	// setup queue item
+	bytes, err := json.Marshal(_item)
+	if err != nil {
+		t.Errorf("unable to marshal queue item: %v", err)
+	}
+
+	// setup redis mock
+	_redis, err := NewTest("vela", "vela:second", "vela:third")
+	if err != nil {
+		t.Errorf("unable to create queue service: %v", err)
+	}
+
+	// setup tests
+	tests := []struct {
+		channels []string
+		want     int64
+	}{
+		{
+			channels: []string{"vela"},
+			want:     1,
+		},
+		{
+			channels: []string{"vela", "vela:second", "vela:third"},
+			want:     4,
+		},
+		{
+			channels: []string{"vela", "vela:second", "phony"},
+			want:     6,
+		},
+	}
+
+	// run tests
+	for _, test := range tests {
+		for _, channel := range test.channels {
+			err := _redis.Push(context.Background(), channel, bytes)
+			if err != nil {
+				t.Errorf("unable to push item to queue: %v", err)
+			}
+		}
+		got, err := _redis.Length(context.Background())
+
+		if err != nil {
+			t.Errorf("Length returned err: %v", err)
+		}
+
+		if !reflect.DeepEqual(got, test.want) {
+			t.Errorf("Length is %v, want %v", got, test.want)
+		}
+	}
+}

queue/service.go

@@ -20,6 +20,10 @@ type Service interface {
 	// the configured queue driver.
 	Driver() string
 
+	// Length defines a function that outputs
+	// the length of a queue channel
+	Length(context.Context) (int64, error)
+
 	// Pop defines a function that grabs an
 	// item off the queue.
 	Pop(context.Context) (*types.Item, error)