[3/N] feat(orchestrator): publish request stage progress

submitqueue/orchestrator/controller/build/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//platform/base/messagequeue:go_default_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/buildrunner:go_default_library",

submitqueue/orchestrator/controller/build/build.go

@@ -23,6 +23,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"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/buildrunner"
@@ -137,6 +138,15 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err)
 	}
 
+	if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusBuilding, map[string]string{
+		"batch_id":   batch.ID,
+		"build_id":   buildID.ID,
+		"controller": "build",
+	}); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1)
+		return fmt.Errorf("failed to publish building request logs for batch %s: %w", batch.ID, err)
+	}
+
 	build := entity.Build{
 		ID:      buildID.ID,
 		BatchID: batch.ID,

submitqueue/orchestrator/controller/build/build_test.go

@@ -102,7 +102,10 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock
 	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
 
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
 	)
 	require.NoError(t, err)
 
@@ -179,22 +182,31 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) {
 	// base is the dependency batches (identity); head is this batch.
 	br.EXPECT().Trigger(gomock.Any(), []entity.Batch{depBatch}, headBatch, gomock.Nil()).Return(entity.BuildID{ID: "build-xyz"}, nil)
 
-	var publishedTopic string
 	var published entity.BuildID
+	var buildingLogs []entity.RequestLog
 	mockPub := queuemock.NewMockPublisher(ctrl)
 	mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
-			publishedTopic = topic
-			bid, err := entity.BuildIDFromBytes(msg.Payload)
-			require.NoError(t, err)
-			published = bid
+			switch topic {
+			case "buildsignal":
+				bid, err := entity.BuildIDFromBytes(msg.Payload)
+				require.NoError(t, err)
+				published = bid
+			case "log":
+				logEntry, err := entity.RequestLogFromBytes(msg.Payload)
+				require.NoError(t, err)
+				buildingLogs = append(buildingLogs, logEntry)
+			}
 			return nil
 		},
-	)
+	).Times(3)
 	mockQ := queuemock.NewMockQueue(ctrl)
 	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
 	)
 	require.NoError(t, err)
 
@@ -207,9 +219,13 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) {
 
 	require.NoError(t, controller.Process(context.Background(), delivery))
 
-	// Only the build ID is published to buildsignal.
-	assert.Equal(t, "buildsignal", publishedTopic)
 	assert.Equal(t, "build-xyz", published.ID)
+	require.Len(t, buildingLogs, 2)
+	assert.Equal(t, entity.RequestStatusBuilding, buildingLogs[0].Status)
+	assert.Equal(t, headBatch.Contains[0], buildingLogs[0].RequestID)
+	assert.Equal(t, "build-xyz", buildingLogs[0].Metadata["build_id"])
+	assert.Equal(t, entity.RequestStatusBuilding, buildingLogs[1].Status)
+	assert.Equal(t, headBatch.Contains[1], buildingLogs[1].RequestID)
 
 	// The full Build is persisted to storage (the source of truth the poll
 	// loop reloads), and its ID matches what was published.
@@ -252,7 +268,10 @@ func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) {
 	mockQ := queuemock.NewMockQueue(ctrl)
 	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
 	)
 	require.NoError(t, err)
 	controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build")

submitqueue/orchestrator/controller/buildsignal/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//platform/base/messagequeue:go_default_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/buildrunner:go_default_library",

submitqueue/orchestrator/controller/buildsignal/buildsignal.go

@@ -30,6 +30,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"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/buildrunner"
@@ -173,6 +174,17 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
 	}
 
+	if status == entity.BuildStatusSucceeded {
+		if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusBuilt, map[string]string{
+			"batch_id":   batch.ID,
+			"build_id":   build.ID,
+			"controller": "buildsignal",
+		}); err != nil {
+			metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1)
+			return fmt.Errorf("failed to publish built request logs for batch %s: %w", batch.ID, err)
+		}
+	}
+
 	// Re-evaluate the batch state machine with the latest build status.
 	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, build.BatchID, msg.PartitionKey); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)

submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go

