feat(metrics): wire instrumentation through queue, handlers, middleware

internal/updater/queue.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/metrics"
 )
 
 // Status describes the outcome of executing a single Job.
@@ -30,10 +31,11 @@ type Result struct {
 // It also satisfies the "single global FIFO worker" design choice in
 // spec section 5.7.
 type Queue struct {
-	exec Executor
-	ch   chan submission
-	stop chan struct{}
-	done chan struct{}
+	exec    Executor
+	metrics *metrics.Metrics
+	ch      chan submission
+	stop    chan struct{}
+	done    chan struct{}
 }
 
 type submission struct {
@@ -42,13 +44,15 @@ type submission struct {
 	results chan []Result
 }
 
-// NewQueue constructs a queue bound to the given executor.
-func NewQueue(exec Executor) *Queue {
+// NewQueue constructs a queue bound to the given executor. m may be nil,
+// in which case metrics recording is silently skipped.
+func NewQueue(exec Executor, m *metrics.Metrics) *Queue {
 	return &Queue{
-		exec: exec,
-		ch:   make(chan submission, 16),
-		stop: make(chan struct{}),
-		done: make(chan struct{}),
+		exec:    exec,
+		metrics: m,
+		ch:      make(chan submission, 16),
+		stop:    make(chan struct{}),
+		done:    make(chan struct{}),
 	}
 }
 
@@ -70,6 +74,9 @@ func (q *Queue) Stop() {
 func (q *Queue) Submit(ctx context.Context, jobs []discovery.Job) []Result {
 	resCh := make(chan []Result, 1)
 	q.ch <- submission{ctx: ctx, jobs: jobs, results: resCh}
+	if q.metrics != nil {
+		q.metrics.QueueDepth.Set(float64(len(q.ch)))
+	}
 	return <-resCh
 }
 
@@ -97,6 +104,10 @@ func (q *Queue) runOne(ctx context.Context, job discovery.Job) Result {
 		r.Status = StatusRefused
 		r.Error = job.RefusedReason
 		r.DurationMs = time.Since(start).Milliseconds()
+		if q.metrics != nil {
+			q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(StatusRefused)).Inc()
+			q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(time.Since(start).Seconds())
+		}
 		return r
 	}
 
@@ -112,6 +123,15 @@ func (q *Queue) runOne(ctx context.Context, job discovery.Job) Result {
 		r.Status = StatusFailed
 		r.Error = err.Error()
 	}
+
+	if q.metrics != nil {
+		elapsed := time.Since(start).Seconds()
+		q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(r.Status)).Inc()
+		q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(elapsed)
+		if r.Status == StatusUpdated {
+			q.metrics.LastUpdateTime.WithLabelValues(job.Project, job.Service).SetToCurrentTime()
+		}
+	}
 	return r
 }
 

internal/updater/queue_test.go

@@ -7,7 +7,10 @@ import (
 	"testing"
 	"time"
 
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/metrics"
 	"github.com/shcizo/package-updater/internal/updater"
 	"github.com/stretchr/testify/require"
 )
@@ -44,7 +47,7 @@ func (f *fakeExec) callsCopy() []discovery.Job {
 
 func TestQueue_ProcessesFIFO(t *testing.T) {
 	exec := &fakeExec{delay: 20 * time.Millisecond}
-	q := updater.NewQueue(exec)
+	q := updater.NewQueue(exec, nil)
 	q.Start(context.Background())
 	defer q.Stop()
 
@@ -69,7 +72,7 @@ func TestQueue_ProcessesFIFO(t *testing.T) {
 
 func TestQueue_ReturnsPerJobResults(t *testing.T) {
 	exec := &fakeExec{errForSvc: map[string]error{"failing": errors.New("boom")}}
-	q := updater.NewQueue(exec)
+	q := updater.NewQueue(exec, nil)
 	q.Start(context.Background())
 	defer q.Stop()
 
@@ -90,7 +93,7 @@ func TestQueue_ReturnsPerJobResults(t *testing.T) {
 
 func TestQueue_Timeout(t *testing.T) {
 	exec := &fakeExec{delay: 200 * time.Millisecond}
-	q := updater.NewQueue(exec)
+	q := updater.NewQueue(exec, nil)
 	q.Start(context.Background())
 	defer q.Stop()
 
@@ -103,3 +106,24 @@ func TestQueue_Timeout(t *testing.T) {
 	require.Len(t, results, 1)
 	require.Equal(t, updater.StatusTimeout, results[0].Status)
 }
+
+func TestQueue_RecordsMetrics(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	m := metrics.New(reg)
+	exec := &fakeExec{}
+	q := updater.NewQueue(exec, m)
+	q.Start(context.Background())
+	defer q.Stop()
+
+	q.Submit(context.Background(), []discovery.Job{{Project: "p", Service: "s"}})
+
+	// Verify UpdateJobs counter was incremented for the successful job.
+	metric := &dto.Metric{}
+	require.NoError(t, m.UpdateJobs.WithLabelValues("p", "s", "updated").Write(metric))
+	require.Equal(t, 1.0, metric.GetCounter().GetValue())
+
+	// Verify LastUpdateTime was set (non-zero).
+	tsMetric := &dto.Metric{}
+	require.NoError(t, m.LastUpdateTime.WithLabelValues("p", "s").Write(tsMetric))
+	require.Greater(t, tsMetric.GetGauge().GetValue(), 0.0)
+}

internal/updater/worker_test.go

@@ -21,7 +21,7 @@ func (c *counterExec) Execute(_ context.Context, _ discovery.Job) error {
 
 func TestWorker_RunsExactlyOneAtATime(t *testing.T) {
 	exec := &counterExec{}
-	q := updater.NewQueue(exec)
+	q := updater.NewQueue(exec, nil)
 	q.Start(context.Background())
 	defer q.Stop()