diff --git a/platform/publish/publish.go b/platform/publish/publish.go index 5ab6a90e2..4f280c315 100644 --- a/platform/publish/publish.go +++ b/platform/publish/publish.go @@ -47,6 +47,13 @@ import ( // which is what makes redelivery safe, while a new cause about the same entity // can never be swallowed by an older row. func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string) error { + return MessageWithMetadata(ctx, registry, key, msgID, payload, partitionKey, nil) +} + +// MessageWithMetadata is Message with side-band message metadata (headers/attributes) +// attached to the delivery. Use it to carry diagnostic context that is not part of +// the payload — the backend persists and redelivers metadata alongside the message. +func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string, metadata map[string]string) error { q, ok := registry.Queue(key) if !ok { return fmt.Errorf("no queue registered for topic key %s", key) @@ -56,7 +63,7 @@ func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer. return fmt.Errorf("no topic name registered for topic key %s", key) } - msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) + msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadata) return q.Publisher().Publish(ctx, topicName, msg) } diff --git a/submitqueue/client/view.go b/submitqueue/client/view.go index ee1b93843..949be748f 100644 --- a/submitqueue/client/view.go +++ b/submitqueue/client/view.go @@ -272,7 +272,11 @@ func summarize(rows []*Row) error { var failed []string for _, rw := range rows { if rw.Status != string(entity.RequestStatusLanded) { - failed = append(failed, fmt.Sprintf("%s=%s", rw.SQID, rw.Status)) + entry := fmt.Sprintf("%s=%s", rw.SQID, rw.Status) + if rw.Note != "" { + entry += ": " + rw.Note + } + failed = append(failed, entry) } } if len(failed) > 0 { diff --git a/submitqueue/client/view_test.go b/submitqueue/client/view_test.go index a58d195a9..9821a3308 100644 --- a/submitqueue/client/view_test.go +++ b/submitqueue/client/view_test.go @@ -504,6 +504,12 @@ func TestOutcome(t *testing.T) { func TestSummarize(t *testing.T) { assert.NoError(t, summarize([]*Row{{SQID: "q/1", Status: "landed"}})) assert.Error(t, summarize([]*Row{{SQID: "q/1", Status: "landed"}, {SQID: "q/2", Status: "error"}})) + + // A failure reason, when the request carries one, is part of the summary so a + // scripted run reports why rather than only that. + err := summarize([]*Row{{SQID: "q/2", Status: "error", Note: "merge failed: conflict in foo.go"}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "q/2=error: merge failed: conflict in foo.go") } // TestRowLineAlignment is the column contract: on every row the stage begins at diff --git a/submitqueue/core/topickey/topickey.go b/submitqueue/core/topickey/topickey.go index 1950ed262..fa31076fa 100644 --- a/submitqueue/core/topickey/topickey.go +++ b/submitqueue/core/topickey/topickey.go @@ -51,3 +51,9 @@ const ( // TopicKeyLog is the pipeline stage where per-request logs are written. TopicKeyLog TopicKey = "log" ) + +// MetadataKeyFailureReason is the conclude message's metadata attribute carrying +// a failed batch's human-readable reason. Set by the failure sites (merge and +// speculate) on the conclude publish and read by conclude to stamp the request's +// terminal log; absent on the landed and cancelled paths. +const MetadataKeyFailureReason = "failure_reason" diff --git a/submitqueue/orchestrator/controller/conclude/BUILD.bazel b/submitqueue/orchestrator/controller/conclude/BUILD.bazel index 4cff6b197..91992843d 100644 --- a/submitqueue/orchestrator/controller/conclude/BUILD.bazel +++ b/submitqueue/orchestrator/controller/conclude/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", "//submitqueue/core/request:go_default_library", + "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index 88e6bf2ce..8f7c28d38 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -22,6 +22,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" corerequest "github.com/uber/submitqueue/submitqueue/core/request" + "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" "go.uber.org/zap" @@ -123,8 +124,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // the batch but missing from the store is a hard error — the whole batch is // retried (and eventually dead-lettered) rather than silently skipped. We // translate the result into per-outcome logs and metrics. + // The failure reason rides the conclude message, not the batch: the failing + // stage stamps it here and it is empty on the landed and cancelled paths. + failureReason := msg.Metadata[topickey.MetadataKeyFailureReason] for _, requestID := range batch.Contains { - res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, "", map[string]string{ + res, err := corerequest.TerminateRequest(ctx, store, c.registry, requestID, requestState, failureReason, map[string]string{ "batch_id": batch.ID, }) if err != nil { diff --git a/submitqueue/orchestrator/controller/conclude/conclude_test.go b/submitqueue/orchestrator/controller/conclude/conclude_test.go index 72661566e..f39910da0 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude_test.go +++ b/submitqueue/orchestrator/controller/conclude/conclude_test.go @@ -450,6 +450,57 @@ func TestController_Process(t *testing.T) { } } +// TestController_Process_FailedBatchCarriesReasonToRequestLog is the propagation +// this change exists for: the failure reason carried on the conclude message +// reaches the request's terminal error log, instead of the empty message the +// request used to carry. +func TestController_Process_FailedBatchCarriesReasonToRequestLog(t *testing.T) { + ctrl := gomock.NewController(t) + + const reason = "merge failed: conflict in pkg/a/foo.go" + batch := entity.Batch{ + ID: "test-queue/batch/9", + Queue: "test-queue", + Contains: []string{"test-queue/9"}, + State: entity.BatchStateFailed, + Version: 2, + } + + batchStore := storagemock.NewMockBatchStore(ctrl) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + + request := entity.Request{ID: "test-queue/9", Queue: "test-queue", Version: 1, State: entity.RequestStateProcessing} + requestStore := storagemock.NewMockRequestStore(ctrl) + requestStore.EXPECT().Get(gomock.Any(), "test-queue/9").Return(request, nil) + requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil) + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes() + + controller, pub := newTestController(t, ctrl, store, false) + + var logged entity.RequestLog + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + log, err := entity.RequestLogFromBytes(msg.Payload) + require.NoError(t, err) + logged = log + return nil + }, + ) + + msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, map[string]string{topickey.MetadataKeyFailureReason: reason}) + delivery := consumermock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + + require.NoError(t, controller.Process(context.Background(), delivery)) + + assert.Equal(t, entity.RequestStatusError, logged.Status) + assert.Equal(t, reason, logged.LastError) +} + func TestController_Process_StorageFailure(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go index 795c1d994..7611dd837 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go @@ -122,6 +122,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } + // A merge failure's reason travels to conclude on the fan-out message, not on + // the batch, so it reaches the request's terminal log without becoming durable + // batch state. Empty on the merged path. Computed before the idempotency check + // so a redelivered failed batch re-fans-out with its reason intact. + var failureReason string + if result.Outcome != runwaypb.Outcome_SUCCEEDED { + failureReason = result.Reason + if failureReason == "" { + failureReason = "merge failed" + } + } + // Idempotency: a previous delivery already transitioned this batch to a // terminal state. Repair the membership record (a prior attempt may have // CAS'd without completing the record move), re-fan-out in case that @@ -132,7 +144,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er metrics.NamedCounter(c.metricsScope, opName, "state_update_errors", 1) return err } - return c.fanout(ctx, batch.ID, batch.Queue) + return c.fanout(ctx, batch.ID, batch.Queue, failureReason) } var newState entity.BatchState @@ -157,7 +169,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return err } - return c.fanout(ctx, batch.ID, batch.Queue) + return c.fanout(ctx, batch.ID, batch.Queue, failureReason) } // fanout publishes the batch ID to conclude (so requests are updated) and to @@ -170,12 +182,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // scoped the same way because speculate publishes there too when a batch goes // terminal on its own; the two mean the same thing but are decided at // different moments, so neither may swallow the other. -func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { - if err := c.publish(ctx, topickey.TopicKeyConclude, publish.IntentID(batchID, "conclude", "merged"), batchID, queue); err != nil { +func (c *Controller) fanout(ctx context.Context, batchID, queue, failureReason string) error { + var concludeMeta map[string]string + if failureReason != "" { + concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: failureReason} + } + if err := c.publish(ctx, topickey.TopicKeyConclude, publish.IntentID(batchID, "conclude", "merged"), batchID, queue, concludeMeta); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_conclude_errors", 1) return fmt.Errorf("failed to publish to conclude: %w", err) } - if err := c.publish(ctx, topickey.TopicKeySpeculate, publish.IntentID(batchID, "merged"), batchID, queue); err != nil { + if err := c.publish(ctx, topickey.TopicKeySpeculate, publish.IntentID(batchID, "merged"), batchID, queue, nil); err != nil { metrics.NamedCounter(c.metricsScope, "process", "publish_speculate_errors", 1) return fmt.Errorf("failed to publish to speculate: %w", err) } @@ -183,14 +199,15 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { } // publish publishes a batch ID to the given topic key under msgID, stamped -// with and partitioned by the batch's queue. -func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string) error { +// with and partitioned by the batch's queue. metadata rides the message as +// side-band headers (nil for none). +func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue string, metadata map[string]string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - if err := publish.Message(ctx, c.registry, key, msgID, payload, queue); err != nil { + if err := publish.MessageWithMetadata(ctx, c.registry, key, msgID, payload, queue, metadata); err != nil { return fmt.Errorf("failed to publish message: %w", err) } diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go index f4f4565fd..73b78c3e3 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go @@ -218,21 +218,41 @@ func TestProcess_NotMergedMarksBatchFailed(t *testing.T) { Version: 3, } batchStore.EXPECT().Get(gomock.Any(), testBatchID).Return(batch, nil) + // The failed transition records only the terminal state; the reason travels + // to conclude on the message, not on the batch. batchStore.EXPECT().Update(gomock.Any(), batchWithState(batch, entity.BatchStateFailed), int32(3), int32(4)).Return(nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes() store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() - var got []string - c := newController(t, store, recordingRegistry(t, ctrl, &got)) + byTopic := map[string]entityqueue.Message{} + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, topic string, msg entityqueue.Message) error { + byTopic[topic] = msg + return nil + }, + ).AnyTimes() + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: topickey.TopicKeyConclude, Name: "conclude", Queue: q}, + {Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: q}, + }) + require.NoError(t, err) res := runwaymq.MergeResult{Id: testBatchID, Outcome: runwaypb.Outcome_FAILED, Reason: "conflict in foo.go"} msg := entityqueue.NewMessage(testBatchID, resultPayload(t, res), testQueue, nil) // Not-merged is an expected terminal outcome, so Process acks (no error). - require.NoError(t, c.Process(context.Background(), newDelivery(ctrl, msg))) + require.NoError(t, newController(t, store, registry).Process(context.Background(), newDelivery(ctrl, msg))) - assert.ElementsMatch(t, []string{"conclude", "speculate"}, got) + require.Contains(t, byTopic, "conclude") + require.Contains(t, byTopic, "speculate") + // The merge reason rides the conclude message so conclude can stamp it on + // the request's terminal log; the speculate wake-up carries none. + assert.Equal(t, "conflict in foo.go", byTopic["conclude"].Metadata[topickey.MetadataKeyFailureReason]) + assert.Empty(t, byTopic["speculate"].Metadata[topickey.MetadataKeyFailureReason]) } func TestProcess_CancellingShortCircuit(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index a74f81f20..8096e0863 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -346,12 +346,19 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba } case outcomeFail, outcomeCancel: + // A failed batch carries its reason to conclude on the message, so the + // requests' terminal log records why speculation could not land it. + // Cancellation carries none — the cancel path owns that reason. + var concludeMeta map[string]string + if decision == outcomeFail { + concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: "no speculation path could pass; every candidate build failed"} + } // Named for the run that decided it, so a redelivery re-deriving the // same outcome does not conclude the batch twice, and so it stays // distinct from the conclude mergesignal sends for a merged batch. // A conclude that goes missing is recovered by fanout, which is // deliberately un-deduplicated. - if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, publish.IntentID(batch.ID, "conclude", "speculate"), batch.ID, batch.Queue, batch.Queue); err != nil { + if err := c.publishBatchIDWithMetadata(ctx, topickey.TopicKeyConclude, publish.IntentID(batch.ID, "conclude", "speculate"), batch.ID, batch.Queue, batch.Queue, concludeMeta); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return true, fmt.Errorf("failed to publish batch %s to conclude: %w", batch.ID, err) } diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index a2ad82983..306373640 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -784,6 +784,11 @@ func TestRun_FailedHeadConcludesAfterTheStateWrite(t *testing.T) { assert.Equal(t, []string{"conclude"}, h.published) assert.Empty(t, publishedBeforeWrite, "conclude rejects a non-terminal batch, so it must not be published before the write") + // The failure reason rides the conclude message so conclude can stamp it on + // the failed requests' terminal log, without persisting it as batch state. + require.Len(t, h.messages, 1) + assert.Equal(t, "no speculation path could pass; every candidate build failed", + h.messages[0].Metadata[topickey.MetadataKeyFailureReason]) } // A failure resolves a dependency, which can fail everything stacked on it. The diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 17f29f8c3..ec7c90425 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -224,11 +224,18 @@ func (c *Controller) fanout(ctx context.Context, batchID, queue string) error { // want the queue, so a queue's batches are processed in order, but the build // dispatch partitions by batch so heads dispatch in parallel. func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue, partitionKey string) error { + return c.publishBatchIDWithMetadata(ctx, key, msgID, batchID, queue, partitionKey, nil) +} + +// publishBatchIDWithMetadata is publishBatchID with side-band message metadata +// attached to the delivery (nil for none). Used to carry a failed batch's reason +// to conclude without persisting it as batch state. +func (c *Controller) publishBatchIDWithMetadata(ctx context.Context, key consumer.TopicKey, msgID, batchID, queue, partitionKey string, metadata map[string]string) error { payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - return publish.Message(ctx, c.registry, key, msgID, payload, partitionKey) + return publish.MessageWithMetadata(ctx, c.registry, key, msgID, payload, partitionKey, metadata) } // attributed records what a failure was about and counts it by subject type.