diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 2b9d7717e..807188495 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -282,15 +282,15 @@ func run() error { // Each factory is constructed once and threaded through every consumer of // it, so a real (stateful) backend introduced later is shared rather than // silently duplicated across controllers. - scf := fakeSourceControlFactory{} + sourceControl := fakeSourceControlFactory{} brf := fakeBuildRunnerFactory{} storageFty := storageFactory{backend: store} - primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, scf, brf) + primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf) if err != nil { return err } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry) + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl) if err != nil { return err } @@ -318,7 +318,7 @@ func run() error { logger.Sugar(), scope, newInMemoryCounterFactory(), - scf, + sourceControl, storageFty, registry, ) @@ -398,7 +398,7 @@ func registerPrimaryControllers( scope tally.Scope, store storage.Factory, registry consumer.TopicRegistry, - scf sourcecontrol.Factory, + sourceControl sourcecontrol.Factory, brf buildrunner.Factory, ) (int, error) { var count int @@ -408,7 +408,7 @@ func registerPrimaryControllers( scope, store, queueconfigdefault.NewStore(), - scf, + sourceControl, registry, stovepipemq.TopicKeyProcess, "stovepipe-process", @@ -430,7 +430,7 @@ func registerPrimaryControllers( } count++ - recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record") + recordController := record.NewController(logger, scope, store, sourceControl, stovepipemq.TopicKeyRecord, "stovepipe-record") if err := c.Register(recordController); err != nil { return count, fmt.Errorf("failed to register record controller: %w", err) } @@ -447,6 +447,7 @@ func registerDLQControllers( scope tally.Scope, store storage.Factory, registry consumer.TopicRegistry, + sourceControl sourcecontrol.Factory, ) (int, error) { var count int @@ -468,6 +469,12 @@ func registerDLQControllers( } count++ + recordDLQController := record.NewController(logger, scope, store, sourceControl, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq") + if err := c.Register(recordDLQController); err != nil { + return count, fmt.Errorf("failed to register record dlq controller: %w", err) + } + count++ + return count, nil } @@ -529,6 +536,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Queue: q, Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildsignal-dlq"), }, + { + Key: dlq.TopicKey(stovepipemq.TopicKeyRecord), + Name: "record_dlq", + Queue: q, + Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-record-dlq"), + }, }) } diff --git a/stovepipe/controller/record/BUILD.bazel b/stovepipe/controller/record/BUILD.bazel index 154efed3f..c131992f3 100644 --- a/stovepipe/controller/record/BUILD.bazel +++ b/stovepipe/controller/record/BUILD.bazel @@ -24,6 +24,7 @@ go_test( embed = [":go_default_library"], deps = [ "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index a14120f58..5f7727605 100644 --- a/stovepipe/controller/record/record.go +++ b/stovepipe/controller/record/record.go @@ -44,12 +44,12 @@ import ( // when that fact is green advances the queue's last-green bookmark and promotes // the commit. Implements consumer.Controller. type Controller struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - stores storage.Factory - sourceControls sourcecontrol.Factory - topicKey consumer.TopicKey - consumerGroup string + logger *zap.SugaredLogger + metricsScope tally.Scope + stores storage.Factory + sourceControl sourcecontrol.Factory + topicKey consumer.TopicKey + consumerGroup string } // Verify Controller implements consumer.Controller interface at compile time. @@ -68,17 +68,18 @@ func NewController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, - sourceControls sourcecontrol.Factory, + sourceControl sourcecontrol.Factory, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { + name := string(topicKey) + "_controller" return &Controller{ - logger: logger.Named("record_controller"), - metricsScope: scope.SubScope("record_controller"), - stores: stores, - sourceControls: sourceControls, - topicKey: topicKey, - consumerGroup: consumerGroup, + logger: logger.Named(name), + metricsScope: scope.SubScope(name), + stores: stores, + sourceControl: sourceControl, + topicKey: topicKey, + consumerGroup: consumerGroup, } } @@ -246,7 +247,7 @@ func (c *Controller) reportFailureDetectionLatency(ctx context.Context, request return } - sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue}) + sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { c.failureDetectionUnobserved(request, "resolve_source_control", err) return @@ -365,7 +366,7 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) { queueTag := metrics.NewTag("queue", request.Queue) - sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue}) + sourceControl, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, queueTag) c.logger.Warnw("failed to resolve source control to report the last green timestamp", @@ -420,7 +421,7 @@ func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity. // harmlessly. A commit that a rewritten history dropped from the ref cannot be // promoted by any retry, so that case is counted and skipped rather than failed. func (c *Controller) promote(ctx context.Context, request entity.Request) error { - sc, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue}) + sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue}) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, metrics.NewTag("stage", "resolve"), @@ -477,7 +478,7 @@ func (c *Controller) loadRequest(ctx context.Context, store storage.Storage, id // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { - return "record" + return string(c.topicKey) } // TopicKey returns the topic key this controller subscribes to. diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 404f3fc62..02aeda08a 100644 --- a/stovepipe/controller/record/record_test.go +++ b/stovepipe/controller/record/record_test.go @@ -24,6 +24,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" @@ -94,6 +95,10 @@ func (failingSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.Sour } func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMocks) { + return newControllerForTopic(t, ctrl, stovepipemq.TopicKeyRecord, "stovepipe-record") +} + +func newControllerForTopic(t *testing.T, ctrl *gomock.Controller, topicKey consumer.TopicKey, consumerGroup string) (*Controller, recordMocks) { t.Helper() scope := tally.NewTestScope("", nil) @@ -115,12 +120,20 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMo scope, staticStorageFactory{store: store}, staticSourceControlFactory{sourceControl: m.sourceControl}, - stovepipemq.TopicKeyRecord, - "stovepipe-record", + topicKey, + consumerGroup, ) return c, m } +func TestControllerIdentity(t *testing.T) { + c, _ := newControllerForTopic(t, gomock.NewController(t), consumer.TopicKey("record_dlq"), "stovepipe-record-dlq") + + assert.Equal(t, "record_dlq", c.Name()) + assert.Equal(t, consumer.TopicKey("record_dlq"), c.TopicKey()) + assert.Equal(t, "stovepipe-record-dlq", c.ConsumerGroup()) +} + func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { t.Helper() d := consumermock.NewMockDelivery(ctrl) @@ -283,7 +296,7 @@ func TestProcess_TimestampReportingFailureDoesNotFailRecord(t *testing.T) { func TestProcess_UnresolvableSourceControlCountsTimestampFailure(t *testing.T) { ctrl := gomock.NewController(t) c, m := newController(t, ctrl) - c.sourceControls = failingSourceControlFactory{} + c.sourceControl = failingSourceControlFactory{} m.reqStore.EXPECT().Get(gomock.Any(), testID). Return(requestWithState(entity.RequestStateSucceeded), nil) @@ -385,7 +398,7 @@ func TestProcess_UnobservableDetectionLatencyDoesNotFailRecord(t *testing.T) { name: "source control cannot be resolved", step: "resolve_source_control", setup: func(c *Controller, _ recordMocks) { - c.sourceControls = failingSourceControlFactory{} + c.sourceControl = failingSourceControlFactory{} }, }, { @@ -566,7 +579,7 @@ func TestProcess_PromotionErrorsPropagate(t *testing.T) { { name: "source control resolve fails", setup: func(c *Controller, _ recordMocks) { - c.sourceControls = failingSourceControlFactory{} + c.sourceControl = failingSourceControlFactory{} }, }, {