Skip to content

feat(orchestrator): poll builds and stop the ones nothing wants #504

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
behinddwalls wants to merge 1 commit into from
Open

feat(orchestrator): poll builds and stop the ones nothing wants #504

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 submitqueue/orchestrator/controller/buildsignal/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,9 @@ go_library(
importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal",
visibility = ["//visibility:public"],
deps = [
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/metrics:go_default_library",
"//submitqueue/core/publish:go_default_library",
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/buildrunner:go_default_library",
Expand All @@ -26,11 +26,11 @@ go_test(
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/consumer/mock:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/messagequeue/mock:go_default_library",
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/entity:go_default_library",
"//submitqueue/extension/buildrunner/mock:go_default_library",
"//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mock:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
"@com_github_stretchr_testify//require:go_default_library",
Expand Down
259 changes: 189 additions & 70 deletions submitqueue/orchestrator/controller/buildsignal/buildsignal.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,24 +12,42 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Package buildsignal implements the build poll loop. Each message carries
// a Build; the controller calls BuildRunner.Status, writes the latest
// status to the BuildStore, publishes the batch ID to TopicKeySpeculate
// so the state machine re-evaluates, and holds the delivery for the next
// poll when the build has not yet reached a terminal state. Each message
// partitions by batch ID, so slow polls on one batch's build do not block
// others. A webhook-capable backend can publish into this same topic — the
// controller cannot tell a poll-driven message from a push.
// Package buildsignal implements the build poll loop. Each message names one
// build by the runner's own ID; the controller calls BuildRunner.Status, writes
// the latest status to that build's record, and wakes the speculate run for its
// batch, holding its delivery between polls while the build is still in
// flight.
//
// The poll loop is also where builds are stopped. It follows every build to a
// terminal state anyway, so on each poll it checks whether anything still
// wants the build running — the batch not halted, the path's current attempt,
// with a status that is not a stop, linked to this very build — and asks the
// runner to cancel when nothing does. That makes cancellation level-triggered:
// the speculate run records intent in the path set and nothing more, no cancel
// message exists to go stale, and a check that misses one poll is remade on
// the next.
//
// The path set is read here as that kill list, and never written — the
// speculate run stays its only writer, which is what lets a run hold one
// version of a head's paths across its whole decision without a poll
// invalidating it. The read must come from the primary: a stale replica read
// could report a wanted build unwanted, and a cancel is irreversible.
//
// Each build partitions independently, so slow polls on one build do not block
// another, and successive polls of one build stay ordered. A webhook-capable
// backend can publish into this same topic — the controller cannot tell a
// poll-driven message from a push.
package buildsignal

import (
"context"
"errors"
"fmt"

"github.com/uber-go/tally"
entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
"github.com/uber/submitqueue/platform/consumer"
"github.com/uber/submitqueue/platform/metrics"
"github.com/uber/submitqueue/submitqueue/core/publish"
"github.com/uber/submitqueue/submitqueue/core/topickey"
"github.com/uber/submitqueue/submitqueue/entity"
"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
Expand All @@ -52,6 +70,9 @@ var (
PollDelayRunningMs int64 = 2000
)

// opName is the metric operation name shared by every emit in this file.
const opName = "process"

// Controller consumes build signal messages, polls BuildRunner.Status,
// persists the result, and drives the polling loop.
type Controller struct {
Expand Down Expand Up @@ -88,20 +109,30 @@ func NewController(
}
}

// Process polls the build's current status, persists it, publishes the
// batch ID to speculate so the state machine re-evaluates, and holds the
// delivery for the next poll when the build is still in flight.
// Process polls one attempt's build status, stops the build if nothing wants
// it running any more, persists the status, wakes the speculate run, and
// holds the delivery for the next poll while the build is still in flight.
// Returns nil to ack (success), or error to nack/reject.
//
// Error classification: deserialize, Status, Update, and the speculate
// publish stay non-retryable — they reject straight to DLQ on the first
// failure, where the operational republish path is the recovery mechanism.
// The poll loop's continuation is a hold, not a publish: the framework
// postpones the delivery, and a failed postpone write lapses into a normal
// visibility-timeout redelivery, so the loop cannot stall on an enqueue.
// There is deliberately no short-circuit for halted batches. A cancelling batch
// reaches its terminal state only once its paths stop, and this loop is the
// only thing watching them stop — and, now, the thing stopping them: speculate
// marks a path cancelling, the next poll here asks the runner to cancel, and a
// later poll observes CI actually stop and records it. Skipping the work —
// including the hold — for a halted batch would leave every cancelled
// batch stranded in Cancelling forever.
//
// Error classification: deserialize, Status, the kill-list reads, the
// persistence writes, and the speculate publish stay non-retryable — they
// reject straight to DLQ on the first failure, where the operational republish
// path is the recovery mechanism. The Cancel call is best-effort instead:
// failing the message for it would kill the poll chain that is the only thing
// that will retry the cancel, so a failure is logged and the next poll remakes
// the whole check. The poll loop's continuation is a hold, not a publish: the
// framework postpones the delivery, and a failed postpone write lapses into a
// normal visibility-timeout redelivery, so the loop cannot stall on an
// enqueue.
func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
const opName = "process"

msg := delivery.Message()

buildID, err := entity.BuildIDFromBytes(msg.Payload)
Expand All @@ -111,30 +142,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
return fmt.Errorf("failed to deserialize build ID: %w", err)
}

// Only the build ID travels on the queue; load the full Build from
// storage, which is the single source of truth for its BatchID and the
// snapshot the poll loop updates.
// Only the build ID travels on the queue; the record is the source of truth
// for which batch this build belongs to and what it last reported.
build, err := c.store.GetBuildStore().Get(ctx, buildID.ID)
if err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to get build %s: %w", buildID.ID, err)
}

c.logger.Debugw("polling build status",
"build_id", build.ID,
"batch_id", build.BatchID,
"attempt", delivery.Attempt(),
"partition_key", msg.PartitionKey,
)

// Load the batch first: it gives us the queue (needed to build the right
// BuildRunner) and lets us short-circuit halted batches before polling.
batch, err := c.store.GetBatchStore().Get(ctx, build.BatchID)
if err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to get batch %s: %w", build.BatchID, err)
}

c.logger.Debugw("polling build status",
"build_id", build.ID,
"batch_id", build.BatchID,
"delivery_attempt", delivery.Attempt(),
"partition_key", msg.PartitionKey,
)

buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue})
if err != nil {
metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1)
Expand All @@ -147,39 +175,56 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
return fmt.Errorf("failed to get status for build %s: %w", buildID.ID, err)
}

