Skip to content

Commit 62c8e8a

Browse files
committed
feat(stovepipe): Wire build analysis pipeline
1 parent 8394d14 commit 62c8e8a

18 files changed

Lines changed: 603 additions & 18 deletions

File tree

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
# Build issue observation
2+
3+
Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success or recovery.
4+
5+
## Start and completion notices
6+
7+
After persisting a Build and its trigger occurrence, build publishes to buildsignal and buildanalysis. On completion, buildsignal persists status, releases the build slot, records the Request outcome, and publishes to record and buildanalysis. Each pair of publications attempts both destinations even if one fails. Publication is not atomic: one destination can receive its message while the other publication fails.
8+
9+
Start and completion notices use distinct deterministic message IDs and the same build-ID partition. Retries of one notice deduplicate without suppressing the other. A held start delivery can delay the completion notice in that partition, but it also reloads Build status on every pass and can perform terminal analysis itself. Both notices may analyze the terminal Build; duplicate observations are safe.
10+
11+
Publication errors are returned through the producing stage's retry/DLQ policy. Analysis processing does not gate record, but notification publication is still part of the producing stage: a failed start publication can send the build-stage message to its existing fail-closed recovery path.
12+
13+
## Observations and polling
14+
15+
The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build observation persistence, polling, and the stop condition for Requests that end without a recorded build result. It only accumulates historical observations; it does not maintain open issues or infer recovery from absence.
16+
17+
Status-polling DLQ recovery does not publish a build-completion notice. The analysis polling loop observes the ended Request through storage instead.
18+
19+
## Analysis errors and dead letters
20+
21+
Analysis errors are returned to the normal consumer whether the Build is running or terminal; errors do not use the two-minute polling hold. The reference service uses backend error classification and the extended retry configuration: a maximum of 10 attempts, with retry delays of 5, 10, 20, 40, then 60 seconds, capped at 60 seconds. A successful hold resets failure accounting. Non-retryable errors and exhausted retries send the message to the analysis DLQ, so final analysis is attempted rather than guaranteed to succeed.
22+
23+
The analysis DLQ uses the existing DLQ consumer, with its always-retryable processor and unlimited-retry subscription. The recovery controller records incomplete analysis in Request history without rerunning detection or changing the validation outcome. Only state reads and history persistence retry; unusable messages are logged and acknowledged. See [DLQ recovery](../../../../stovepipe/controller/buildanalysis/README.md#dlq-recovery) for the history identity and error-handling rules.

‎doc/rfc/stovepipe/workflow.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ The pipeline is a queue-driven chain of small, single-purpose controllers, in th
66

77
> **poll for a new head → ingest → process the build strategy → build → buildsignal → record whole-repo greenness (and any project results the resolver returns) → hook.**
88
9+
The [build issue observation](steps/buildanalysis.md) branch runs alongside this chain. It accumulates issues while builds run and attempts final analysis after build completion. Record does not wait for analysis to succeed. The reference host wires a no-op detector; deployments supply a real detector factory.
10+
911
There is no `analyze` controller and no `analyze` topic. A later project-analysis stage that would map a target graph onto another `build` pass is design only, described under [Designed, not built](#designed-not-built-project-analysis). `record` resolves project results inline on the same delivery that writes the repository fact.
1012

1113
## What Stovepipe is agnostic about

‎service/stovepipe/server/BUILD.bazel‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ go_library(
2626
"//service/stovepipe/server/handler:go_default_library",
2727
"//stovepipe/controller:go_default_library",
2828
"//stovepipe/controller/build:go_default_library",
29+
"//stovepipe/controller/buildanalysis:go_default_library",
2930
"//stovepipe/controller/buildsignal:go_default_library",
3031
"//stovepipe/controller/dlq:go_default_library",
3132
"//stovepipe/controller/process:go_default_library",
@@ -34,6 +35,8 @@ go_library(
3435
"//stovepipe/core/queuelog:go_default_library",
3536
"//stovepipe/core/queuestate:go_default_library",
3637
"//stovepipe/core/requestlog:go_default_library",
38+
"//stovepipe/extension/buildissue:go_default_library",
39+
"//stovepipe/extension/buildissue/noop:go_default_library",
3740
"//stovepipe/extension/buildrunner:go_default_library",
3841
"//stovepipe/extension/buildrunner/fake:go_default_library",
3942
"//stovepipe/extension/projectstatus/noop:go_default_library",
@@ -89,6 +92,7 @@ go_test(
8992
"//stovepipe/controller/dlq:go_default_library",
9093
"//stovepipe/core/messagequeue:go_default_library",
9194
"//stovepipe/core/requestlog:go_default_library",
95+
"//stovepipe/extension/buildissue/noop:go_default_library",
9296
"@com_github_stretchr_testify//assert:go_default_library",
9397
"@com_github_stretchr_testify//require:go_default_library",
9498
"@com_github_uber_go_tally//:go_default_library",

‎service/stovepipe/server/main.go‎

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ import (
4848
"github.com/uber/submitqueue/service/stovepipe/server/handler"
4949
"github.com/uber/submitqueue/stovepipe/controller"
5050
"github.com/uber/submitqueue/stovepipe/controller/build"
51+
"github.com/uber/submitqueue/stovepipe/controller/buildanalysis"
5152
"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
5253
"github.com/uber/submitqueue/stovepipe/controller/dlq"
5354
"github.com/uber/submitqueue/stovepipe/controller/process"
@@ -56,6 +57,8 @@ import (
5657
"github.com/uber/submitqueue/stovepipe/core/queuelog"
5758
"github.com/uber/submitqueue/stovepipe/core/queuestate"
5859
"github.com/uber/submitqueue/stovepipe/core/requestlog"
60+
"github.com/uber/submitqueue/stovepipe/extension/buildissue"
61+
buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
5962
"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
6063
buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake"
6164
projectstatusnoop "github.com/uber/submitqueue/stovepipe/extension/projectstatus/noop"
@@ -214,6 +217,7 @@ func run() error {
214217
subscriberName = fmt.Sprintf("stovepipe-%d", time.Now().Unix())
215218
}
216219

220+
detectorFactory := buildissuenoop.New()
217221
registry, err := newTopicRegistry(mysqlQueue, subscriberName)
218222
if err != nil {
219223
return fmt.Errorf("failed to create topic registry: %w", err)
@@ -254,7 +258,7 @@ func run() error {
254258
}
255259
}
256260
materializer := requestlog.NewMaterializer(scope)
257-
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{})
261+
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, materializer, registry, sourceControl, brf, hookResolver{}, detectorFactory)
258262
if err != nil {
259263
return err
260264
}
@@ -373,6 +377,7 @@ func registerPrimaryControllers(
373377
sourceControl sourcecontrol.Factory,
374378
brf buildrunner.Factory,
375379
hooks hookext.Hooks,
380+
detectorFactory buildissue.DetectorFactory,
376381
) (int, error) {
377382
var count int
378383

@@ -404,6 +409,12 @@ func registerPrimaryControllers(
404409
}
405410
count++
406411

412+
analysisController := buildanalysis.NewController(logger, scope, store, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
413+
if err := c.Register(analysisController); err != nil {
414+
return count, fmt.Errorf("failed to register buildanalysis controller: %w", err)
415+
}
416+
count++
417+
407418
recordController := record.NewController(logger, scope, store, materializer, projectstatusnoop.New(), sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
408419
if err := c.Register(recordController); err != nil {
409420
return count, fmt.Errorf("failed to register record controller: %w", err)
@@ -448,6 +459,12 @@ func registerDLQControllers(
448459
}
449460
count++
450461

462+
analysisDLQController := buildanalysis.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis), "stovepipe-buildanalysis-dlq")
463+
if err := c.Register(analysisDLQController); err != nil {
464+
return count, fmt.Errorf("failed to register buildanalysis dlq controller: %w", err)
465+
}
466+
count++
467+
451468
recordDLQController := record.NewDLQController(logger, scope, store, materializer, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
452469
if err := c.Register(recordDLQController); err != nil {
453470
return count, fmt.Errorf("failed to register record dlq controller: %w", err)
@@ -544,6 +561,18 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
544561
Queue: q,
545562
Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-hook-dlq"),
546563
},
564+
{
565+
Key: stovepipemq.TopicKeyBuildAnalysis,
566+
Name: "buildanalysis",
567+
Queue: q,
568+
Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-buildanalysis"),
569+
},
570+
{
571+
Key: dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis),
572+
Name: "buildanalysis_dlq",
573+
Queue: q,
574+
Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-buildanalysis-dlq"),
575+
},
547576
})
548577
}
549578

‎service/stovepipe/server/main_test.go‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"github.com/uber/submitqueue/stovepipe/controller/dlq"
2929
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
3030
"github.com/uber/submitqueue/stovepipe/core/requestlog"
31+
buildissuenoop "github.com/uber/submitqueue/stovepipe/extension/buildissue/noop"
3132
"go.uber.org/zap/zaptest"
3233
)
3334

@@ -60,7 +61,7 @@ func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Con
6061
deadLetter := &recordingConsumer{}
6162

6263
_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope), registry,
63-
fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{})
64+
fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{}, buildissuenoop.New())
6465
require.NoError(t, err)
6566

