diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 2ddb3dbbb..2b9d7717e 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -450,7 +450,7 @@ func registerDLQControllers( ) (int, error) { var count int - processDLQController := dlq.NewController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + processDLQController := dlq.NewDLQRequestController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") if err := c.Register(processDLQController); err != nil { return count, fmt.Errorf("failed to register process dlq controller: %w", err) } @@ -462,7 +462,7 @@ func registerDLQControllers( } count++ - buildSignalDLQController := dlq.NewBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq") + buildSignalDLQController := dlq.NewDLQBuildSignalController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyBuildSignal), "stovepipe-buildsignal-dlq") if err := c.Register(buildSignalDLQController); err != nil { return count, fmt.Errorf("failed to register buildsignal dlq controller: %w", err) } diff --git a/stovepipe/controller/dlq/buildsignal.go b/stovepipe/controller/dlq/buildsignal.go index 5de1ba290..c68d2f2d0 100644 --- a/stovepipe/controller/dlq/buildsignal.go +++ b/stovepipe/controller/dlq/buildsignal.go @@ -30,7 +30,7 @@ import ( // _buildSignalOpName is the metric operation name shared by every emit in this file. const _buildSignalOpName = "buildsignal_dlq" -// BuildSignalController is the DLQ reconciler for the buildsignal stage. The +// buildSignalController is the DLQ reconciler for the buildsignal stage. The // payload names a build, not a request, so it takes one more step than the // process reconciler: read the build to get its RequestID, then fail that // request via failRequest. @@ -49,7 +49,7 @@ const _buildSignalOpName = "buildsignal_dlq" // There is nothing useful to fix: record decides greenness from Request.State, // not Build.Status, and writing a terminal status here would claim we saw an // outcome we never saw. -type BuildSignalController struct { +type buildSignalController struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory @@ -57,22 +57,23 @@ type BuildSignalController struct { consumerGroup string } -// Verify BuildSignalController implements consumer.Controller at compile time. -var _ consumer.Controller = (*BuildSignalController)(nil) +// Verify buildSignalController implements consumer.Controller at compile time. +var _ consumer.Controller = (*buildSignalController)(nil) -// NewBuildSignalController creates a DLQ controller for the buildsignal stage's +// NewDLQBuildSignalController creates a DLQ controller for the buildsignal stage's // dead-letter topic. topicKey is typically // dlq.TopicKey(stovepipemq.TopicKeyBuildSignal). -func NewBuildSignalController( +func NewDLQBuildSignalController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string, -) *BuildSignalController { - return &BuildSignalController{ - logger: logger.Named("buildsignal_dlq_controller"), - metricsScope: scope.SubScope("buildsignal_dlq_controller"), +) consumer.Controller { + name := string(topicKey) + "_controller" + return &buildSignalController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), stores: stores, topicKey: topicKey, consumerGroup: consumerGroup, @@ -83,7 +84,7 @@ func NewBuildSignalController( // to ack (success) or an error to nack (retry) — pair this controller only with a // consumer wired with errs.AlwaysRetryableProcessor so a transient reconcile // failure retries instead of dead-lettering the DLQ message itself. -func (c *BuildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { +func (c *buildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() sig := &stovepipemq.BuildSignal{} @@ -152,16 +153,16 @@ func (c *BuildSignalController) Process(ctx context.Context, delivery consumer.D } // Name returns the controller name for logging and metrics. -func (c *BuildSignalController) Name() string { - return "buildsignal_dlq" +func (c *buildSignalController) Name() string { + return string(c.topicKey) } // TopicKey returns the topic key this controller subscribes to. -func (c *BuildSignalController) TopicKey() consumer.TopicKey { +func (c *buildSignalController) TopicKey() consumer.TopicKey { return c.topicKey } // ConsumerGroup returns the consumer group for offset tracking. -func (c *BuildSignalController) ConsumerGroup() string { +func (c *buildSignalController) ConsumerGroup() string { return c.consumerGroup } diff --git a/stovepipe/controller/dlq/buildsignal_test.go b/stovepipe/controller/dlq/buildsignal_test.go index 1ae08af14..dbe88812f 100644 --- a/stovepipe/controller/dlq/buildsignal_test.go +++ b/stovepipe/controller/dlq/buildsignal_test.go @@ -38,7 +38,7 @@ type buildSignalDLQMocks struct { buildStore *storagemock.MockBuildStore } -func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (*BuildSignalController, buildSignalDLQMocks) { +func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (consumer.Controller, buildSignalDLQMocks) { t.Helper() m := buildSignalDLQMocks{ @@ -52,7 +52,7 @@ func newBuildSignalController(t *testing.T, ctrl *gomock.Controller) (*BuildSign store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() store.EXPECT().GetBuildStore().Return(m.buildStore).AnyTimes() - c := NewBuildSignalController( + c := NewDLQBuildSignalController( zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index a99712353..ed3962ea6 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -48,7 +48,7 @@ type staticStorageFactory struct{ store storage.Storage } // For returns the fixed store aggregate for any queue. func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil } -func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks) { +func newController(t *testing.T, ctrl *gomock.Controller) (consumer.Controller, dlqMocks) { t.Helper() m := dlqMocks{ @@ -60,7 +60,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() - c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + c := NewDLQRequestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") return c, m } diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go index a58fefb73..bdf8f0169 100644 --- a/stovepipe/controller/dlq/request.go +++ b/stovepipe/controller/dlq/request.go @@ -26,11 +26,11 @@ import ( "go.uber.org/zap" ) -// Controller is the DLQ reconciler for the process stage. It is registered against the +// requestController is the DLQ reconciler for the process stage. It is registered against the // process topic's DLQ (see TopicKey) and, on each delivery, decodes the request id from // the same ProcessRequest payload the primary process controller consumes, then drives // the referenced request to a terminal failed state via failRequest. -type Controller struct { +type requestController struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory @@ -38,24 +38,25 @@ type Controller struct { consumerGroup string } -// Verify Controller implements consumer.Controller at compile time. -var _ consumer.Controller = (*Controller)(nil) +// Verify requestController implements consumer.Controller at compile time. +var _ consumer.Controller = (*requestController)(nil) // _opName is the metric operation name shared by every emit in this file. const _opName = "process_dlq" -// NewController creates a new DLQ controller for the process stage's dead-letter topic. +// NewDLQRequestController creates a DLQ controller for the process stage's dead-letter topic. // topicKey is typically dlq.TopicKey(stovepipemq.TopicKeyProcess). -func NewController( +func NewDLQRequestController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string, -) *Controller { - return &Controller{ - logger: logger.Named("process_dlq_controller"), - metricsScope: scope.SubScope("process_dlq_controller"), +) consumer.Controller { + name := string(topicKey) + "_controller" + return &requestController{ + logger: logger.Named(name), + metricsScope: scope.SubScope(name), stores: stores, topicKey: topicKey, consumerGroup: consumerGroup, @@ -66,7 +67,7 @@ func NewController( // (success) or an error to nack (retry) — pair this controller only with a consumer // wired with errs.AlwaysRetryableProcessor so a transient reconcile failure retries // instead of dead-lettering the DLQ message itself. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { +func (c *requestController) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() pr := &stovepipemq.ProcessRequest{} @@ -113,16 +114,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // Name returns the controller name for logging and metrics. -func (c *Controller) Name() string { - return "process_dlq" +func (c *requestController) Name() string { + return string(c.topicKey) } // TopicKey returns the topic key this controller subscribes to. -func (c *Controller) TopicKey() consumer.TopicKey { +func (c *requestController) TopicKey() consumer.TopicKey { return c.topicKey } // ConsumerGroup returns the consumer group for offset tracking. -func (c *Controller) ConsumerGroup() string { +func (c *requestController) ConsumerGroup() string { return c.consumerGroup }