@@ -44,6 +44,7 @@ type testHarness struct {
 	batchStore   *storagemock.MockBatchStore
 	signalPub    *queuemock.MockPublisher
 	speculatePub *queuemock.MockPublisher
+	logPub       *queuemock.MockPublisher
 }
 
 func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness {
@@ -59,9 +60,14 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness {
 	speculateQ := queuemock.NewMockQueue(ctrl)
 	speculateQ.EXPECT().Publisher().Return(speculatePub).AnyTimes()
 
+	logPub := queuemock.NewMockPublisher(ctrl)
+	logQ := queuemock.NewMockQueue(ctrl)
+	logQ.EXPECT().Publisher().Return(logPub).AnyTimes()
+
 	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{Key: topickey.TopicKeyBuildSignal, Name: "buildsignal", Queue: signalQ},
 		{Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: speculateQ},
+		{Key: topickey.TopicKeyLog, Name: "log", Queue: logQ},
 	})
 	require.NoError(t, err)
 
@@ -87,6 +93,7 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness {
 		batchStore:   batchStore,
 		signalPub:    signalPub,
 		speculatePub: speculatePub,
+		logPub:       logPub,
 	}
 }
 
@@ -155,6 +162,38 @@ func TestController_Process_Terminal(t *testing.T) {
 	}
 }
 
+func TestController_Process_SucceededPublishesBuiltLog(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	h := newTestHarness(t, ctrl)
+
+	build := entity.Build{ID: "b-built", BatchID: "batch-built", Status: entity.BuildStatusRunning}
+	batch := entity.Batch{
+		ID:       build.BatchID,
+		Queue:    "test-queue",
+		Contains: []string{"test-queue/1"},
+		State:    entity.BatchStateSpeculating,
+	}
+
+	h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
+	h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(batch, nil)
+	h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusSucceeded, nil, nil)
+	h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, entity.BuildStatusSucceeded).Return(nil)
+	h.logPub.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ string, msg entityqueue.Message) error {
+			logEntry, err := entity.RequestLogFromBytes(msg.Payload)
+			require.NoError(t, err)
+			assert.Equal(t, entity.RequestStatusBuilt, logEntry.Status)
+			assert.Equal(t, batch.Contains[0], logEntry.RequestID)
+			assert.Equal(t, batch.ID, logEntry.Metadata["batch_id"])
+			assert.Equal(t, build.ID, logEntry.Metadata["build_id"])
+			return nil
+		},
+	)
+	h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil)
+
+	require.NoError(t, h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)))
+}
+
 // TestController_Process_NonTerminal verifies a non-terminal poll persists
 // the status, publishes to speculate, AND re-publishes to buildsignal via
 // PublishAfter with the per-status delay.

submitqueue/orchestrator/controller/merge/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//submitqueue/core/request:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",

submitqueue/orchestrator/controller/merge/merge.go

@@ -35,6 +35,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"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/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 )
@@ -142,6 +143,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return fmt.Errorf("failed to build merge request for batch %s: %w", batch.ID, err)
 	}
 
+	if err := corerequest.PublishBatchLogs(ctx, c.registry, batch.Contains, entity.RequestStatusLanding, map[string]string{
+		"batch_id":   batch.ID,
+		"controller": "merge",
+	}); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1)
+		return fmt.Errorf("failed to publish landing request logs for batch %s: %w", batch.ID, err)
+	}
+
 	if err := c.publish(ctx, c.runwayTopicKey, req, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to runway merge: %w", err)

submitqueue/orchestrator/controller/merge/merge_test.go

@@ -117,30 +117,45 @@ func TestProcess_PublishesFullPayloadToRunway(t *testing.T) {
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
-	var gotTopic string
-	var gotPayload []byte
+	var landingLogs []entity.RequestLog
+	var mergePayload []byte
 	pub := queuemock.NewMockPublisher(ctrl)
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
-			gotTopic = topic
-			gotPayload = msg.Payload
+			switch topic {
+			case "log":
+				logEntry, err := entity.RequestLogFromBytes(msg.Payload)
+				require.NoError(t, err)
+				landingLogs = append(landingLogs, logEntry)
+			case "runway-merge":
+				mergePayload = msg.Payload
+			}
 			return nil
 		},
-	)
+	).Times(3)
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
+		[]consumer.TopicConfig{
+			{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: q},
+		},
 	)
 	require.NoError(t, err)
 
 	c := newController(t, store, registry)
 	require.NoError(t, c.Process(context.Background(), newDelivery(t, ctrl, batchID, batch.Queue)))
 
