refactor(stovepipe): bind freshness reporters to queues

Resolve storage and source control through a queue-scoped reporter factory so
individual reporters operate on concrete queue dependencies.

Author: mnoah1

stovepipe/controller/record/record.go

@@ -45,7 +45,7 @@ type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	stores        storage.Factory
-	reporter      observability.Reporter
+	reporters     observability.Factory
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -64,10 +64,10 @@ const wholeRepositoryProject = ""
 // Option configures a Controller.
 type Option func(*Controller)
 
-// WithReporter configures best-effort queue observability reporting.
-func WithReporter(reporter observability.Reporter) Option {
+// WithReporterFactory configures best-effort queue observability reporting.
+func WithReporterFactory(reporters observability.Factory) Option {
 	return func(c *Controller) {
-		c.reporter = reporter
+		c.reporters = reporters
 	}
 }
 
@@ -129,8 +129,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
 		return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID)
 	}
-	if c.reporter != nil {
-		defer c.reporter.Report(ctx, request.Queue)
+	if c.reporters != nil {
+		reporter, err := c.reporters.For(observability.Config{QueueName: request.Queue})
+		if err != nil {
+			metrics.NamedCounter(c.metricsScope, _opName, "reporter_resolve_errors", 1)
+		} else {
+			defer reporter.Report(ctx)
+		}
 	}
 
 	switch request.State {

stovepipe/extension/observability/lastgreen/BUILD.bazel

@@ -19,6 +19,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/observability:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/sourcecontrol/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",

stovepipe/extension/observability/lastgreen/lastgreen.go

@@ -3,6 +3,7 @@ package lastgreen
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/uber-go/tally"
@@ -11,32 +12,67 @@ import (
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
+type factory struct {
+	scope          tally.Scope
+	stores         storage.Factory
+	sourceControls sourcecontrol.Factory
+}
+
 type reporter struct {
 	scope         tally.Scope
-	stores        storage.Factory
-	sourceControl sourcecontrol.Factory
+	queue         string
+	store         storage.Storage
+	sourceControl sourcecontrol.SourceControl
 }
 
+var _ observability.Factory = (*factory)(nil)
 var _ observability.Reporter = (*reporter)(nil)
 
+// NewFactory creates queue-scoped last-known-green reporters.
+func NewFactory(
+	scope tally.Scope,
+	stores storage.Factory,
+	sourceControls sourcecontrol.Factory,
+) observability.Factory {
+	return &factory{
+		scope:          scope,
+		stores:         stores,
+		sourceControls: sourceControls,
+	}
+}
+
+// For resolves the concrete dependencies needed to report for one queue.
+func (f *factory) For(cfg observability.Config) (observability.Reporter, error) {
+	store, err := f.stores.For(storage.Config{QueueName: cfg.QueueName})
+	if err != nil {
+		return nil, fmt.Errorf("resolve storage for queue %q: %w", cfg.QueueName, err)
+	}
+	sourceControl, err := f.sourceControls.For(sourcecontrol.Config{QueueName: cfg.QueueName})
+	if err != nil {
+		return nil, fmt.Errorf("resolve source control for queue %q: %w", cfg.QueueName, err)
+	}
+	return New(f.scope, cfg.QueueName, store, sourceControl), nil
+}
+
 // New creates a Reporter for a queue's last-known-green age.
-func New(scope tally.Scope, stores storage.Factory, sourceControl sourcecontrol.Factory) observability.Reporter {
+func New(
+	scope tally.Scope,
+	queue string,
+	store storage.Storage,
+	sourceControl sourcecontrol.SourceControl,
+) observability.Reporter {
 	return &reporter{
 		scope:         scope.SubScope("last_green"),
-		stores:        stores,
+		queue:         queue,
+		store:         store,
 		sourceControl: sourceControl,
 	}
 }
 
 // Report updates the queue's current last-known-green age gauge.
-func (r *reporter) Report(ctx context.Context, queue string) {
-	tags := map[string]string{"queue": queue}
-	store, err := r.stores.For(storage.Config{QueueName: queue})
-	if err != nil {
-		r.error(tags, "resolve_storage")
-		return
-	}
-	queueRow, err := store.GetQueueStore().Get(ctx, queue)
+func (r *reporter) Report(ctx context.Context) {
+	tags := map[string]string{"queue": r.queue}
+	queueRow, err := r.store.GetQueueStore().Get(ctx, r.queue)
 	if err != nil {
 		r.error(tags, "get_queue")
 		return
@@ -45,12 +81,7 @@ func (r *reporter) Report(ctx context.Context, queue string) {
 		r.scope.Tagged(tags).Counter("age_missing").Inc(1)
 		return
 	}
-	control, err := r.sourceControl.For(sourcecontrol.Config{QueueName: queue})
-	if err != nil {
-		r.error(tags, "resolve_source_control")
-		return
-	}
-	info, err := control.ChangeInfo(ctx, queueRow.LastGreenURI)
+	info, err := r.sourceControl.ChangeInfo(ctx, queueRow.LastGreenURI)
 	if err != nil || info.CreatedAt.IsZero() {
 		r.error(tags, "get_change_info")
 		return

stovepipe/extension/observability/lastgreen/lastgreen_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/observability"
 	sourcecontrol "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -39,7 +40,9 @@ func TestReport_EmitsLastGreenAge(t *testing.T) {
 		CreatedAt: createdAt,
 	}, nil)
 
-	New(scope, stores, sourceControls).Report(context.Background(), testQueue)
+	reporter, err := NewFactory(scope, stores, sourceControls).For(observability.Config{QueueName: testQueue})
+	require.NoError(t, err)
+	reporter.Report(context.Background())
 
 	gauge, ok := scope.Snapshot().Gauges()["stovepipe.last_green.age_seconds+queue=monorepo/main"]
 	require.True(t, ok)
@@ -49,15 +52,13 @@ func TestReport_EmitsLastGreenAge(t *testing.T) {
 func TestReport_RecordsMissingLastGreen(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	scope := tally.NewTestScope("stovepipe", nil)
-	stores := storagemock.NewMockFactory(ctrl)
 	store := storagemock.NewMockStorage(ctrl)
 	queueStore := storagemock.NewMockQueueStore(ctrl)
 
-	stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(store, nil)
 	store.EXPECT().GetQueueStore().Return(queueStore)
 	queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue}, nil)
 
-	New(scope, stores, sourcecontrolmock.NewMockFactory(ctrl)).Report(context.Background(), testQueue)
+	New(scope, testQueue, store, sourcecontrolmock.NewMockSourceControl(ctrl)).Report(context.Background())
 
 	counter, ok := scope.Snapshot().Counters()["stovepipe.last_green.age_missing+queue=monorepo/main"]
 	require.True(t, ok)

stovepipe/extension/observability/observability.go

@@ -3,7 +3,17 @@ package observability
 
 import "context"
 
-// Reporter emits best-effort observability data for a queue.
+// Reporter emits best-effort observability data for one queue.
 type Reporter interface {
-	Report(context.Context, string)
+	Report(context.Context)
+}
+
+// Config identifies the queue whose observability reporter is requested.
+type Config struct {
+	QueueName string
+}
+
+// Factory resolves a Reporter bound to one queue.
+type Factory interface {
+	For(Config) (Reporter, error)
 }