// Short-circuit if the batch is already halted (terminal OR cancelling).
// Speculate is already idempotent on terminal, but skipping the publish
// avoids noise. For Cancelling batches the cancel controller owns the
// terminal write and the downstream fan-out, so further pipeline work
// would race against it; silent ack is the only safe action.
if entity.IsBatchStateHalted(batch.State) {
metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1)
c.logger.Infow("skipping buildsignal publish for halted batch",
"batch_id", batch.ID,
"state", string(batch.State),
)
return nil
// Reconcile before recording: a build still running that nothing wants any
// more is asked to stop. Best-effort by design — the reschedule below is
// what guarantees the request is remade, so a failed Cancel must not fail
// the message and take that reschedule with it.
if !status.IsTerminal() {
stop, err := c.unwanted(ctx, batch, build)
if err != nil {
return err
}
if stop {
if err := buildRunner.Cancel(ctx, buildID); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "cancel_errors", 1)
c.logger.Warnw("failed to cancel an unwanted build; the next poll retries",
"build_id", build.ID,
"batch_id", build.BatchID,
"error", err,
)
} else {
metrics.NamedCounter(c.metricsScope, opName, "build_cancelled", 1)
c.logger.Infow("requested cancellation of a build nothing wants running",
"build_id", build.ID,
"batch_id", build.BatchID,
"path_id", build.PathID,
"attempt", build.Attempt,
)
}
}
}

updatedBuild := build
updatedBuild.Status = status

if err := c.store.GetBuildStore().Update(ctx, updatedBuild); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
if status != build.Status {
build.Status = status
if err := c.store.GetBuildStore().Update(ctx, build); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
}
}

// Re-evaluate the batch state machine with the latest build status.
if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, msg.PartitionKey); err != nil {
// Wake the speculate run so it re-plans the queue with this result. It
// reads the status from the record above rather than being told it, so a
// duplicated or reordered signal costs nothing.
if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
return fmt.Errorf("failed to publish to speculate: %w", err)
}