6667
_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, requestlog.NewMaterializer(tally.NoopScope))
@@ -129,6 +130,7 @@ func TestPipelineSubscriptionRetryPolicies(t *testing.T) {
129130
{name: "process", topicKey: stovepipemq.TopicKeyProcess, consumerGroup: "stovepipe-process", wantRetry: defaultRetry},
130131
{name: "build", topicKey: stovepipemq.TopicKeyBuild, consumerGroup: "stovepipe-build", wantRetry: defaultRetry},
131132
{name: "buildsignal", topicKey: stovepipemq.TopicKeyBuildSignal, consumerGroup: "stovepipe-buildsignal", wantRetry: extendedRetry},
133+
{name: "buildanalysis", topicKey: stovepipemq.TopicKeyBuildAnalysis, consumerGroup: "stovepipe-buildanalysis", wantRetry: extendedRetry},
132134
{name: "record", topicKey: stovepipemq.TopicKeyRecord, consumerGroup: "stovepipe-record", wantRetry: extendedRetry},
133135
{name: "hook", topicKey: basehook.TopicKeyHook, consumerGroup: "stovepipe-hook", wantRetry: defaultRetry},
134136
}
@@ -154,3 +156,18 @@ func TestDLQSubscriptionRetryPoliciesRemainUnlimited(t *testing.T) {
154156
})
155157
}
156158
}
159+
160+
func TestBuildAnalysisRegistration(t *testing.T) {
161+
registry, primary, deadLetter := registeredControllers(t)
162+
assert.Contains(t, topicKeys(primary), stovepipemq.TopicKeyBuildAnalysis)
163+
assert.Contains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis))
164+
165+
dlqName, ok := registry.TopicName(dlq.TopicKey(stovepipemq.TopicKeyBuildAnalysis))
166+
require.True(t, ok)
167+
assert.Equal(t, "buildanalysis_dlq", dlqName)
168+
169+
cfg, ok := registry.SubscriptionConfig(stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
170+
require.True(t, ok)
171+
assert.True(t, cfg.DLQ.Enabled)
172+
assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix)
173+
}

