diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 10e32fe4..8dca48be 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -216,7 +216,15 @@ func run() error { ), ) - processController := process.NewController(logger.Sugar(), scope, store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process") + processController := process.NewController( + logger.Sugar(), + scope, + store, + queueconfigdefault.NewStore(), + fakeSourceControlFactory{}, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) if err := primaryConsumer.Register(processController); err != nil { return fmt.Errorf("failed to register process controller: %w", err) } diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index a856f647..b55513c1 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -12,6 +12,7 @@ go_library( "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -30,6 +31,8 @@ go_test( "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", + "//stovepipe/extension/sourcecontrol:go_default_library", + "//stovepipe/extension/sourcecontrol/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index dd80fdd1..1dce57c1 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -30,6 +30,7 @@ import ( stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/queueconfig" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -42,6 +43,7 @@ type Controller struct { metricsScope tally.Scope store storage.Storage queueConfigs queueconfig.Store + sourceControl sourcecontrol.Factory topicKey consumer.TopicKey consumerGroup string } @@ -58,6 +60,7 @@ func NewController( scope tally.Scope, store storage.Storage, queueConfigs queueconfig.Store, + sourceControl sourcecontrol.Factory, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -66,6 +69,7 @@ func NewController( metricsScope: scope.SubScope("process_controller"), store: store, queueConfigs: queueConfigs, + sourceControl: sourceControl, topicKey: topicKey, consumerGroup: consumerGroup, } @@ -174,6 +178,11 @@ func (c *Controller) coalesce(ctx context.Context, request entity.Request, lates // re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate // defers (acks) rather than failing. func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request, queueRow entity.Queue, maxConcurrent int32) error { + var sc sourcecontrol.SourceControl + var strategy entity.BuildStrategy + var baseURI string + var err error + for { if queueRow.InFlightCount >= maxConcurrent { // TODO: re-enqueue the request via PublishAfter on the process topic with GateWaitDelayMs. @@ -186,7 +195,22 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request return nil } - err := c.claimBuildSlot(ctx, &queueRow) + if queueRow.LastGreenURI != "" && sc == nil { + 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"), + ) + return fmt.Errorf("ProcessController failed to resolve source control for queue %s: %w", request.Queue, err) + } + } + + strategy, baseURI, err = c.deriveBuildStrategy(ctx, sc, queueRow, request) + if err != nil { + return err + } + + err = c.claimBuildSlot(ctx, &queueRow) if err == nil { break } @@ -201,11 +225,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request } } - // TODO(build-strategy): derive from queue last_green_uri + SourceControl.IsAncestor. - request.BuildStrategy = entity.BuildStrategyFull - request.BaseURI = "" - - transitioned, err := c.markProcessing(ctx, &request) + transitioned, err := c.markProcessing(ctx, &request, strategy, baseURI) if err != nil { // Slot claimed but never admitted: release best-effort so the slot isn't leaked // (a redelivery would find the gate closed by its own claim and nothing decrements it). @@ -220,15 +240,51 @@ func (c *Controller) admitLatestHead(ctx context.Context, request entity.Request // TODO(build-publish): publish BuildRequest to the build stage here. + metrics.NamedCounter(c.metricsScope, _opName, "admitted", 1, + metrics.NewTag("strategy", string(request.BuildStrategy)), + ) c.logger.Infow("admitted request to build", "request_id", request.ID, "queue", request.Queue, "uri", request.URI, "build_strategy", string(request.BuildStrategy), + "base_uri", request.BaseURI, ) return nil } +// deriveBuildStrategy chooses the validation scope and baseline from the queue's last-known-good commit. +// The caller resolves source control once and persists the returned values only after successfully claiming a build slot. +func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.SourceControl, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) { + if queueRow.LastGreenURI == "" { + return entity.BuildStrategyFull, "", nil + } + + isAncestor, err := sc.IsAncestor(ctx, queueRow.LastGreenURI, request.URI) + if err != nil { + if sourcecontrol.IsNotFound(err) { + metrics.NamedCounter(c.metricsScope, _opName, "strategy_fallbacks", 1, + metrics.NewTag("reason", "unknown_ancestry"), + ) + c.logger.Warnw("last-green URI is not in request history; using full build", + "queue", request.Queue, + "last_green_uri", queueRow.LastGreenURI, + "request_uri", request.URI, + ) + return entity.BuildStrategyFull, "", nil + } + metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1, + metrics.NewTag("stage", "ancestry"), + ) + return entity.BuildStrategyUnknown, "", fmt.Errorf("ProcessController failed to check ancestry for queue %s: %w", request.Queue, err) + } + + if isAncestor { + return entity.BuildStrategyIncrementalSinceGreen, queueRow.LastGreenURI, nil + } + return entity.BuildStrategyFull, "", nil +} + // claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it // reloads queueRow and returns ErrVersionMismatch so the caller can retry. func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) error { @@ -253,11 +309,12 @@ func (c *Controller) claimBuildSlot(ctx context.Context, queueRow *entity.Queue) return nil } -// markProcessing CAS-marks request accepted→processing, persisting BuildStrategy and BaseURI -// already set by the admit workflow. Retries on version conflicts. transitioned is true only -// when this call performed the CAS; false means a concurrent writer already advanced the -// request past accepted (a lost admit race), so the caller must release its claimed slot. -func (c *Controller) markProcessing(ctx context.Context, request *entity.Request) (transitioned bool, err error) { +// markProcessing CAS-marks request accepted→processing and persists the strategy chosen after the +// build slot was claimed. It reapplies the values after a request reload so an accepted concurrent +// update cannot discard them. transitioned is true only when this call performed the CAS; false +// means a concurrent writer already advanced the request past accepted, so the caller must release +// its claimed slot. +func (c *Controller) markProcessing(ctx context.Context, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) { reqStore := c.store.GetRequestStore() for { @@ -267,6 +324,8 @@ func (c *Controller) markProcessing(ctx context.Context, request *entity.Request updated := *request updated.State = entity.RequestStateProcessing + updated.BuildStrategy = strategy + updated.BaseURI = baseURI newVersion := request.Version + 1 if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 610223a0..13008712 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -29,6 +29,8 @@ import ( stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" + sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" @@ -43,23 +45,39 @@ const ( ) type processMocks struct { - reqStore *storagemock.MockRequestStore - queueStore *storagemock.MockQueueStore + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + sourceFactory *sourcecontrolmock.MockFactory + sourceControl *sourcecontrolmock.MockSourceControl } func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() + return newControllerWithScope(t, ctrl, tally.NewTestScope("test", nil)) +} +func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.Scope) (*Controller, processMocks) { + t.Helper() m := processMocks{ - reqStore: storagemock.NewMockRequestStore(ctrl), - queueStore: storagemock.NewMockQueueStore(ctrl), + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + sourceFactory: sourcecontrolmock.NewMockFactory(ctrl), + sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), } store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() - c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, queueconfigdefault.NewStore(), stovepipemq.TopicKeyProcess, "stovepipe-process") + c := NewController( + zap.NewNop().Sugar(), + scope, + store, + queueconfigdefault.NewStore(), + m.sourceFactory, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) return c, m } @@ -103,6 +121,240 @@ func expectAdmit(m processMocks, id string) { m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) } +func TestDeriveBuildStrategy(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" + + tests := []struct { + name string + queue entity.Queue + setup func(m processMocks) + wantStrategy entity.BuildStrategy + wantBaseURI string + wantErr bool + }{ + { + name: "cold start uses full build without source control", + queue: entity.Queue{Name: testQueue}, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "ancestor uses incremental build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) + }, + wantStrategy: entity.BuildStrategyIncrementalSinceGreen, + wantBaseURI: lastGreenURI, + }, + { + name: "history rewrite uses full build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, nil) + }, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "unknown ancestry uses full build", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, sourcecontrol.ErrNotFound) + }, + wantStrategy: entity.BuildStrategyFull, + }, + { + name: "ancestry error fails", + queue: entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + setup: func(m processMocks) { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, errors.New("source control unavailable")) + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } + + var sc sourcecontrol.SourceControl + if tt.queue.LastGreenURI != "" { + sc = m.sourceControl + } + strategy, baseURI, err := c.deriveBuildStrategy(context.Background(), sc, tt.queue, acceptedRequest(testID)) + + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantStrategy, strategy) + assert.Equal(t, tt.wantBaseURI, baseURI) + }) + } +} + +func TestDeriveBuildStrategyEmitsSourceControlMetrics(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" + + tests := []struct { + name string + ancestryErr error + metricName string + metricTags string + }{ + { + name: "unknown ancestry records fallback", + ancestryErr: sourcecontrol.ErrNotFound, + metricName: "strategy_fallbacks", + metricTags: "reason=unknown_ancestry", + }, + { + name: "source control failure records error", + ancestryErr: errors.New("source control unavailable"), + metricName: "source_control_errors", + metricTags: "stage=ancestry", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(false, tt.ancestryErr) + + _, _, err := c.deriveBuildStrategy( + context.Background(), + m.sourceControl, + entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, + acceptedRequest(testID), + ) + + if sourcecontrol.IsNotFound(tt.ancestryErr) { + require.NoError(t, err) + } else { + require.Error(t, err) + } + counter, ok := scope.Snapshot().Counters()["test.process_controller.process."+tt.metricName+"+"+tt.metricTags] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) + }) + } +} + +func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) { + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + expectAdmit(m, testID) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + + counter, ok := scope.Snapshot().Counters()["test.process_controller.process.admitted+strategy=full"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) +} + +func TestProcessEmitsSourceControlResolutionMetric(t *testing.T) { + const lastGreenURI = "git://repo/monorepo/main/green" + + ctrl := gomock.NewController(t) + scope := tally.NewTestScope("test", nil) + c, m := newControllerWithScope(t, ctrl, scope) + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: lastGreenURI, + Version: 1, + }, nil) + m.sourceFactory.EXPECT(). + For(sourcecontrol.Config{QueueName: testQueue}). + Return(nil, errors.New("source control unavailable")) + + require.Error(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + + counter, ok := scope.Snapshot().Counters()["test.process_controller.process.source_control_errors+stage=resolve"] + require.True(t, ok) + assert.Equal(t, int64(1), counter.Value()) +} + +func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { + const ( + initialLastGreen = "git://repo/monorepo/main/green-old" + reloadedLastGreen = "git://repo/monorepo/main/green-new" + ) + + tests := []struct { + name string + initialLastGreen string + }{ + { + name: "rederives against changed baseline", + initialLastGreen: initialLastGreen, + }, + { + name: "resolves source control after baseline appears", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + request := acceptedRequest(testID) + + initialQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: tt.initialLastGreen, + Version: 1, + } + reloadedQueue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: reloadedLastGreen, + Version: 2, + } + initialClaim := initialQueue + initialClaim.InFlightCount = 1 + claimedQueue := reloadedQueue + claimedQueue.InFlightCount = 1 + + updatedRequest := request + updatedRequest.State = entity.RequestStateProcessing + updatedRequest.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + updatedRequest.BaseURI = reloadedLastGreen + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(request, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(initialQueue, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + if tt.initialLastGreen != "" { + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), tt.initialLastGreen, testURI).Return(true, nil) + } + m.queueStore.EXPECT().Update(gomock.Any(), initialClaim, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(reloadedQueue, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), reloadedLastGreen, testURI).Return(true, nil) + m.queueStore.EXPECT().Update(gomock.Any(), claimedQueue, int32(2), int32(3)).Return(nil) + m.reqStore.EXPECT().Update(gomock.Any(), updatedRequest, int32(1), int32(2)).Return(nil) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) + }) + } +} + func TestProcess(t *testing.T) { tests := []struct { name string @@ -148,6 +400,22 @@ func TestProcess(t *testing.T) { expectAdmit(m, testID) }, }, + { + name: "source control failure does not claim a build slot", + wantErr: true, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + LastGreenURI: "git://repo/monorepo/main/green", + Version: 1, + }, nil) + m.sourceFactory.EXPECT(). + For(sourcecontrol.Config{QueueName: testQueue}). + Return(nil, errors.New("source control unavailable")) + }, + }, { name: "accepted with empty latest pointer awaits ingest stamp", setup: func(m processMocks) { @@ -166,6 +434,7 @@ func TestProcess(t *testing.T) { Name: testQueue, LatestRequestID: testID, InFlightCount: 1, + LastGreenURI: "git://repo/monorepo/main/green", Version: 1, }, nil) }, @@ -230,6 +499,38 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil) }, }, + { + name: "mark processing retry preserves derived strategy after accepted reload", + setup: func(m processMocks) { + const lastGreenURI = "git://repo/monorepo/main/green" + + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, LastGreenURI: lastGreenURI, Version: 1, + }, nil) + m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, LastGreenURI: lastGreenURI, Version: 1, + }, int32(1), int32(2)).Return(nil) + + firstAttempt := acceptedRequest(testID) + firstAttempt.State = entity.RequestStateProcessing + firstAttempt.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + firstAttempt.BaseURI = lastGreenURI + m.reqStore.EXPECT().Update(gomock.Any(), firstAttempt, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + + reloaded := acceptedRequest(testID) + reloaded.Version = 2 + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(reloaded, nil) + + retry := reloaded + retry.State = entity.RequestStateProcessing + retry.BuildStrategy = entity.BuildStrategyIncrementalSinceGreen + retry.BaseURI = lastGreenURI + m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil) + }, + }, { name: "mark processing lost race releases slot and skips admit", setup: func(m processMocks) {