+	require.Len(t, landingLogs, 2)
+	assert.Equal(t, entity.RequestStatusLanding, landingLogs[0].Status)
+	assert.Equal(t, req1.ID, landingLogs[0].RequestID)
+	assert.Equal(t, batch.ID, landingLogs[0].Metadata["batch_id"])
+	assert.Equal(t, entity.RequestStatusLanding, landingLogs[1].Status)
+	assert.Equal(t, req2.ID, landingLogs[1].RequestID)
+
 	// Full payload published to runway, keyed by the batch id (the correlation id).
-	assert.Equal(t, "runway-merge", gotTopic)
 	got := &runwaymq.MergeRequest{}
-	require.NoError(t, runwaymq.Unmarshal(gotPayload, got))
+	require.NoError(t, runwaymq.Unmarshal(mergePayload, got))
 	assert.Equal(t, batch.ID, got.Id)
 	assert.Equal(t, batch.Queue, got.QueueName)
 	require.Len(t, got.Steps, 2)
@@ -206,12 +221,20 @@ func TestProcess_PublishFailureReturnsError(t *testing.T) {
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
-	pub := queuemock.NewMockPublisher(ctrl)
-	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("enqueue failed"))
-	q := queuemock.NewMockQueue(ctrl)
-	q.EXPECT().Publisher().Return(pub).AnyTimes()
+	logPub := queuemock.NewMockPublisher(ctrl)
+	logPub.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).Return(nil)
+	logQ := queuemock.NewMockQueue(ctrl)
+	logQ.EXPECT().Publisher().Return(logPub).AnyTimes()
+
+	mergePub := queuemock.NewMockPublisher(ctrl)
+	mergePub.EXPECT().Publish(gomock.Any(), "runway-merge", gomock.Any()).Return(fmt.Errorf("enqueue failed"))
+	mergeQ := queuemock.NewMockQueue(ctrl)
+	mergeQ.EXPECT().Publisher().Return(mergePub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: q}},
+		[]consumer.TopicConfig{
+			{Key: runwaymq.TopicKeyMerge, Name: "runway-merge", Queue: mergeQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: logQ},
+		},
 	)
 	require.NoError(t, err)
 

submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go

@@ -130,6 +130,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		return nil
 	}
 
+	logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusValidated, request.Version, "", map[string]string{
+		"controller": "mergeconflictsignal",
+	})
+	if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1)
+		return fmt.Errorf("failed to publish validated request log for %s: %w", request.ID, err)
+	}
+
 	if err := c.publishRequestID(ctx, topickey.TopicKeyBatch, request.ID, request.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to batch: %w", err)

submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go

@@ -61,20 +61,21 @@ func TestProcess_MergeablePublishesToBatch(t *testing.T) {
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes()
 
-	var gotTopic string
-	var gotPayload []byte
+	published := make(map[string][]byte)
 	pub := queuemock.NewMockPublisher(ctrl)
 	pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
 		func(_ context.Context, topic string, msg entityqueue.Message) error {
-			gotTopic = topic
-			gotPayload = msg.Payload
+			published[topic] = msg.Payload
 			return nil
 		},
-	)
+	).Times(2)
 	q := queuemock.NewMockQueue(ctrl)
 	q.EXPECT().Publisher().Return(pub).AnyTimes()
 	registry, err := consumer.NewTopicRegistry(
-		[]consumer.TopicConfig{{Key: topickey.TopicKeyBatch, Name: "batch", Queue: q}},
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeyBatch, Name: "batch", Queue: q},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: q},
+		},
 	)
 	require.NoError(t, err)
 
@@ -85,8 +86,12 @@ func TestProcess_MergeablePublishesToBatch(t *testing.T) {
 	msg := entityqueue.NewMessage(testRequestID, resultPayload(t, res), testQueue, nil)
 	require.NoError(t, controller.Process(context.Background(), newDelivery(ctrl, msg)))
 
-	assert.Equal(t, "batch", gotTopic)
-	rid, err := entity.RequestIDFromBytes(gotPayload)
+	logEntry, err := entity.RequestLogFromBytes(published["log"])
+	require.NoError(t, err)
+	assert.Equal(t, entity.RequestStatusValidated, logEntry.Status)
+	assert.Equal(t, "mergeconflictsignal", logEntry.Metadata["controller"])
+
+	rid, err := entity.RequestIDFromBytes(published["batch"])
 	require.NoError(t, err)
 	assert.Equal(t, testRequestID, rid.ID)
 }

submitqueue/orchestrator/controller/speculate/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//platform/base/messagequeue:go_default_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",