Skip to content

refactor(stovepipe): align DLQ controller conventions #619

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mnoah1 wants to merge 1 commit into from
Open

refactor(stovepipe): align DLQ controller conventions #619

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions service/stovepipe/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand Down
31 changes: 16 additions & 15 deletions stovepipe/controller/dlq/buildsignal.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -49,30 +49,31 @@ 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
topicKey consumer.TopicKey
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,
Expand All @@ -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{}
Expand Down Expand Up @@ -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
}
4 changes: 2 additions & 2 deletions stovepipe/controller/dlq/buildsignal_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand All @@ -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},
Expand Down
4 changes: 2 additions & 2 deletions stovepipe/controller/dlq/dlq_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand All @@ -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
}

Expand Down
31 changes: 16 additions & 15 deletions stovepipe/controller/dlq/request.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,36 +26,37 @@ 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
topicKey consumer.TopicKey
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,
Expand All @@ -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{}
Expand Down Expand Up @@ -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
}
Loading