diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 5860b070..cea33e08 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -38,7 +38,14 @@ go_library( "//submitqueue/extension/conflict/fake:go_default_library", "//submitqueue/extension/conflict/fileoverlap:go_default_library", "//submitqueue/extension/conflict/none:go_default_library", + "//submitqueue/extension/scorer:go_default_library", + "//submitqueue/extension/scorer/composite:go_default_library", + "//submitqueue/extension/scorer/fake:go_default_library", + "//submitqueue/extension/scorer/heuristic:go_default_library", + "//submitqueue/extension/speculation/allocator/sticky:go_default_library", + "//submitqueue/extension/speculation/generator/bestfirst:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", + "//submitqueue/extension/speculation/speculator/standard:go_default_library", "//submitqueue/extension/storage/mysql:go_default_library", "//submitqueue/extension/validator/fake:go_default_library", "//submitqueue/orchestrator:go_default_library", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 2d05b957..cb9ede13 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -42,13 +42,11 @@ import ( "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/platform/pipeline" "github.com/uber/submitqueue/submitqueue/core/changeset" - "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" cpfake "github.com/uber/submitqueue/submitqueue/extension/changeprovider/fake" githubprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/github" phabprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/phabricator" routingprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/routing" - "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake" "github.com/uber/submitqueue/submitqueue/orchestrator" @@ -198,12 +196,8 @@ func run() error { BuildRunner: profiles.BuildRunnerFactory(), ChangeProvider: profiles.ChangeProviderFactory(), Analyzer: profiles.AnalyzerFactory(), - // Speculation is wired but inert: the placeholder below proposes - // nothing, so no path is ever funded and no speculative build starts. - // The wiring change at the top of this stack replaces it with real - // per-queue speculators composed from each profile's scorer. - Speculator: noopSpeculators{}, - Validator: validatorfake.NewFactory(), + Speculator: profiles.SpeculatorFactory(), + Validator: validatorfake.NewFactory(), } // Assemble the pipeline: one call builds the topic registry, creates @@ -438,20 +432,3 @@ func parseTimeout(envVal string, defaultVal time.Duration) time.Duration { } return defaultVal } - -// noopSpeculators resolves every queue to a speculator that proposes nothing. -// It keeps the speculate stage inert — no path funded, no build started — -// until per-queue speculators are composed in the profiles. -type noopSpeculators struct{} - -// For returns the propose-nothing speculator for any queue. -func (noopSpeculators) For(speculator.Config) (speculator.Speculator, error) { - return noopSpeculator{}, nil -} - -type noopSpeculator struct{} - -// Speculate proposes no actions, whatever the queue looks like. -func (noopSpeculator) Speculate(context.Context, []entity.Batch, []entity.SpeculationPathSet) ([]entity.Speculation, error) { - return nil, nil -} diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index c1c0f82f..7c82a7f7 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -15,10 +15,12 @@ package main import ( + "context" "fmt" "github.com/uber-go/tally" "github.com/uber/submitqueue/submitqueue/core/changeset" + "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" buildfake "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" @@ -27,6 +29,14 @@ import ( conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake" "github.com/uber/submitqueue/submitqueue/extension/conflict/fileoverlap" "github.com/uber/submitqueue/submitqueue/extension/conflict/none" + "github.com/uber/submitqueue/submitqueue/extension/scorer" + "github.com/uber/submitqueue/submitqueue/extension/scorer/composite" + scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake" + "github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic" + "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky" + "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst" + "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" + specstandard "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard" "go.uber.org/zap" ) @@ -43,6 +53,15 @@ type Profile struct { // Analyzer detects conflicts between concurrent batches in this queue. Analyzer conflict.Analyzer + + // Scorer holds this queue's scoring profile. There is no scoring stage: the + // scorer feeds the queue's speculator, which ranks candidate paths by how + // likely their assumptions are to hold. + Scorer scorer.Scorer + + // Speculator decides which of this queue's speculation paths to build and + // which running ones to preempt, within the build budget. + Speculator speculator.Speculator } // Profiles maps a queue name to its extension Profile, falling back to a @@ -86,6 +105,14 @@ func (p Profiles) AnalyzerFactory() conflict.Factory { }) } +// SpeculatorFactory returns a speculator.Factory that resolves the Speculator +// for each queue from the profile registry. +func (p Profiles) SpeculatorFactory() speculator.Factory { + return speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) { + return p.For(c.QueueName).Speculator, nil + }) +} + // Thin func-type adapters — the http.HandlerFunc trick applied to each // extension Factory interface. Each func type satisfies the Factory contract, // letting Profiles cross the host/library boundary without dedicated structs. @@ -104,10 +131,14 @@ type analyzerFunc func(conflict.Config) (conflict.Analyzer, error) func (f analyzerFunc) For(c conflict.Config) (conflict.Analyzer, error) { return f(c) } +type speculatorFunc func(speculator.Config) (speculator.Speculator, error) + +func (f speculatorFunc) For(c speculator.Config) (speculator.Speculator, error) { return f(c) } + // newProfiles builds the per-queue extension profiles for the example. // Edge integrations (change provider) and the build runner form a shared // baseline; each per-queue profile starts from that baseline and overrides -// only the extensions that differ — here the conflict analyzer. +// only the extensions that differ — here the conflict analyzer and the scorer. // Queues without an explicit profile fall back to the baseline. func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (Profiles, error) { cp, err := newChangeProvider(logger, scope) @@ -115,15 +146,23 @@ func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resol return Profiles{}, fmt.Errorf("failed to create change provider: %w", err) } + // batchLines buckets a batch by total lines changed across all its changes — + // larger batches are likelier to fail to land. + batchLines := func(_ context.Context, changes entity.BatchChanges) (int, error) { + return changes.TotalLinesChanged(), nil + } + // Baseline profile: shared edge integrations + a fake build runner (every - // build succeeds unless a head URI carries a failure marker). The build - // runner instance is shared by the build and buildsignal controllers (same - // profile, same instance) so a build's recorded outcome survives across - // their separate factory lookups. + // build succeeds unless a head URI carries a failure marker), plus permissive + // defaults for scorer and conflict. The build runner instance is shared by + // the build and buildsignal controllers (same profile, same instance) so a + // build's recorded outcome survives across their separate factory lookups. // - // The analyzer is wrapped by conflictfake with a nil predicate - // (passthrough) — swap the predicate (e.g. conflictfake.FailAlways) on a - // queue to exercise the analyzer error path, as e2e-conflict-error-queue + // The scorer is wrapped by scorerfake so a change URI carrying + // "sq-fake=score-error" forces a scoring error end-to-end; it is a pure + // passthrough otherwise. The analyzer is wrapped by conflictfake with a nil + // predicate (passthrough) — swap the predicate (e.g. conflictfake.FailAlways) + // on a queue to exercise the analyzer error path, as e2e-conflict-error-queue // below does. base := Profile{ ChangeProvider: cp, @@ -131,10 +170,29 @@ func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resol // TODO: replace the delegate with a real analyzer (e.g. Tango target // analysis). "all" serializes the queue conservatively. Analyzer: conflictfake.New(all.New(), nil), + Scorer: scorerfake.New(resolver, heuristic.New( + resolver, + []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.5}}, + batchLines, scope.SubScope("scorer.default"), + )), } + // test-queue: bucketed heuristic scorer; conservative (serialized) conflicts + // inherited from the baseline. + testQueue := base + testQueue.Scorer = scorerfake.New(resolver, heuristic.New( + resolver, + []heuristic.Bucket{ + {Min: 0, Max: 1, Score: 0.95}, + {Min: 2, Max: 5, Score: 0.80}, + {Min: 6, Max: 20, Score: 0.60}, + {Min: 21, Max: 1<<31 - 1, Score: 0.40}, + }, + batchLines, scope.SubScope("scorer.test-queue"), + )) + // e2e-conflict-error-queue: every conflict analysis fails, exercising the - // analyzer error path. Edge integrations inherit the baseline. + // analyzer error path. Scorer/edge integrations inherit the baseline. conflictErrQueue := base conflictErrQueue.Analyzer = conflictfake.New(all.New(), conflictfake.FailAlways) @@ -143,16 +201,44 @@ func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resol fileOverlapQueue := base fileOverlapQueue.Analyzer = fileoverlap.New(resolver) - // e2e-test-queue: no conflicts (maximum parallelism). + // e2e-test-queue: composite scorer; no conflicts (maximum parallelism). e2eQueue := base e2eQueue.Analyzer = conflictfake.New(none.New(), nil) + e2eQueue.Scorer = scorerfake.New(resolver, composite.New( + map[string]scorer.Scorer{ + "size": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.8}}, batchLines, scope), + "flat": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.6}}, batchLines, scope), + }, + composite.Avg, scope.SubScope("scorer.e2e-test-queue"), + )) + // The speculator is composed last, because it is built from whatever scorer + // the profile ended up with. return Profiles{ - defaultProfile: base, + defaultProfile: withSpeculator(base), byQueue: map[string]Profile{ - "e2e-test-queue": e2eQueue, - "e2e-conflict-error-queue": conflictErrQueue, - "file-overlap-queue": fileOverlapQueue, + "test-queue": withSpeculator(testQueue), + "e2e-test-queue": withSpeculator(e2eQueue), + "e2e-conflict-error-queue": withSpeculator(conflictErrQueue), + "file-overlap-queue": withSpeculator(fileOverlapQueue), }, }, nil } + +// defaultBuildBudget caps how many builds a queue may have occupying CI at +// once. It is the only rationing lever the allocator has. +// +// TODO: move this onto entity.QueueConfig so operators can tune it per queue +// without a code change. QueueConfig carries only the queue name today. +const defaultBuildBudget = 4 + +// withSpeculator returns the profile with its speculator composed from its own +// scorer: bestfirst ranks a queue's candidate paths by how likely all their +// assumptions are to hold, and sticky spends the build budget down that ranking +// without preempting builds already running. Swapping either part changes the +// policy without touching the speculate controller, which depends only on the +// Speculator contract. +func withSpeculator(p Profile) Profile { + p.Speculator = specstandard.New(bestfirst.New(p.Scorer), sticky.New(defaultBuildBudget)) + return p +}