‎stovepipe/controller/build/BUILD.bazel‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,19 +24,28 @@ go_library(
2424

2525
go_test(
2626
name = "go_default_test",
27-
srcs = ["build_test.go"],
27+
srcs = [
28+
"build_analysis_test.go",
29+
"build_test.go",
30+
],
2831
embed = [":go_default_library"],
2932
deps = [
3033
"//platform/base/messagequeue:go_default_library",
3134
"//platform/consumer:go_default_library",
3235
"//platform/consumer/mock:go_default_library",
3336
"//platform/errs:go_default_library",
37+
"//platform/errs/mysql:go_default_library",
3438
"//platform/extension/messagequeue/mock:go_default_library",
3539
"//platform/metrics:go_default_library",
40+
"//platform/publish:go_default_library",
41+
"//stovepipe/controller/buildanalysis:go_default_library",
42+
"//stovepipe/controller/buildsignal:go_default_library",
3643
"//stovepipe/core/messagequeue:go_default_library",
3744
"//stovepipe/core/requestlog:go_default_library",
3845
"//stovepipe/core/requestlog/mock:go_default_library",
3946
"//stovepipe/entity:go_default_library",
47+
"//stovepipe/extension/buildissue:go_default_library",
48+
"//stovepipe/extension/buildissue/mock:go_default_library",
4049
"//stovepipe/extension/buildrunner:go_default_library",
4150
"//stovepipe/extension/buildrunner/mock:go_default_library",
4251
"//stovepipe/extension/storage:go_default_library",

‎stovepipe/controller/build/build.go‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
// Package build holds the build-stage queue controller. It consumes BuildRequest
1616
// messages (a request id), reloads the Request, triggers the build-runner for
1717
// the scope process already decided, persists a Build row, and publishes the
18-
// build id to buildsignal.
18+
// build id to buildsignal and buildanalysis.
1919
package build
2020

2121
import (
@@ -39,7 +39,7 @@ import (
3939

4040
// Controller consumes BuildRequest messages, reloads the referenced Request,
4141
// triggers a build for its already-decided scope, and publishes the resulting
42-
// build id to buildsignal. Implements consumer.Controller.
42+
// build id to status polling and analysis. Implements consumer.Controller.
4343
type Controller struct {
4444
logger *zap.SugaredLogger
4545
metricsScope tally.Scope
@@ -51,9 +51,6 @@ type Controller struct {
5151
consumerGroup string
5252
}
5353

54-
// Verify Controller implements consumer.Controller interface at compile time.
55-
var _ consumer.Controller = (*Controller)(nil)
56-
5754
// _opName is the metric operation name shared by every emit in this file.
5855
const _opName = "build"
5956

@@ -155,8 +152,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
155152
return err
156153
}
157154

158-
if err := c.publishBuildSignal(ctx, build.ID, request.Queue); err != nil {
159-
return fmt.Errorf("failed to publish build signal for %s: %w", build.ID, err)
155+
signalErr := c.publishBuildSignal(ctx, build.ID, request.Queue)
156+
if signalErr != nil {
157+
signalErr = fmt.Errorf("failed to publish build signal for %s: %w", build.ID, signalErr)
158+
}
159+
160+
analysisErr := c.publishBuildAnalysis(ctx, build.ID, request.Queue)
161+
if err := errs.Group(signalErr, analysisErr); err != nil {
162+
return err
160163
}
161164

162165
c.logger.Debugw("triggered build",
@@ -219,3 +222,23 @@ func (c *Controller) TopicKey() consumer.TopicKey {
219222
func (c *Controller) ConsumerGroup() string {
220223
return c.consumerGroup
221224
}
225+
226+
func (c *Controller) publishBuildAnalysis(ctx context.Context, buildID, queue string) error {
227+
payload, err := stovepipemq.Marshal(&stovepipemq.BuildAnalysis{Id: buildID, QueueName: queue})
228+
if err != nil {
229+
return fmt.Errorf("failed to serialize build analysis: %w", err)
230+
}
231+
232+
if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyBuildAnalysis, publish.MessageParams{
233+
Tenant: queue,
234+
ID: publish.IntentID(buildID, "started"),
235+
Payload: payload,
236+
PartitionKey: buildID,
237+
}); err != nil {
238+
return fmt.Errorf("failed to publish build analysis for %s: %w", buildID, err)
239+
}
240+
241+
return nil
242+
}
243+
244+
var _ consumer.Controller = (*Controller)(nil)

0 commit comments

Comments
 (0)