if status.IsTerminal() {
metrics.NamedCounter(c.metricsScope, opName, "terminal", 1, metrics.NewTag("status", string(status)))
c.logger.Infow("build reached terminal status",
"build_id", updatedBuild.ID,
"batch_id", updatedBuild.BatchID,
"build_id", build.ID,
"batch_id", build.BatchID,
"status", string(status),
)
return nil
Expand All @@ -192,7 +237,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
delivery.Hold(delayMs)

c.logger.Debugw("holding for next build status poll",
"build_id", updatedBuild.ID,
"build_id", build.ID,
"status", string(status),
"delay_ms", delayMs,
)
Expand All @@ -210,31 +255,105 @@ func pollDelay(status entity.BuildStatus) int64 {
}
}

// publishBatchID publishes a batch ID to the topic identified by key.
func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
bid := entity.BatchID{ID: batchID}
payload, err := bid.ToBytes()
// unwanted reports whether nothing wants this build running any more: its
// batch has halted, its path was called off or has moved to another attempt,
// or the attempt's link names a different build (this one lost a dispatch
// race). Every one of those conditions is permanent once true, so a stale read
// can only err toward keeping a build — never toward cancelling a wanted one.
//
// The two anomaly cases run the other way on purpose. A missing set or entry
// cannot legitimately happen — the dispatch read the entry out of the set to
// start this build, and entries are not removed — and a missing link cannot
// either, because the signal that led here is published after the link. Both
// therefore indicate store corruption, and since a cancel is irreversible, a
// corrupt kill list keeps the build rather than killing it; a halted batch is
// still caught by the first check, which needs none of those records.
func (c *Controller) unwanted(ctx context.Context, batch entity.Batch, build entity.Build) (bool, error) {
if entity.IsBatchStateHalted(batch.State) {
return true, nil
}

// A build without path coordinates predates per-path dispatch; the batch
// state above is the only kill list it has.
if build.PathID == "" {
return false, nil
}

set, err := c.store.GetSpeculationPathSetStore().Get(ctx, batch.ID)
if err != nil {
return fmt.Errorf("failed to serialize batch ID: %w", err)
if errors.Is(err, storage.ErrNotFound) {
metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1)
c.logger.Warnw("build exists but its head has no path set; keeping the build",
"build_id", build.ID,
"batch_id", batch.ID,
)
return false, nil
}
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return false, fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err)
}

entry, found := findEntry(set, build.PathID)
if !found {
metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1)
c.logger.Warnw("build exists but its path is gone from the set; keeping the build",
"build_id", build.ID,
"batch_id", batch.ID,
"path_id", build.PathID,
)
return false, nil
}

msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
switch entry.Status {
case entity.SpeculationPathStatusCancelling, entity.SpeculationPathStatusCancelled:
return true, nil
}

q, ok := c.registry.Queue(key)
if !ok {
return fmt.Errorf("no queue registered for topic key %s", key)
// The path has moved on to a newer attempt; this build belongs to a
// superseded one.
if entry.Attempt != build.Attempt {
return true, nil
}

topicName, ok := c.registry.TopicName(key)
if !ok {
return fmt.Errorf("no topic name registered for topic key %s", key)
link, err := c.store.GetPathBuildStore().Get(ctx, build.PathID, build.Attempt)
if err != nil {
if errors.Is(err, storage.ErrNotFound) {
metrics.NamedCounter(c.metricsScope, opName, "kill_list_anomalies", 1)
c.logger.Warnw("build exists but its attempt has no link; keeping the build",
"build_id", build.ID,
"path_id", build.PathID,
"attempt", build.Attempt,
)
return false, nil
}
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return false, fmt.Errorf("failed to look up build for path %s attempt %d: %w", build.PathID, build.Attempt, err)
}

if err := q.Publisher().Publish(ctx, topicName, msg); err != nil {
return fmt.Errorf("failed to publish message: %w", err)
// The attempt's build is a different one: this build lost the dispatch
// race, and nothing downstream will ever look at it.
return link.BuildID != build.ID, nil
}

// findEntry returns the set's entry for a path ID.
func findEntry(set entity.SpeculationPathSet, pathID string) (entity.SpeculationPathEntry, bool) {
for _, entry := range set.Paths {
if entry.ID == pathID {
return entry, true
}
}
return entity.SpeculationPathEntry{}, false
}

return nil
// publishBatchID publishes a batch ID to the topic identified by key, with a
// distinct message ID per publish (publish.UniqueID) so a later wake-up for the
// same batch is never deduplicated away.
func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
payload, err := entity.BatchID{ID: batchID}.ToBytes()
if err != nil {
return fmt.Errorf("failed to serialize batch ID: %w", err)
}
return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey)
}

// Name returns the controller name for logging and metrics.
Expand Down
Loading
Loading