From 3fa7709b1782d169f711a91fa4860b7a264f0500 Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Wed, 15 Jul 2026 23:38:27 -0700 Subject: [PATCH] feat(sqsim): run scenarios against local stack --- sqsim/runner/BUILD.bazel | 39 ++ sqsim/runner/grpc.go | 126 ++++++ sqsim/runner/grpc_test.go | 34 ++ sqsim/runner/local.go | 335 ++++++++++++++ sqsim/runner/local_test.go | 41 ++ sqsim/runner/runner.go | 413 ++++++++++++++++++ sqsim/runner/runner_test.go | 162 +++++++ sqsim/runner/text.go | 53 +++ sqsim/runner/text_test.go | 37 ++ .../controller/buildsignal/buildsignal.go | 11 +- .../buildsignal/buildsignal_test.go | 13 +- tool/sqsim/BUILD.bazel | 10 +- tool/sqsim/main.go | 56 +++ tool/sqsim/main_test.go | 28 ++ 14 files changed, 1346 insertions(+), 12 deletions(-) create mode 100644 sqsim/runner/BUILD.bazel create mode 100644 sqsim/runner/grpc.go create mode 100644 sqsim/runner/grpc_test.go create mode 100644 sqsim/runner/local.go create mode 100644 sqsim/runner/local_test.go create mode 100644 sqsim/runner/runner.go create mode 100644 sqsim/runner/runner_test.go create mode 100644 sqsim/runner/text.go create mode 100644 sqsim/runner/text_test.go diff --git a/sqsim/runner/BUILD.bazel b/sqsim/runner/BUILD.bazel new file mode 100644 index 00000000..26242ad0 --- /dev/null +++ b/sqsim/runner/BUILD.bazel @@ -0,0 +1,39 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "grpc.go", + "local.go", + "runner.go", + "text.go", + ], + importpath = "github.com/uber/submitqueue/sqsim/runner", + visibility = ["//visibility:public"], + deps = [ + "//api/base/change/protopb:go_default_library", + "//api/base/mergestrategy/protopb:go_default_library", + "//api/submitqueue/gateway/protopb:go_default_library", + "//sqsim:go_default_library", + "//sqsim/model:go_default_library", + "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_grpc//credentials/insecure:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "grpc_test.go", + "local_test.go", + "runner_test.go", + "text_test.go", + ], + embed = [":go_default_library"], + deps = [ + "//api/submitqueue/gateway/protopb:go_default_library", + "//sqsim:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/sqsim/runner/grpc.go b/sqsim/runner/grpc.go new file mode 100644 index 00000000..6b5e2523 --- /dev/null +++ b/sqsim/runner/grpc.go @@ -0,0 +1,126 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "context" + "fmt" + + changepb "github.com/uber/submitqueue/api/base/change/protopb" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// GRPCGateway calls the public Gateway gRPC API. +type GRPCGateway struct { + conn *grpc.ClientConn + client gatewaypb.SubmitQueueGatewayClient +} + +// NewGRPCGateway connects to a Gateway address. +func NewGRPCGateway(address string) (*GRPCGateway, error) { + if address == "" { + return nil, fmt.Errorf("gateway address is required") + } + conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, fmt.Errorf("connect to gateway: %w", err) + } + return &GRPCGateway{conn: conn, client: gatewaypb.NewSubmitQueueGatewayClient(conn)}, nil +} + +// Close closes the Gateway connection. +func (g *GRPCGateway) Close() error { + return g.conn.Close() +} + +// Ping verifies that the Gateway is serving requests. +func (g *GRPCGateway) Ping(ctx context.Context) error { + _, err := g.client.Ping(ctx, &gatewaypb.PingRequest{Message: "sqsim"}) + return err +} + +// Land submits one synthetic change. +func (g *GRPCGateway) Land(ctx context.Context, queue, changeURI string) (string, error) { + response, err := g.client.Land(ctx, &gatewaypb.LandRequest{ + Queue: queue, + Change: &changepb.Change{Uris: []string{changeURI}}, + Strategy: mergestrategypb.Strategy_REBASE, + }) + if err != nil { + return "", err + } + return response.GetSqid(), nil +} + +// List returns one page of queue request summaries. +func (g *GRPCGateway) List(ctx context.Context, queue string, receivedAtOrAfterMs, receivedBeforeMs int64, pageToken string) ([]Summary, string, error) { + response, err := g.client.List(ctx, &gatewaypb.ListRequest{ + Queue: queue, + ReceivedAtOrAfterMs: receivedAtOrAfterMs, + ReceivedBeforeMs: receivedBeforeMs, + PageSize: 100, + PageToken: pageToken, + }) + if err != nil { + return nil, "", err + } + summaries := make([]Summary, len(response.GetRequests())) + for i, summary := range response.GetRequests() { + summaries[i] = fromProtoSummary(summary) + } + return summaries, response.GetNextPageToken(), nil +} + +// Summary returns the current public request summary. +func (g *GRPCGateway) Summary(ctx context.Context, sqid string) (Summary, error) { + response, err := g.client.GetRequestSummaryByID(ctx, &gatewaypb.GetRequestSummaryByIDRequest{Sqid: sqid}) + if err != nil { + return Summary{}, err + } + if response.GetRequest() == nil { + return Summary{}, fmt.Errorf("request summary %q is empty", sqid) + } + return fromProtoSummary(response.GetRequest()), nil +} + +// History returns retained public lifecycle events. +func (g *GRPCGateway) History(ctx context.Context, sqid string) ([]HistoryEvent, error) { + response, err := g.client.GetRequestHistoryByID(ctx, &gatewaypb.GetRequestHistoryByIDRequest{Sqid: sqid}) + if err != nil { + return nil, err + } + events := make([]HistoryEvent, len(response.GetEvents())) + for i, event := range response.GetEvents() { + events[i] = HistoryEvent{ + TimestampMs: event.GetTimestampMs(), + Status: event.GetStatus(), + LastError: event.GetLastError(), + Metadata: cloneMetadata(event.GetMetadata()), + } + } + return events, nil +} + +func fromProtoSummary(summary *gatewaypb.RequestSummary) Summary { + return Summary{ + SQID: summary.GetSqid(), + Status: summary.GetStatus(), + LastError: summary.GetLastError(), + Metadata: cloneMetadata(summary.GetMetadata()), + } +} diff --git a/sqsim/runner/grpc_test.go b/sqsim/runner/grpc_test.go new file mode 100644 index 00000000..98c664dd --- /dev/null +++ b/sqsim/runner/grpc_test.go @@ -0,0 +1,34 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "testing" + + "github.com/stretchr/testify/assert" + gatewaypb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" +) + +func TestFromProtoSummaryCopiesPublicFields(t *testing.T) { + got := fromProtoSummary(&gatewaypb.RequestSummary{ + Sqid: "sqsim/1", + Status: "building", + LastError: "none", + Metadata: map[string]string{"controller": "build"}, + }) + assert.Equal(t, "sqsim/1", got.SQID) + assert.Equal(t, "building", got.Status) + assert.Equal(t, "build", got.Metadata["controller"]) +} diff --git a/sqsim/runner/local.go b/sqsim/runner/local.go new file mode 100644 index 00000000..fa6ced84 --- /dev/null +++ b/sqsim/runner/local.go @@ -0,0 +1,335 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "bytes" + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + + "github.com/uber/submitqueue/sqsim" + "github.com/uber/submitqueue/sqsim/model" +) + +// LocalOptions configure a fresh local Compose run. +type LocalOptions struct { + // ScenarioName is the selected public scenario name. + ScenarioName string + // Scenario is the immutable workload. + Scenario sqsim.Scenario + // Observer receives visible state changes. + Observer Observer + // Stdout receives build and Compose progress. + Stdout io.Writer + // Stderr receives service logs on failure. + Stderr io.Writer + // RepoRoot overrides repository discovery. + RepoRoot string + // PollInterval overrides the public API poll interval. + PollInterval time.Duration +} + +// RunLocal executes a scenario against one fresh local Compose stack. +func RunLocal(ctx context.Context, options LocalOptions) (report Report, retErr error) { + if options.Stdout == nil { + options.Stdout = io.Discard + } + if options.Stderr == nil { + options.Stderr = io.Discard + } + if options.PollInterval <= 0 { + options.PollInterval = 250 * time.Millisecond + } + repoRoot := options.RepoRoot + if repoRoot == "" { + var err error + repoRoot, err = findRepoRoot(ctx) + if err != nil { + return Report{}, err + } + } + profile, err := model.Compile(options.ScenarioName, options.Scenario) + if err != nil { + return Report{}, err + } + profileDir, err := os.MkdirTemp("", "sqsim-profile-*") + if err != nil { + return Report{}, fmt.Errorf("create profile directory: %w", err) + } + defer os.RemoveAll(profileDir) + if err := model.Write(filepath.Join(profileDir, "profile.json"), profile); err != nil { + return Report{}, err + } + + stack := newLocalStack(repoRoot, profileDir, options.Stdout, options.Stderr) + if err := stack.Start(ctx); err != nil { + stack.Logs(ctx) + stack.Close(context.Background()) + return Report{}, err + } + defer func() { + if retErr != nil || !report.Passed { + stack.Logs(context.Background()) + } + if err := stack.Close(context.Background()); err != nil && retErr == nil { + retErr = err + } + }() + + gateway, err := waitForGateway(ctx, stack.GatewayAddress(ctx), options.Scenario.TimeoutMs) + if err != nil { + return Report{}, err + } + defer gateway.Close() + + return Run(ctx, Options{ + ScenarioName: options.ScenarioName, + Scenario: options.Scenario, + Gateway: gateway, + Clock: model.RealClock{}, + PollInterval: options.PollInterval, + Observer: options.Observer, + }) +} + +type localStack struct { + repoRoot string + project string + compose string + stdout io.Writer + stderr io.Writer + exec commandExecutor + baseEnv []string + started bool +} + +type commandExecutor interface { + Run(ctx context.Context, directory string, env []string, stdin io.Reader, stdout, stderr io.Writer, name string, args ...string) error + Output(ctx context.Context, directory string, env []string, name string, args ...string) ([]byte, error) +} + +type osExecutor struct{} + +func (osExecutor) Run(ctx context.Context, directory string, env []string, stdin io.Reader, stdout, stderr io.Writer, name string, args ...string) error { + command := exec.CommandContext(ctx, name, args...) + command.Dir = directory + command.Env = env + command.Stdin = stdin + command.Stdout = stdout + command.Stderr = stderr + return command.Run() +} + +func (osExecutor) Output(ctx context.Context, directory string, env []string, name string, args ...string) ([]byte, error) { + command := exec.CommandContext(ctx, name, args...) + command.Dir = directory + command.Env = env + return command.Output() +} + +func newLocalStack(repoRoot, profileDir string, stdout, stderr io.Writer) *localStack { + project := fmt.Sprintf("sqsim-%x-%x", os.Getpid(), time.Now().UnixNano()) + return &localStack{ + repoRoot: repoRoot, + project: project, + compose: filepath.Join(repoRoot, "service/submitqueue/docker-compose.yml"), + stdout: stdout, + stderr: stderr, + exec: osExecutor{}, + baseEnv: append(os.Environ(), + "REPO_ROOT="+repoRoot, + "SQSIM_PROFILE_DIR="+profileDir, + "SQSIM_SCENARIO_PATH=/sqsim/profile.json", + ), + } +} + +func (s *localStack) Start(ctx context.Context) error { + fmt.Fprintf(s.stdout, "Building local SubmitQueue binaries...\n") + if err := s.exec.Run(ctx, s.repoRoot, s.baseEnv, nil, s.stdout, s.stderr, "make", + "build-submitqueue-gateway-linux", + "build-submitqueue-orchestrator-linux", + "build-runway-linux", + ); err != nil { + return fmt.Errorf("build local binaries: %w", err) + } + + fmt.Fprintf(s.stdout, "Starting fresh stack %s...\n", s.project) + if err := s.composeRun(ctx, nil, s.stdout, s.stderr, "up", "-d", "--wait", "mysql-app", "mysql-queue"); err != nil { + return fmt.Errorf("start databases: %w", err) + } + s.started = true + if err := s.applySchemas(ctx); err != nil { + return err + } + if err := s.composeRun(ctx, nil, s.stdout, s.stderr, "up", "-d", "--build", "gateway-service", "orchestrator-service", "runway-service"); err != nil { + return fmt.Errorf("start services: %w", err) + } + return nil +} + +func (s *localStack) applySchemas(ctx context.Context) error { + groups := []struct { + service string + dirs []string + }{ + { + service: "mysql-app", + dirs: []string{ + "submitqueue/extension/storage/mysql/schema", + "platform/extension/counter/mysql/schema", + }, + }, + { + service: "mysql-queue", + dirs: []string{"platform/extension/messagequeue/mysql/schema"}, + }, + } + for _, group := range groups { + for _, directory := range group.dirs { + files, err := filepath.Glob(filepath.Join(s.repoRoot, directory, "*.sql")) + if err != nil { + return fmt.Errorf("list schema files: %w", err) + } + sort.Strings(files) + if len(files) == 0 { + return fmt.Errorf("no schema files found in %s", directory) + } + for _, path := range files { + content, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read schema %s: %w", path, err) + } + fmt.Fprintf(s.stdout, "Applying %s to %s...\n", filepath.Base(path), group.service) + if err := s.composeRun(ctx, bytes.NewReader(content), s.stdout, s.stderr, + "exec", "-T", group.service, "mysql", "-uroot", "-proot", "submitqueue", + ); err != nil { + return fmt.Errorf("apply schema %s: %w", path, err) + } + } + } + } + return nil +} + +func (s *localStack) GatewayAddress(ctx context.Context) string { + output, err := s.composeOutput(ctx, "port", "gateway-service", "8080") + if err != nil { + return "" + } + port, err := parsePort(string(output)) + if err != nil { + return "" + } + return "localhost:" + strconv.Itoa(port) +} + +func (s *localStack) Logs(ctx context.Context) { + if !s.started { + return + } + _ = s.composeRun(ctx, nil, s.stderr, s.stderr, "logs", "--no-color") +} + +func (s *localStack) Close(ctx context.Context) error { + if !s.started { + return nil + } + s.started = false + if err := s.composeRun(ctx, nil, s.stdout, s.stderr, "down", "-v", "--rmi", "local"); err != nil { + return fmt.Errorf("tear down stack %s: %w", s.project, err) + } + return nil +} + +func (s *localStack) composeRun(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, args ...string) error { + base := []string{"compose", "-f", s.compose, "-p", s.project} + return s.exec.Run(ctx, s.repoRoot, s.baseEnv, stdin, stdout, stderr, "docker", append(base, args...)...) +} + +func (s *localStack) composeOutput(ctx context.Context, args ...string) ([]byte, error) { + base := []string{"compose", "-f", s.compose, "-p", s.project} + return s.exec.Output(ctx, s.repoRoot, s.baseEnv, "docker", append(base, args...)...) +} + +func waitForGateway(ctx context.Context, address string, timeoutMs int64) (*GRPCGateway, error) { + if address == "" { + return nil, fmt.Errorf("gateway port is unavailable") + } + gateway, err := NewGRPCGateway(address) + if err != nil { + return nil, err + } + deadline := time.NewTimer(min(time.Duration(timeoutMs)*time.Millisecond, 30*time.Second)) + defer deadline.Stop() + ticker := time.NewTicker(250 * time.Millisecond) + defer ticker.Stop() + for { + pingCtx, cancel := context.WithTimeout(ctx, time.Second) + err := gateway.Ping(pingCtx) + cancel() + if err == nil { + return gateway, nil + } + select { + case <-ctx.Done(): + gateway.Close() + return nil, ctx.Err() + case <-deadline.C: + gateway.Close() + return nil, fmt.Errorf("gateway %s did not become ready", address) + case <-ticker.C: + } + } +} + +func parsePort(output string) (int, error) { + value := strings.TrimSpace(output) + index := strings.LastIndex(value, ":") + if index < 0 { + return 0, fmt.Errorf("invalid port output %q", value) + } + port, err := strconv.Atoi(value[index+1:]) + if err != nil { + return 0, fmt.Errorf("invalid port output %q: %w", value, err) + } + return port, nil +} + +func findRepoRoot(ctx context.Context) (string, error) { + for _, key := range []string{"REPO_ROOT", "BUILD_WORKSPACE_DIRECTORY"} { + if root := os.Getenv(key); root != "" { + return root, nil + } + } + command := exec.CommandContext(ctx, "git", "rev-parse", "--show-toplevel") + output, err := command.Output() + if err != nil { + return "", fmt.Errorf("find repository root: %w", err) + } + root := strings.TrimSpace(string(output)) + if root == "" { + return "", fmt.Errorf("repository root is empty") + } + return root, nil +} diff --git a/sqsim/runner/local_test.go b/sqsim/runner/local_test.go new file mode 100644 index 00000000..e3bb495d --- /dev/null +++ b/sqsim/runner/local_test.go @@ -0,0 +1,41 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParsePort(t *testing.T) { + tests := map[string]int{ + "0.0.0.0:49153\n": 49153, + "[::]:49154": 49154, + } + for output, want := range tests { + got, err := parsePort(output) + require.NoError(t, err) + assert.Equal(t, want, got) + } +} + +func TestFindRepoRootUsesWorkspaceEnvironment(t *testing.T) { + t.Setenv("REPO_ROOT", "/tmp/repo") + root, err := findRepoRoot(t.Context()) + require.NoError(t, err) + assert.Equal(t, "/tmp/repo", root) +} diff --git a/sqsim/runner/runner.go b/sqsim/runner/runner.go new file mode 100644 index 00000000..5c1c755c --- /dev/null +++ b/sqsim/runner/runner.go @@ -0,0 +1,413 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package runner submits and verifies SQSim scenarios against SubmitQueue. +package runner + +import ( + "context" + "fmt" + "sort" + "time" + + "github.com/uber/submitqueue/sqsim" + "github.com/uber/submitqueue/sqsim/model" +) + +// Gateway is the public SubmitQueue surface used by SQSim. +type Gateway interface { + // Land submits one synthetic change. + Land(ctx context.Context, queue, changeURI string) (string, error) + // List returns one page of request summaries for a queue. + List(ctx context.Context, queue string, receivedAtOrAfterMs, receivedBeforeMs int64, pageToken string) ([]Summary, string, error) + // Summary returns the current summary for one sqid. + Summary(ctx context.Context, sqid string) (Summary, error) + // History returns retained lifecycle events for one sqid. + History(ctx context.Context, sqid string) ([]HistoryEvent, error) +} + +// Summary is the public current view of one request. +type Summary struct { + // SQID is the request identifier returned by Land. + SQID string + // Status is the customer-facing request status. + Status string + // LastError is the latest reported failure. + LastError string + // Metadata is display and debugging context. + Metadata map[string]string +} + +// HistoryEvent is one retained public lifecycle event. +type HistoryEvent struct { + // TimestampMs is the event creation time in Unix milliseconds. + TimestampMs int64 + // Status is the customer-facing request status. + Status string + // LastError is the error associated with the event. + LastError string + // Metadata is event-specific context. + Metadata map[string]string +} + +// Request is the runner's current view of one scenario Land. +type Request struct { + // Name is the Land name from the scenario. + Name string + // SQID is the request identifier returned by Gateway Land. + SQID string + // Status is the latest public request status. + Status string + // Expected is the required terminal status. + Expected string + // LastError is the latest public failure. + LastError string + // Metadata is the latest public display context. + Metadata map[string]string + // History is populated lazily after a terminal status is confirmed. + History []HistoryEvent +} + +// Snapshot is one observable point in a scenario run. +type Snapshot struct { + // Scenario is the public scenario name. + Scenario string + // StartedAt is the run start time. + StartedAt time.Time + // Now is the observation time. + Now time.Time + // Requests are in scenario declaration order. + Requests []Request + // Done reports whether every request is terminal. + Done bool +} + +// Observer receives snapshots when visible state changes. +type Observer interface { + // Observe receives an immutable snapshot. + Observe(Snapshot) +} + +// ObserverFunc adapts a function to Observer. +type ObserverFunc func(Snapshot) + +// Observe calls the wrapped function. +func (f ObserverFunc) Observe(snapshot Snapshot) { + f(snapshot) +} + +// Report is the verified result of a completed run. +type Report struct { + // Scenario is the public scenario name. + Scenario string + // Requests are the final request views. + Requests []Request + // Passed is true when every terminal status matched its expectation. + Passed bool +} + +// Options configure one engine run. +type Options struct { + // ScenarioName is the selected public scenario name. + ScenarioName string + // Scenario is the immutable workload. + Scenario sqsim.Scenario + // Gateway is the public API client. + Gateway Gateway + // Clock supplies wall time and cancellable waits. + Clock model.Clock + // PollInterval bounds public API polling frequency. + PollInterval time.Duration + // Observer receives visible state changes. + Observer Observer +} + +type requestState struct { + request Request + queue string + changeURI string + submitAfter time.Duration + submitted bool + terminalObserved bool + historyLoaded bool +} + +// Run executes and verifies one scenario. +func Run(ctx context.Context, options Options) (Report, error) { + if options.Gateway == nil { + return Report{}, fmt.Errorf("gateway is required") + } + if options.Clock == nil { + return Report{}, fmt.Errorf("clock is required") + } + if options.PollInterval <= 0 { + return Report{}, fmt.Errorf("poll interval must be positive") + } + if err := sqsim.Validate(options.Scenario); err != nil { + return Report{}, fmt.Errorf("validate scenario: %w", err) + } + + runCtx, cancel := context.WithTimeout(ctx, time.Duration(options.Scenario.TimeoutMs)*time.Millisecond) + defer cancel() + + startedAt := options.Clock.Now() + states := make([]requestState, len(options.Scenario.Lands)) + for i, land := range options.Scenario.Lands { + changeURI, err := model.ChangeURI(options.ScenarioName, land.Name) + if err != nil { + return Report{}, err + } + states[i] = requestState{ + request: Request{ + Name: land.Name, + Expected: string(land.Expectation.Status), + Metadata: map[string]string{}, + }, + queue: land.Queue, + changeURI: changeURI, + submitAfter: time.Duration(land.SubmitAfterMs) * time.Millisecond, + } + } + + emit(options, startedAt, states, false) + for { + now := options.Clock.Now() + changed := false + submitted := 0 + for i := range states { + state := &states[i] + if state.submitted { + submitted++ + continue + } + if now.Sub(startedAt) < state.submitAfter { + continue + } + sqid, err := options.Gateway.Land(runCtx, state.queue, state.changeURI) + if err != nil { + return Report{}, fmt.Errorf("land %q: %w", state.request.Name, err) + } + if sqid == "" { + return Report{}, fmt.Errorf("land %q returned an empty sqid", state.request.Name) + } + state.request.SQID = sqid + state.request.Status = "submitted" + state.submitted = true + submitted++ + changed = true + } + + pollChanged, err := poll(runCtx, options, startedAt, states) + if err != nil { + if runCtx.Err() != nil { + return Report{}, fmt.Errorf("scenario %q timed out: %w", options.ScenarioName, runCtx.Err()) + } + return Report{}, err + } + changed = changed || pollChanged + + done := submitted == len(states) && allTerminal(states) + if changed || done { + emit(options, startedAt, states, done) + } + if done { + requests := requestsFrom(states) + passed := true + for _, request := range requests { + if request.Status != request.Expected { + passed = false + } + } + return Report{Scenario: options.ScenarioName, Requests: requests, Passed: passed}, nil + } + + wait := options.PollInterval + for i := range states { + if states[i].submitted { + continue + } + untilSubmit := states[i].submitAfter - options.Clock.Now().Sub(startedAt) + if untilSubmit < wait { + wait = untilSubmit + } + } + if wait < 0 { + wait = 0 + } + if err := options.Clock.Wait(runCtx, wait); err != nil { + return Report{}, fmt.Errorf("scenario %q timed out: %w", options.ScenarioName, err) + } + } +} + +func poll(ctx context.Context, options Options, startedAt time.Time, states []requestState) (bool, error) { + if len(states) == 0 { + return false, nil + } + bySQID := make(map[string]*requestState, len(states)) + queues := make(map[string]struct{}) + for i := range states { + if !states[i].submitted { + continue + } + bySQID[states[i].request.SQID] = &states[i] + queues[states[i].queue] = struct{}{} + } + + found := make(map[string]struct{}, len(states)) + queueNames := make([]string, 0, len(queues)) + for queue := range queues { + queueNames = append(queueNames, queue) + } + sort.Strings(queueNames) + + changed := false + for _, queue := range queueNames { + pageToken := "" + for { + summaries, nextToken, err := options.Gateway.List( + ctx, + queue, + startedAt.Add(-time.Second).UnixMilli(), + options.Clock.Now().Add(time.Minute).UnixMilli(), + pageToken, + ) + if err != nil { + break + } + for _, summary := range summaries { + state, ok := bySQID[summary.SQID] + if !ok { + continue + } + found[summary.SQID] = struct{}{} + if applySummary(state, summary) { + changed = true + } + } + if nextToken == "" { + break + } + pageToken = nextToken + } + } + + for i := range states { + state := &states[i] + if !state.submitted { + continue + } + _, listed := found[state.request.SQID] + if !listed || (isTerminal(state.request.Status) && !state.terminalObserved) { + summary, err := options.Gateway.Summary(ctx, state.request.SQID) + if err == nil { + if applySummary(state, summary) { + changed = true + } + } + } + if isTerminal(state.request.Status) { + state.terminalObserved = true + if !state.historyLoaded { + history, err := options.Gateway.History(ctx, state.request.SQID) + if err == nil { + state.request.History = append([]HistoryEvent(nil), history...) + state.historyLoaded = true + changed = true + } + } + } + } + return changed, nil +} + +func applySummary(state *requestState, summary Summary) bool { + if summary.Status == "" { + return false + } + changed := state.request.Status != summary.Status || + state.request.LastError != summary.LastError || + !equalMetadata(state.request.Metadata, summary.Metadata) + state.request.Status = summary.Status + state.request.LastError = summary.LastError + state.request.Metadata = cloneMetadata(summary.Metadata) + return changed +} + +func allTerminal(states []requestState) bool { + for _, state := range states { + if !state.submitted || !isTerminal(state.request.Status) || !state.historyLoaded { + return false + } + } + return true +} + +func isTerminal(status string) bool { + return status == string(sqsim.RequestLanded) || + status == string(sqsim.RequestError) || + status == string(sqsim.RequestCancelled) +} + +func emit(options Options, startedAt time.Time, states []requestState, done bool) { + if options.Observer == nil { + return + } + options.Observer.Observe(Snapshot{ + Scenario: options.ScenarioName, + StartedAt: startedAt, + Now: options.Clock.Now(), + Requests: requestsFrom(states), + Done: done, + }) +} + +func requestsFrom(states []requestState) []Request { + requests := make([]Request, len(states)) + for i, state := range states { + requests[i] = state.request + requests[i].Metadata = cloneMetadata(state.request.Metadata) + requests[i].History = cloneHistory(state.request.History) + } + return requests +} + +func cloneHistory(history []HistoryEvent) []HistoryEvent { + cloned := make([]HistoryEvent, len(history)) + for i, event := range history { + cloned[i] = event + cloned[i].Metadata = cloneMetadata(event.Metadata) + } + return cloned +} + +func cloneMetadata(metadata map[string]string) map[string]string { + cloned := make(map[string]string, len(metadata)) + for key, value := range metadata { + cloned[key] = value + } + return cloned +} + +func equalMetadata(left, right map[string]string) bool { + if len(left) != len(right) { + return false + } + for key, value := range left { + if right[key] != value { + return false + } + } + return true +} diff --git a/sqsim/runner/runner_test.go b/sqsim/runner/runner_test.go new file mode 100644 index 00000000..9003b65b --- /dev/null +++ b/sqsim/runner/runner_test.go @@ -0,0 +1,162 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/sqsim" +) + +func TestRunSubmitsObservesAndVerifiesScenario(t *testing.T) { + scenario, err := sqsim.NewScenario(). + Timeout(time.Minute). + Land( + sqsim.NewLand("l1").Queue("sqsim").SubmitAfter(time.Second).Behavior(testBehavior()).Expect(sqsim.RequestError), + sqsim.NewLand("l2").Queue("sqsim").Behavior(testBehavior()).Expect(sqsim.RequestLanded), + ). + Build() + require.NoError(t, err) + + clock := &testClock{now: time.Unix(0, 0)} + gateway := &testGateway{ + statuses: map[string][]string{ + "sqsim/1": {"accepted", "landed"}, + "sqsim/2": {"accepted", "building", "error"}, + }, + } + var snapshots []Snapshot + report, err := Run(context.Background(), Options{ + ScenarioName: "mixed", + Scenario: scenario, + Gateway: gateway, + Clock: clock, + PollInterval: time.Second, + Observer: ObserverFunc(func(snapshot Snapshot) { + snapshots = append(snapshots, snapshot) + }), + }) + require.NoError(t, err) + assert.True(t, report.Passed) + assert.Equal(t, []string{ + "sqsim://local/mixed/l2", + "sqsim://local/mixed/l1", + }, gateway.uris) + assert.True(t, snapshots[len(snapshots)-1].Done) + require.Len(t, report.Requests[1].History, 1) +} + +func TestRunReportsExpectationMismatch(t *testing.T) { + scenario, err := sqsim.NewScenario(). + Timeout(time.Minute). + Land(sqsim.NewLand("l1").Queue("sqsim").Behavior(testBehavior()).Expect(sqsim.RequestLanded)). + Build() + require.NoError(t, err) + gateway := &testGateway{statuses: map[string][]string{"sqsim/1": {"error"}}} + + report, err := Run(context.Background(), Options{ + ScenarioName: "mismatch", + Scenario: scenario, + Gateway: gateway, + Clock: &testClock{now: time.Unix(0, 0)}, + PollInterval: time.Second, + }) + require.NoError(t, err) + assert.False(t, report.Passed) +} + +func testBehavior() *sqsim.BehaviorBuilder { + return sqsim.NewBehavior(). + BuildRunner(sqsim.SuccessfulBuildRunner()). + MergeConflictCheck(sqsim.SuccessfulMergeConflictCheck()). + Merge(sqsim.SuccessfulMerge()) +} + +type testGateway struct { + mu sync.Mutex + uris []string + statuses map[string][]string + polls map[string]int +} + +func (g *testGateway) Land(_ context.Context, _ string, uri string) (string, error) { + g.mu.Lock() + defer g.mu.Unlock() + g.uris = append(g.uris, uri) + return fmt.Sprintf("sqsim/%d", len(g.uris)), nil +} + +func (g *testGateway) List(_ context.Context, _ string, _, _ int64, _ string) ([]Summary, string, error) { + g.mu.Lock() + defer g.mu.Unlock() + if g.polls == nil { + g.polls = make(map[string]int) + } + summaries := make([]Summary, 0, len(g.uris)) + for i := range g.uris { + sqid := fmt.Sprintf("sqsim/%d", i+1) + statuses := g.statuses[sqid] + poll := g.polls[sqid] + if poll >= len(statuses) { + poll = len(statuses) - 1 + } + summaries = append(summaries, Summary{SQID: sqid, Status: statuses[poll]}) + g.polls[sqid]++ + } + return summaries, "", nil +} + +func (g *testGateway) Summary(_ context.Context, sqid string) (Summary, error) { + g.mu.Lock() + defer g.mu.Unlock() + statuses := g.statuses[sqid] + if len(statuses) == 0 { + return Summary{}, fmt.Errorf("not found") + } + return Summary{SQID: sqid, Status: statuses[len(statuses)-1]}, nil +} + +func (g *testGateway) History(_ context.Context, _ string) ([]HistoryEvent, error) { + return []HistoryEvent{{Status: "terminal"}}, nil +} + +type testClock struct { + mu sync.Mutex + now time.Time +} + +func (c *testClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *testClock) Wait(ctx context.Context, duration time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + c.mu.Lock() + c.now = c.now.Add(duration) + c.mu.Unlock() + return nil + } +} diff --git a/sqsim/runner/text.go b/sqsim/runner/text.go new file mode 100644 index 00000000..1522d597 --- /dev/null +++ b/sqsim/runner/text.go @@ -0,0 +1,53 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "fmt" + "io" + "sync" +) + +// TextObserver prints timestamped request transitions. +type TextObserver struct { + mu sync.Mutex + writer io.Writer + statuses map[string]string +} + +// NewTextObserver returns a headless transition observer. +func NewTextObserver(writer io.Writer) *TextObserver { + return &TextObserver{writer: writer, statuses: make(map[string]string)} +} + +// Observe prints request statuses that changed since the previous snapshot. +func (o *TextObserver) Observe(snapshot Snapshot) { + o.mu.Lock() + defer o.mu.Unlock() + for _, request := range snapshot.Requests { + key := request.Name + status := request.Status + if status == "" || o.statuses[key] == status { + continue + } + o.statuses[key] = status + fmt.Fprintf(o.writer, "[%s] %-16s %-18s %s\n", + snapshot.Now.Sub(snapshot.StartedAt).Round(10_000_000), + request.Name, + status, + request.SQID, + ) + } +} diff --git a/sqsim/runner/text_test.go b/sqsim/runner/text_test.go new file mode 100644 index 00000000..fe576061 --- /dev/null +++ b/sqsim/runner/text_test.go @@ -0,0 +1,37 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package runner + +import ( + "bytes" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestTextObserverPrintsEachTransitionOnce(t *testing.T) { + var output bytes.Buffer + observer := NewTextObserver(&output) + snapshot := Snapshot{ + StartedAt: time.Unix(0, 0), + Now: time.Unix(1, 0), + Requests: []Request{{Name: "l1", SQID: "sqsim/1", Status: "building"}}, + } + observer.Observe(snapshot) + observer.Observe(snapshot) + assert.Contains(t, output.String(), "building") + assert.Equal(t, 1, bytes.Count(output.Bytes(), []byte("building"))) +} diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index a6af9252..dddba2ad 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -167,11 +167,12 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return nil } - build.Status = status - - if err := c.store.GetBuildStore().UpdateStatus(ctx, build.ID, status); 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 build.Status != status { + if err := c.store.GetBuildStore().UpdateStatus(ctx, build.ID, status); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) + return fmt.Errorf("failed to update status for build %s: %w", build.ID, err) + } + build.Status = status } if status == entity.BuildStatusSucceeded { diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index 3d87c79e..3e7c0982 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -195,16 +195,17 @@ func TestController_Process_SucceededPublishesBuiltLog(t *testing.T) { } // 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. +// status transitions, publishes to speculate, AND re-publishes to buildsignal +// via PublishAfter with the per-status delay. func TestController_Process_NonTerminal(t *testing.T) { tests := []struct { name string status entity.BuildStatus wantDelayMs int64 + wantUpdate bool }{ - {"accepted uses accepted delay", entity.BuildStatusAccepted, PollDelayAcceptedMs}, - {"running uses running delay", entity.BuildStatusRunning, PollDelayRunningMs}, + {"unchanged accepted uses accepted delay", entity.BuildStatusAccepted, PollDelayAcceptedMs, false}, + {"transition to running uses running delay", entity.BuildStatusRunning, PollDelayRunningMs, true}, } for _, tt := range tests { @@ -217,7 +218,9 @@ func TestController_Process_NonTerminal(t *testing.T) { h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(tt.status, entity.BuildMetadata{}, nil) h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, tt.status).Return(nil) + if tt.wantUpdate { + h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, tt.status).Return(nil) + } h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1) h.signalPub.EXPECT(). PublishAfter(gomock.Any(), "buildsignal", gomock.AssignableToTypeOf(entityqueue.Message{}), tt.wantDelayMs). diff --git a/tool/sqsim/BUILD.bazel b/tool/sqsim/BUILD.bazel index fb7faf48..6ac1d17e 100644 --- a/tool/sqsim/BUILD.bazel +++ b/tool/sqsim/BUILD.bazel @@ -5,7 +5,10 @@ go_library( srcs = ["main.go"], importpath = "github.com/uber/submitqueue/tool/sqsim", visibility = ["//visibility:private"], - deps = ["//sqsim/scenarios:go_default_library"], + deps = [ + "//sqsim/runner:go_default_library", + "//sqsim/scenarios:go_default_library", + ], ) go_binary( @@ -18,5 +21,8 @@ go_test( name = "go_default_test", srcs = ["main_test.go"], embed = [":go_default_library"], - deps = ["@com_github_stretchr_testify//assert:go_default_library"], + deps = [ + "//sqsim/runner:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + ], ) diff --git a/tool/sqsim/main.go b/tool/sqsim/main.go index 8d54728d..d49e9e95 100644 --- a/tool/sqsim/main.go +++ b/tool/sqsim/main.go @@ -15,10 +15,14 @@ package main import ( + "context" + "flag" "fmt" "io" "os" + "time" + "github.com/uber/submitqueue/sqsim/runner" "github.com/uber/submitqueue/sqsim/scenarios" ) @@ -53,6 +57,8 @@ func run(args []string, stdout, stderr io.Writer) int { } fmt.Fprintf(stdout, "%s is valid\n", args[1]) return 0 + case "run": + return runScenarioCommand(args[1:], stdout, stderr) default: fmt.Fprintf(stderr, "unknown command %q\n", args[0]) printUsage(stderr) @@ -64,4 +70,54 @@ func printUsage(w io.Writer) { fmt.Fprintln(w, "usage:") fmt.Fprintln(w, " sqsim list") fmt.Fprintln(w, " sqsim validate ") + fmt.Fprintln(w, " sqsim run --headless") +} + +var runLocal = runner.RunLocal + +func runScenarioCommand(args []string, stdout, stderr io.Writer) int { + if len(args) == 0 { + printUsage(stderr) + return 2 + } + scenarioName := args[0] + flags := flag.NewFlagSet("sqsim run", flag.ContinueOnError) + flags.SetOutput(stderr) + headless := flags.Bool("headless", false, "print timestamped transitions") + pollInterval := flags.Duration("poll-interval", 250*time.Millisecond, "public API polling interval") + if err := flags.Parse(args[1:]); err != nil { + return 2 + } + if !*headless { + fmt.Fprintln(stderr, "interactive mode is not available yet; pass --headless") + return 2 + } + scenario, err := scenarios.Build(scenarioName) + if err != nil { + fmt.Fprintf(stderr, "invalid scenario: %v\n", err) + return 2 + } + report, err := runLocal(context.Background(), runner.LocalOptions{ + ScenarioName: scenarioName, + Scenario: scenario, + Observer: runner.NewTextObserver(stdout), + Stdout: stdout, + Stderr: stderr, + PollInterval: *pollInterval, + }) + if err != nil { + fmt.Fprintf(stderr, "sqsim infrastructure failure: %v\n", err) + return 2 + } + for _, request := range report.Requests { + result := "PASS" + if request.Status != request.Expected { + result = "FAIL" + } + fmt.Fprintf(stdout, "%s %s: got %s, expected %s\n", result, request.Name, request.Status, request.Expected) + } + if !report.Passed { + return 1 + } + return 0 } diff --git a/tool/sqsim/main_test.go b/tool/sqsim/main_test.go index 2a876f81..2728f24c 100644 --- a/tool/sqsim/main_test.go +++ b/tool/sqsim/main_test.go @@ -16,9 +16,11 @@ package main import ( "bytes" + "context" "testing" "github.com/stretchr/testify/assert" + "github.com/uber/submitqueue/sqsim/runner" ) func TestRun(t *testing.T) { @@ -46,3 +48,29 @@ func TestRun(t *testing.T) { }) } } + +func TestRunScenarioCommandExitCodes(t *testing.T) { + original := runLocal + t.Cleanup(func() { runLocal = original }) + + tests := []struct { + name string + report runner.Report + err error + want int + }{ + {name: "success", report: runner.Report{Passed: true}, want: 0}, + {name: "expectation failure", report: runner.Report{Passed: false}, want: 1}, + {name: "infrastructure failure", err: assert.AnError, want: 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + runLocal = func(context.Context, runner.LocalOptions) (runner.Report, error) { + return tt.report, tt.err + } + var stdout bytes.Buffer + var stderr bytes.Buffer + assert.Equal(t, tt.want, runScenarioCommand([]string{"happy-path", "--headless"}, &stdout, &stderr)) + }) + } +}