From 8796a89501ef241bf7c3324deeea8d4f753494fb Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Mon, 21 Sep 2026 12:47:25 -0700 Subject: [PATCH 1/3] e2e: add the rolling upgrade suite The suite takes a cluster running a release with actors on it and upgrades it to the checkout under test in the order of docs/upgrade.md: podcertificate-controller, atelet, the WorkerPool's workerImage, the API server, ate-controller, atenet, the SandboxConfig, then the rest of the system bundle. Every step except the workerImage edit is one ate-setup command, run from a copy of the install record. The suite stands in for the actors' harness and suspends an actor as soon as its worker pod starts terminating. Each step that deploys a component has to move it to new images, and no actor may be CRASHED at any step. Only the pool roll rolls the workers, plus ate-controller if its pod template changed, and only those steps may evict an actor. Afterwards each actor resumes on a worker running the new image with its memory intact. The suite upgrades the cluster it runs on, so it lives outside internal/e2e/suites, where the default e2e run would pick it up. It runs on gVisor only. The workflow runs it for every push to a release branch, against the newest earlier release on that branch, so a patch release is checked before it is tagged. It also runs on demand from a given release. Two installs on one kind cluster are too slow for every PR. --- .github/workflows/upgrade-e2e.yaml | 149 +++++ internal/e2e/README.md | 8 +- internal/e2e/upgrade/testmain_test.go | 24 + internal/e2e/upgrade/upgrade_test.go | 848 ++++++++++++++++++++++++++ 4 files changed, 1027 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/upgrade-e2e.yaml create mode 100644 internal/e2e/upgrade/testmain_test.go create mode 100644 internal/e2e/upgrade/upgrade_test.go diff --git a/.github/workflows/upgrade-e2e.yaml b/.github/workflows/upgrade-e2e.yaml new file mode 100644 index 0000000000..832880a784 --- /dev/null +++ b/.github/workflows/upgrade-e2e.yaml @@ -0,0 +1,149 @@ +# Copyright 2026 Google LLC +# +# 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. + +# Upgrades a cluster running a release to this checkout, with actors on it, +# following docs/upgrade.md, and checks that every actor survives (see +# internal/e2e/upgrade). Two installs on one kind cluster take too long for +# every PR, so this runs for every push to a release branch, before a patch +# release is tagged from it, plus on demand for any branch. +name: upgrade-e2e +on: + push: + branches: ['release-[0-9]+.[0-9]+'] + workflow_dispatch: + inputs: + from: + description: 'Tag or commit to upgrade from. Required except on a release branch, which defaults to its newest earlier release' + required: false + default: '' +permissions: + contents: read +jobs: + upgrade: + name: Upgrade e2e (gVisor) + runs-on: ubuntu-latest + timeout-minutes: 90 + env: + # Where the previous release's ate-setup writes its install record. + ATE_RECORD_DIR: ${{ github.workspace }}/ate-records + # deploy ate-system requires a credential provider selection. The + # record keeps it for the upgrade. + ATE_CREDENTIAL_PROVIDER: '{"name":"k8s.io"}' + ARTIFACTS: ${{ github.workspace }}/_artifacts + steps: + - name: Checkout the release under test + uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0 + with: + path: new + # Full history brings the release tags the next step picks from. + fetch-depth: 0 + - name: Pick the release to upgrade from + id: from + working-directory: new + env: + FROM: ${{ inputs.from }} + TO: ${{ github.ref_name }} + EVENT: ${{ github.event_name }} + run: | + if [[ -z "${FROM}" && "${TO}" =~ ^release-[0-9]+\.[0-9]+$ ]]; then + # The newest earlier release of the branch's minor (release-0.4: + # v0.4.*): a release branch only takes patch releases. Release tags, + # newest first; versionsort.suffix sorts a pre-release (v0.2.0-rc1) + # below its release. + FROM="$(git -c versionsort.suffix=- tag -l 'v*' --merged HEAD --no-contains HEAD --sort=-v:refname \ + | awk -v minor="v${TO#release-}." 'index($0, minor) == 1 {print; exit}')" + fi + if [[ -z "${FROM}" ]]; then + if [[ "${EVENT}" == workflow_dispatch ]]; then + echo "::error::Nothing to upgrade ${TO} from: set the from input." + exit 1 + fi + echo "::notice::No earlier release on ${TO} to upgrade from yet." + echo "tag=" >> "${GITHUB_OUTPUT}" + exit 0 + fi + echo "Upgrading ${FROM} -> ${TO}" + echo "tag=${FROM}" >> "${GITHUB_OUTPUT}" + - name: Checkout ${{ steps.from.outputs.tag }} + if: steps.from.outputs.tag != '' + uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0 + with: + ref: ${{ steps.from.outputs.tag }} + path: old + - name: Setup Go + if: steps.from.outputs.tag != '' + uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0 + with: + go-version-file: new/go.mod + cache-dependency-path: | + new/go.sum + old/go.sum + - name: Create cluster + if: steps.from.outputs.tag != '' + working-directory: new + run: hack/create-kind-cluster.sh + - name: Install ${{ steps.from.outputs.tag }} + if: steps.from.outputs.tag != '' + # The old release's own installer, so the cluster starts out exactly as + # that release left it. The suite's actors run on the counter demo. + working-directory: old + run: | + hack/install-ate-kind.sh --deploy-ate-system + hack/install-ate-kind.sh --deploy-demo-counter + - name: Upgrade to ${{ github.ref_name }} + if: steps.from.outputs.tag != '' + # The suite performs the upgrade with this checkout's ate-setup, in the + # runbook's order and from a copy of the install record, and checks the + # actors after every step. + working-directory: new + env: + E2E_ATE_SETUP_CONFIG: ${{ github.workspace }}/new.yaml + E2E_JUNIT_FILE: ${{ github.workspace }}/_artifacts/e2e-upgrade.xml + run: | + cp "${ATE_RECORD_DIR}"/installs/*.yaml "${E2E_ATE_SETUP_CONFIG}" + hack/run-e2e-kind.sh ./internal/e2e/upgrade -timeout 60m -v -args --no-color + # A suite that stops matching would otherwise leave the run green and empty. + - name: Require the suite to have run tests + if: always() && steps.from.outputs.tag != '' + working-directory: new + run: go -C tools/junittool run . verify -manifest "${ARTIFACTS}/expected-junit.txt" + - name: Upload test results + if: always() && steps.from.outputs.tag != '' + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: test-results-upgrade-e2e + # The install records carry no secrets. + path: | + ${{ env.ARTIFACTS }} + ${{ env.ATE_RECORD_DIR }} + if-no-files-found: error + - name: Dump diagnostics on failure + if: failure() && steps.from.outputs.tag != '' + run: | + kubectl --context kind-kind get workerpool,pods -A -o wide || true + kubectl --context kind-kind get ds,deploy -n ate-system -o wide || true + dump() { + echo "=== logs: $1/$2 ===" + kubectl --context kind-kind logs -n "$1" "$2" --all-containers --tail=300 2>/dev/null || true + } + for ns in ate-system podcertificate-controller-system; do + for p in $(kubectl --context kind-kind get pods -n "${ns}" -o name 2>/dev/null); do + dump "${ns}" "$p" + done + done + # The actors' worker pods are in the counter demo's pool; their worker + # logs say what happened to each actor. + kubectl --context kind-kind get pods -A -l ate.dev/worker-pool \ + -o 'custom-columns=:.metadata.namespace,:.metadata.name' --no-headers 2>/dev/null \ + | while read -r ns name; do dump "$ns" "$name"; done diff --git a/internal/e2e/README.md b/internal/e2e/README.md index 7b72641665..7f9b0a07fd 100644 --- a/internal/e2e/README.md +++ b/internal/e2e/README.md @@ -8,7 +8,9 @@ $ go test -v ./internal/e2e/suites/... -args --e2e ## Principles * Keep it simple -- use go test for the harness. -* e2e tests live under `internal/e2e/suites/` +* e2e tests live under `internal/e2e/suites/`. The upgrade suite lives in + `internal/e2e/upgrade` instead, so the default run leaves it out: it upgrades + the cluster it runs on. `.github/workflows/upgrade-e2e.yaml` runs it. * Each suite should implement TestMain using e2e.RunTestMain() * e2e tests will be skipped for ordinary unit tests unless the `--e2e` flag is set e.g. `go test ./internal/e2e/suites/... -args --e2e` @@ -54,7 +56,9 @@ installs the fixtures elsewhere: `E2E_SUBSTRATE_TEMPLATE_ATESPACE` / `E2E_SUBSTRATE_POOL_NAME` point the counter fixture somewhere else, and `E2E_TEMPLATE_READY_TIMEOUT` replaces the golden-snapshot budget (90s on gVisor, 10m on micro-VM, where the golden is a cloud-hypervisor cold boot plus -a checkpoint). +a checkpoint). The upgrade suite runs `ate-setup` with `--config +$E2E_ATE_SETUP_CONFIG` when it is set, and otherwise with `--kind`, which +`E2E_ATE_SETUP_FLAGS` replaces. ## Settings diff --git a/internal/e2e/upgrade/testmain_test.go b/internal/e2e/upgrade/testmain_test.go new file mode 100644 index 0000000000..ecc2b4a129 --- /dev/null +++ b/internal/e2e/upgrade/testmain_test.go @@ -0,0 +1,24 @@ +// Copyright 2026 Google LLC +// +// 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 upgrade + +import ( + "os" + "testing" + + "github.com/agent-substrate/substrate/internal/e2e" +) + +func TestMain(m *testing.M) { os.Exit(e2e.RunTestMain(m)) } diff --git a/internal/e2e/upgrade/upgrade_test.go b/internal/e2e/upgrade/upgrade_test.go new file mode 100644 index 0000000000..1734a02f83 --- /dev/null +++ b/internal/e2e/upgrade/upgrade_test.go @@ -0,0 +1,848 @@ +// Copyright 2026 Google LLC +// +// 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 upgrade is the rolling upgrade suite. It takes a cluster that runs +// the previous release with actors on it and upgrades it to this checkout in +// the order of docs/upgrade.md: podcertificate-controller, atelet, the +// WorkerPool's workerImage, the API server, ate-controller, atenet, the +// SandboxConfig, then the rest of the system bundle. Every step except the +// workerImage edit is one ate-setup command. The suite stands in for the +// actors' harness and suspends an actor as soon as its worker pod starts +// terminating. It passes when no actor is CRASHED at any step and every actor +// resumes on a new worker with its memory. +// +// The suite upgrades the cluster it runs on, so it lives outside +// internal/e2e/suites and the default e2e run leaves it out. It runs on gVisor +// only. The actors run on the counter demo the previous release installed +// (hack/install-ate-kind.sh --deploy-demo-counter, from that checkout). From +// this checkout: +// +// hack/run-e2e-kind.sh ./internal/e2e/upgrade -timeout 60m +// +// ate-setup runs with --config E2E_ATE_SETUP_CONFIG, the previous release's +// install record, as the runbook does. Without it, ate-setup runs with --kind, +// or with E2E_ATE_SETUP_FLAGS for another cluster, and the test's kube context. +// .github/workflows/upgrade-e2e.yaml runs the suite for every push to a +// release branch, against the newest release on that branch. +package upgrade + +import ( + "context" + "encoding/json" + "fmt" + "io" + "math/rand/v2" + "net/http" + "os" + "path/filepath" + "regexp" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/portforward" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const ( + // sigtermSleepSeconds is how long the counter keeps running after + // SIGTERM: a day, so a snapshot taken after SIGTERM restores with most + // of a day left before the exit fires. The eviction contract asks the + // same of an actor: stay up until suspended. + sigtermSleepSeconds = 86400 + + pollInterval = time.Second + stateTimeout = 2 * time.Minute + rollTimeout = 15 * time.Minute +) + +// The upgrade steps, in order. The pool roll sits between atelet and the API +// server; the others are ate-setup subcommands. +const ( + stepPodcert = "deploy podcertificate-controller" + stepAtelet = "deploy atelet --rollout-timeout=1h" + stepPoolRoll = "edit workerImage" + stepAPIServer = "deploy apiserver" + stepController = "deploy ate-controller" + stepAtenet = "deploy atenet" + stepSandboxConfig = "deploy sandboxconfig" + stepSystem = "deploy ate-system --setup-csi=none" +) + +// podcertNamespace is where podcertificate-controller runs. +const podcertNamespace = "podcertificate-controller-system" + +// counterResponse matches the counter demo's reply. Every call bumps both +// counters, and the template's snapshots carry both over across every resume. +var counterResponse = regexp.MustCompile(`preserved memory count: (\d+) \| preserved file counter: (\d+)`) + +// startState is how seedActor leaves an actor ahead of the upgrade. A paused +// actor is SUSPENDED with a node-local snapshot. +type startState string + +const ( + startRunning startState = "running" + startPaused startState = "paused" + startSuspended startState = "suspended" +) + +// actor is one of the suite's counter actors and what it expects of it. +type actor struct { + atespace, name string + // before is how the actor is left ahead of the upgrade. + before startState + // calls is the counter value the actor last reported. + calls int +} + +func (a *actor) ref() *ateapipb.ObjectRef { + return &ateapipb.ObjectRef{Atespace: a.atespace, Name: a.name} +} + +func (a *actor) routerRef() resources.ActorRef { + return resources.ActorRef{Atespace: a.atespace, Name: a.name} +} + +func TestRollingUpgrade(t *testing.T) { + // The micro-VM counter pool has two workers, and the suite needs three. + if e2e.IsMicroVM() { + t.Fatal("the upgrade suite runs on gVisor only") + } + ctx := context.Background() + suffix := strconv.FormatUint(uint64(rand.Uint32()), 36) + api := &api{c: e2e.GetClients()} + t.Cleanup(api.close) + k8s := e2e.GetClients().K8s + fixture := e2e.SubstrateCounterFixture() + pool := types.NamespacedName{Namespace: fixture.PoolNamespace, Name: fixture.PoolName} + + installed := describeInstall(ctx, t, api.get(), fixture) + t.Logf("installed: atelet %s, worker image %s", installed.atelet, installed.workerImage) + + newActor := func(name string, before startState) *actor { + return &actor{name: name + "-" + suffix, atespace: fixture.Atespace, before: before} + } + actors := []*actor{ + newActor("running-a", startRunning), + newActor("running-b", startRunning), + newActor("paused", startPaused), + newActor("suspended", startSuspended), + } + router := newRouter(ctx, t) + for _, a := range actors { + createActor(ctx, t, api, fixture, a) + seedActor(ctx, t, api, router, a) + } + + harness := startHarness(t, api, k8s, actors) + defer harness.stop() + + runStep(ctx, t, api, harness, actors, pool, stepPodcert) + runStep(ctx, t, api, harness, actors, pool, stepAtelet) + + // The new atelet serves the old workers: the actors that kept running + // answer, and a suspended one resumes onto the spare old worker. + router = newRouter(ctx, t) + for _, a := range actors[:2] { + if state(ctx, t, api, a) == ateapipb.ActorState_ACTOR_STATE_RUNNING { + callCounter(ctx, t, router, a, "after "+stepAtelet) + } + } + resumeActor(ctx, t, api, router, actors[3]) + callCounter(ctx, t, router, actors[3], "after "+stepAtelet) + + var onOldWorkers []*actor + for _, a := range actors { + if state(ctx, t, api, a) == ateapipb.ActorState_ACTOR_STATE_RUNNING { + onOldWorkers = append(onOldWorkers, a) + } + } + var newImage string + ok := t.Run(stepPoolRoll, func(t *testing.T) { + harness.setStep(stepPoolRoll) + newImage = publishWorkerImage(t) + if newImage == installed.workerImage { + t.Fatalf("this checkout's worker image %s is the installed one; nothing to upgrade", newImage) + } + rollPool(ctx, t, api.get(), fixture.PoolNamespace, fixture.PoolName, newImage) + // Every old worker is gone, so the harness has suspended these. + for _, a := range onOldWorkers { + waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) + } + assertNoneCrashed(ctx, t, api, actors, stepPoolRoll) + }) + if !ok { + t.FailNow() + } + + // One actor runs through the control plane steps: the old API server + // places it on a new worker and the new one has to suspend it later. + router = newRouter(ctx, t) + resumeActor(ctx, t, api, router, actors[0]) + + runStep(ctx, t, api, harness, actors, pool, stepAPIServer) + templates := workerTemplates(ctx, t, k8s, pool) + runStep(ctx, t, api, harness, actors, pool, stepController) + // A new ate-controller rolls the pool again only if it renders a different + // pod template, which the runbook allows. Wait that out, then put the actor + // back on a worker for the steps that follow. + waitForPoolRoll(ctx, t, api.get(), fixture.PoolNamespace, fixture.PoolName, newImage, time.Minute) + controllerRolled := workerTemplates(ctx, t, k8s, pool) != templates + t.Logf("%s rolled the pool: %t", stepController, controllerRolled) + resumeActor(ctx, t, api, router, actors[0]) + runStep(ctx, t, api, harness, actors, pool, stepAtenet) + runStep(ctx, t, api, harness, actors, pool, stepSandboxConfig) + runStep(ctx, t, api, harness, actors, pool, stepSystem) + harness.stop() + + for _, a := range actors { + evictions := harness.evictions(a) + t.Logf("actor %q: before %v, evicted during %q", a.name, a.before, evictions) + var want []string + for _, old := range onOldWorkers { + if old == a { + want = []string{stepPoolRoll} + } + } + // Only the pool roll may evict an actor, and ate-controller if it + // rolled the pool. + got := evictions + if controllerRolled { + got = slices.DeleteFunc(slices.Clone(evictions), func(step string) bool { return step == stepController }) + } + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("actor %q was evicted during %q, want %q", a.name, evictions, want) + } + } + + // Every actor resumes on a new worker with its memory, and so does one + // created after the upgrade from the old golden snapshot. + fresh := newActor("fresh", startSuspended) + createActor(ctx, t, api, fixture, fresh) + router = newRouter(ctx, t) + for _, a := range append(actors, fresh) { + verifyActor(ctx, t, api, router, k8s, a, newImage) + } + // Once more, so a snapshot the new build wrote restores on the new build. + verifyActor(ctx, t, api, router, k8s, actors[1], newImage) +} + +// runStep runs one ate-setup step and checks the actors afterwards. No step +// but ate-controller may roll the workers; its caller checks that one. +func runStep(ctx context.Context, t *testing.T, api *api, h *harness, actors []*actor, pool types.NamespacedName, step string) { + t.Helper() + ok := t.Run(step, func(t *testing.T) { + h.setStep(step) + deployments := stepDeployments(step) + before := make([]string, len(deployments)) + for i, d := range deployments { + before[i] = deploymentImages(ctx, t, h.k8s, d) + } + var atelet string + if step == stepAtelet { + atelet = daemonSetImage(ctx, t, h.k8s, "atelet") + } + templates := workerTemplates(ctx, t, h.k8s, pool) + ateSetup(t, strings.Fields(step)...) + // deploy ate-system reapplies ate-api-server too. + if step == stepAPIServer || step == stepSystem { + api.reconnect(ctx, t) + } + for i, d := range deployments { + if got := deploymentImages(ctx, t, h.k8s, d); got == before[i] { + t.Errorf("Deployment %s still runs %s after %s", d, got, step) + } + } + if step == stepAtelet { + if got := daemonSetImage(ctx, t, h.k8s, "atelet"); got == atelet { + t.Errorf("atelet still runs %s after %s", got, step) + } + } + if step != stepController { + if got := workerTemplates(ctx, t, h.k8s, pool); got != templates { + t.Errorf("%s rolled the workers: pod templates %s, then %s", step, templates, got) + } + } + assertNoneCrashed(ctx, t, api, actors, step) + }) + if !ok { + t.FailNow() + } +} + +// createActor creates a from the fixture's template, SUSPENDED, and deletes +// it when the test ends. +func createActor(ctx context.Context, t *testing.T, api *api, fixture e2e.SubstrateFixture, a *actor) { + t.Helper() + t.Logf("Creating Actor %q...", a.name) + if _, err := api.get().SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: a.atespace, Name: a.name}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: fixture.Atespace, Name: fixture.Name}, + }}); err != nil { + t.Fatalf("failed to create Actor %q: %v", a.name, err) + } + t.Cleanup(func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + _, _ = api.get().SubstrateAPI.DeleteActor(cleanupCtx, &ateapipb.DeleteActorRequest{Actor: a.ref(), AnyState: true}) + }) + waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) +} + +// seedActor runs a once and leaves it in its before state. +func seedActor(ctx context.Context, t *testing.T, api *api, router *e2e.RouterClient, a *actor) { + t.Helper() + resumeActor(ctx, t, api, router, a) + callCounter(ctx, t, router, a, "before upgrade") + switch a.before { + case startRunning: + case startPaused: + if _, err := api.get().SubstrateAPI.PauseActor(ctx, &ateapipb.PauseActorRequest{Actor: a.ref()}); err != nil { + t.Fatalf("failed to pause Actor %q: %v", a.name, err) + } + waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) + case startSuspended: + suspendActor(ctx, t, api, a) + default: + t.Fatalf("actor %q: unsupported starting state %v", a.name, a.before) + } +} + +// verifyActor has a running on a worker with image, checks the counter +// carried over, and suspends it again to free the worker for the next one. +func verifyActor(ctx context.Context, t *testing.T, api *api, router *e2e.RouterClient, k8s kubernetes.Interface, a *actor, image string) { + t.Helper() + running := resumeActor(ctx, t, api, router, a) + wa := running.GetStatus().GetWorkerAssignment() + pod, err := k8s.CoreV1().Pods(wa.GetWorkerNamespace()).Get(ctx, wa.GetWorkerPod(), metav1.GetOptions{}) + if err != nil { + t.Fatalf("actor %q: failed to get its worker pod %s/%s: %v", a.name, wa.GetWorkerNamespace(), wa.GetWorkerPod(), err) + } + if !podUsesImage(pod, image) { + t.Errorf("actor %q resumed on worker pod %s, which does not run %s", a.name, pod.Name, image) + } + callCounter(ctx, t, router, a, "after upgrade") + suspendActor(ctx, t, api, a) +} + +// resumeActor brings a to RUNNING, or returns it as is when it already is. +func resumeActor(ctx context.Context, t *testing.T, api *api, router *e2e.RouterClient, a *actor) *ateapipb.Actor { + t.Helper() + if act := getActor(ctx, t, api, a); act.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_RUNNING { + return act + } + t.Logf("Resuming Actor %q...", a.name) + if _, err := e2e.ResumeActorAwaitCapacity(t, ctx, api.get(), &ateapipb.ResumeActorRequest{Actor: a.ref()}); err != nil { + t.Fatalf("failed to resume Actor %q: %v", a.name, err) + } + act := waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_RUNNING, stateTimeout) + // Re-sent after every resume; a restore that fell back to a cold boot + // would otherwise start at the 15 s default. + body := routerCallRetry(ctx, t, router, a, http.MethodGet, fmt.Sprintf("/set-sigterm-sleep?duration=%d", sigtermSleepSeconds)) + t.Logf("actor %q: %s", a.name, strings.TrimSpace(body)) + return act +} + +func suspendActor(ctx context.Context, t *testing.T, api *api, a *actor) { + t.Helper() + t.Logf("Suspending Actor %q...", a.name) + if _, err := api.get().SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: a.ref()}); err != nil { + t.Fatalf("failed to suspend Actor %q: %v", a.name, err) + } + waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) +} + +func getActor(ctx context.Context, t *testing.T, api *api, a *actor) *ateapipb.Actor { + t.Helper() + act, err := api.get().SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{Actor: a.ref()}) + if err != nil { + t.Fatalf("failed to get Actor %q: %v", a.name, err) + } + return act +} + +func state(ctx context.Context, t *testing.T, api *api, a *actor) ateapipb.ActorState { + t.Helper() + return getActor(ctx, t, api, a).GetStatus().GetState() +} + +// waitForState polls until a reaches want. A CRASHED actor fails the test at +// once: there is no way back from it. +func waitForState(ctx context.Context, t *testing.T, api *api, a *actor, want ateapipb.ActorState, timeout time.Duration) *ateapipb.Actor { + t.Helper() + deadline := time.Now().Add(timeout) + var last ateapipb.ActorState + for time.Now().Before(deadline) { + act, err := api.get().SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{Actor: a.ref()}) + if err == nil { + last = act.GetStatus().GetState() + if last == want { + return act + } + if last == ateapipb.ActorState_ACTOR_STATE_CRASHED { + t.Fatalf("actor %q is CRASHED (wanted %v)", a.name, want) + } + } + time.Sleep(pollInterval) + } + t.Fatalf("timed out after %v waiting for actor %q to be %v (last %v)", timeout, a.name, want, last) + return nil +} + +func assertNoneCrashed(ctx context.Context, t *testing.T, api *api, actors []*actor, step string) { + t.Helper() + for _, a := range actors { + got := state(ctx, t, api, a) + if got == ateapipb.ActorState_ACTOR_STATE_CRASHED { + t.Fatalf("%s: actor %q is CRASHED", step, a.name) + } + t.Logf("%s: actor %q is %v", step, a.name, got) + } +} + +// callCounter bumps a's counter through the router and checks both counters +// continued from the last call. A retried request may already have counted +// before its response was lost, so the value may run ahead by the retries. +func callCounter(ctx context.Context, t *testing.T, router *e2e.RouterClient, a *actor, stage string) { + t.Helper() + body, attempts := routerCallAttempts(ctx, t, router, a, http.MethodPost, "/") + m := counterResponse.FindStringSubmatch(body) + if m == nil { + t.Fatalf("[%s] actor %q: unexpected response: %s", stage, a.name, body) + } + memory, _ := strconv.Atoi(m[1]) + file, _ := strconv.Atoi(m[2]) + if memory != file || memory <= a.calls || memory > a.calls+attempts { + t.Errorf("[%s] actor %q: counters %d/%d after %d calls and %d attempts: %s", stage, a.name, memory, file, a.calls, attempts, strings.TrimSpace(body)) + } + a.calls = memory +} + +// routerCallRetry calls a through the router, retrying for a minute: the +// router can answer 503 for a while after a resume. +func routerCallRetry(ctx context.Context, t *testing.T, router *e2e.RouterClient, a *actor, method, path string) string { + t.Helper() + body, _ := routerCallAttempts(ctx, t, router, a, method, path) + return body +} + +func routerCallAttempts(ctx context.Context, t *testing.T, router *e2e.RouterClient, a *actor, method, path string) (body string, attempts int) { + t.Helper() + deadline := time.Now().Add(time.Minute) + var lastErr error + for attempts = 1; time.Now().Before(deadline); attempts++ { + body, lastErr = routerCall(ctx, router, a, method, path) + if lastErr == nil { + return body, attempts + } + time.Sleep(pollInterval) + } + t.Fatalf("actor %q: %s %s did not answer: %v", a.name, method, path, lastErr) + return "", attempts +} + +func routerCall(ctx context.Context, router *e2e.RouterClient, a *actor, method, path string) (string, error) { + var resp *http.Response + var err error + if method == http.MethodPost { + resp, err = router.PostJSON(ctx, a.routerRef(), path, nil) + } else { + resp, err = router.Get(ctx, a.routerRef(), path) + } + if err != nil { + return "", err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("status %d: %s", resp.StatusCode, body) + } + return string(body), nil +} + +func newRouter(ctx context.Context, t *testing.T) *e2e.RouterClient { + t.Helper() + router, err := e2e.NewRouterClient(ctx) + if err != nil { + t.Fatalf("failed to connect to atenet-router: %v", err) + } + t.Cleanup(router.Close) + return router +} + +// api holds the ate API client and rebuilds it after the API server rolls, +// which takes the port-forward the client rides on with it. +type api struct { + mu sync.Mutex + c *e2e.Clients + owned bool +} + +func (a *api) get() *e2e.Clients { + a.mu.Lock() + defer a.mu.Unlock() + return a.c +} + +func (a *api) reconnect(ctx context.Context, t *testing.T) { + t.Helper() + fresh, err := e2e.NewClients(ctx) + if err != nil { + t.Fatalf("failed to reconnect to the API server: %v", err) + } + a.mu.Lock() + old, owned := a.c, a.owned + a.c, a.owned = fresh, true + a.mu.Unlock() + if owned { + old.Close() + } +} + +// close releases the client reconnect made; the shared one belongs to +// RunTestMain. +func (a *api) close() { + a.mu.Lock() + defer a.mu.Unlock() + if a.owned { + a.c.Close() + a.owned = false + } +} + +// harness stands in for the actors' operator during the upgrade. It watches +// the RUNNING actors' worker pods and suspends an actor as soon as its pod +// starts terminating, which is what the eviction path asks of every actor, +// and records which upgrade step evicted whom. +type harness struct { + t *testing.T + api *api + k8s kubernetes.Interface + actors []*actor + + mu sync.Mutex + step string + evicted map[*actor][]string + handled map[string]bool + + cancel context.CancelFunc + done chan struct{} + once sync.Once +} + +func startHarness(t *testing.T, api *api, k8s kubernetes.Interface, actors []*actor) *harness { + ctx, cancel := context.WithCancel(context.Background()) + h := &harness{ + t: t, + api: api, + k8s: k8s, + actors: actors, + evicted: make(map[*actor][]string), + handled: make(map[string]bool), + cancel: cancel, + done: make(chan struct{}), + } + go h.run(ctx) + return h +} + +func (h *harness) setStep(name string) { + h.mu.Lock() + defer h.mu.Unlock() + h.step = name +} + +// evictions lists the steps during which a's worker pod was terminated. +func (h *harness) evictions(a *actor) []string { + h.mu.Lock() + defer h.mu.Unlock() + return h.evicted[a] +} + +func (h *harness) stop() { + h.once.Do(func() { + h.cancel() + <-h.done + }) +} + +func (h *harness) run(ctx context.Context) { + defer close(h.done) + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + for _, a := range h.actors { + h.check(ctx, a) + } + } +} + +// check suspends a if its worker pod is terminating. Errors are left for the +// next tick: the API server is rolling for part of the upgrade, and the actor +// keeps running for a day after SIGTERM. +func (h *harness) check(ctx context.Context, a *actor) { + clients := h.api.get() + act, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{Actor: a.ref()}) + if err != nil || act.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_RUNNING { + return + } + wa := act.GetStatus().GetWorkerAssignment() + if wa.GetWorkerPod() == "" { + return + } + pod, err := h.k8s.CoreV1().Pods(wa.GetWorkerNamespace()).Get(ctx, wa.GetWorkerPod(), metav1.GetOptions{}) + if err != nil || pod.DeletionTimestamp == nil { + return + } + h.mu.Lock() + key := a.name + "/" + pod.Name + step, handled := h.step, h.handled[key] + h.mu.Unlock() + if handled { + return + } + h.t.Logf("harness: worker pod %s of actor %q is terminating during %q; suspending", pod.Name, a.name, step) + if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: a.ref()}); err != nil { + h.t.Logf("harness: failed to suspend actor %q: %v", a.name, err) + return + } + h.mu.Lock() + h.handled[key] = true + h.evicted[a] = append(h.evicted[a], step) + h.mu.Unlock() +} + +// install is what the cluster ran before the upgrade. +type install struct { + atelet string + workerImage string +} + +func describeInstall(ctx context.Context, t *testing.T, clients *e2e.Clients, fixture e2e.SubstrateFixture) install { + t.Helper() + dss, err := clients.K8s.AppsV1().DaemonSets(e2e.SystemNamespace()).List(ctx, metav1.ListOptions{LabelSelector: "app=atelet"}) + if err != nil || len(dss.Items) != 1 { + t.Fatalf("expected one atelet DaemonSet, found %d (err %v)", len(dss.Items), err) + } + pool, err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(fixture.PoolNamespace).Get(ctx, fixture.PoolName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get WorkerPool %s/%s (deploy with: %s): %v", fixture.PoolNamespace, fixture.PoolName, fixture.DeployWith, err) + } + return install{ + atelet: dss.Items[0].Spec.Template.Spec.Containers[0].Image, + workerImage: pool.Spec.WorkerImage, + } +} + +func daemonSetImage(ctx context.Context, t *testing.T, k8s kubernetes.Interface, app string) string { + t.Helper() + dss, err := k8s.AppsV1().DaemonSets(e2e.SystemNamespace()).List(ctx, metav1.ListOptions{LabelSelector: "app=" + app}) + if err != nil || len(dss.Items) != 1 { + t.Fatalf("expected one %s DaemonSet, found %d (err %v)", app, len(dss.Items), err) + } + return dss.Items[0].Spec.Template.Spec.Containers[0].Image +} + +// stepDeployments are the Deployments a step has to move to this checkout's +// images. atelet is a DaemonSet, which runStep checks on its own. +func stepDeployments(step string) []types.NamespacedName { + system := e2e.SystemNamespace() + switch step { + case stepPodcert: + return []types.NamespacedName{{Namespace: podcertNamespace, Name: "podcertificate-controller"}} + case stepAPIServer: + return []types.NamespacedName{{Namespace: system, Name: e2e.ResourceName("ate-api-server")}} + case stepController: + return []types.NamespacedName{{Namespace: system, Name: e2e.ResourceName("ate-controller")}} + case stepAtenet: + return []types.NamespacedName{ + {Namespace: system, Name: e2e.ResourceName("atenet-router")}, + {Namespace: system, Name: e2e.ResourceName("atenet-egress")}, + } + } + return nil +} + +// deploymentImages lists the images in a Deployment's pod template. +func deploymentImages(ctx context.Context, t *testing.T, k8s kubernetes.Interface, d types.NamespacedName) string { + t.Helper() + dep, err := k8s.AppsV1().Deployments(d.Namespace).Get(ctx, d.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get Deployment %s: %v", d, err) + } + var images []string + for _, c := range dep.Spec.Template.Spec.Containers { + images = append(images, c.Image) + } + return strings.Join(images, ",") +} + +func podUsesImage(pod *corev1.Pod, image string) bool { + for _, c := range pod.Spec.Containers { + if c.Image == image { + return true + } + } + return false +} + +// ateSetup runs this checkout's ate-setup. +func ateSetup(t *testing.T, args ...string) { + t.Helper() + name, cmd := ateSetupCommand(t, args...) + e2e.RunCmd(t, name, cmd...) +} + +func ateSetupCommand(t *testing.T, args ...string) (name string, cmd []string) { + t.Helper() + root, err := e2e.FindRepoRoot() + if err != nil { + t.Fatalf("FindRepoRoot: %v", err) + } + cmd = []string{"run", filepath.Join(root, "cmd", "ate-setup")} + switch { + case os.Getenv("E2E_ATE_SETUP_CONFIG") != "": + cmd = append(cmd, "--config="+os.Getenv("E2E_ATE_SETUP_CONFIG")) + case os.Getenv("E2E_ATE_SETUP_FLAGS") != "": + cmd = append(cmd, strings.Fields(os.Getenv("E2E_ATE_SETUP_FLAGS"))...) + default: + cmd = append(cmd, "--kind") + } + if e2e.KubeContext != "" { + cmd = append(cmd, "--context="+e2e.KubeContext) + } + // The 60 s default leaves little room on a loaded CI runner. The atelet + // step passes the runbook's timeout, which comes later and wins. + cmd = append(cmd, "--rollout-timeout=5m") + cmd = append(cmd, args...) + return "go", cmd +} + +// publishWorkerImage builds and pushes this checkout's gVisor worker image and +// returns its ref. +func publishWorkerImage(t *testing.T) string { + t.Helper() + const image = "ateworker-gvisor" + name, cmd := ateSetupCommand(t, "publish", "worker-images") + out := e2e.RunCmdOutput(t, nil, name, cmd...) + for line := range strings.Lines(string(out)) { + if ref, ok := strings.CutPrefix(strings.TrimSpace(line), image+": "); ok { + return ref + } + } + t.Fatalf("ate-setup publish worker-images printed no %s ref:\n%s", image, out) + return "" +} + +// rollPool points the pool at image and waits until every worker pod runs it. +func rollPool(ctx context.Context, t *testing.T, clients *e2e.Clients, namespace, name, image string) { + t.Helper() + pools := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(namespace) + pool, err := pools.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get WorkerPool %s/%s: %v", namespace, name, err) + } + t.Logf("WorkerPool %s/%s: workerImage %s -> %s", namespace, name, pool.Spec.WorkerImage, image) + patch, err := json.Marshal(map[string]any{"spec": map[string]any{"workerImage": image}}) + if err != nil { + t.Fatalf("failed to build the WorkerPool patch: %v", err) + } + if _, err := pools.Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + t.Fatalf("failed to patch WorkerPool %s/%s: %v", namespace, name, err) + } + waitForPoolRoll(ctx, t, clients, namespace, name, image, 0) +} + +// waitForPoolRoll waits until every worker pod of the pool runs image from one +// pod template, is ready, and none is terminating, and has stayed that way for +// stable. +func waitForPoolRoll(ctx context.Context, t *testing.T, clients *e2e.Clients, namespace, name, image string, stable time.Duration) { + t.Helper() + pool, err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get WorkerPool %s/%s: %v", namespace, name, err) + } + deadline := time.Now().Add(rollTimeout) + var last string + var rolledSince time.Time + for time.Now().Before(deadline) { + pods, err := clients.K8s.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: "ate.dev/worker-pool=" + name}) + if err == nil { + rolled := len(pods.Items) == int(pool.Spec.Replicas) + templates := map[string]bool{} + var states []string + for i := range pods.Items { + pod := &pods.Items[i] + onImage, ready, terminating := podUsesImage(pod, image), portforward.IsPodReady(pod), pod.DeletionTimestamp != nil + if !onImage || !ready || terminating { + rolled = false + } + templates[pod.Labels["pod-template-hash"]] = true + states = append(states, fmt.Sprintf("%s new=%t ready=%t terminating=%t %s", pod.Name, onImage, ready, terminating, e2e.DescribePodState(pod))) + } + if len(templates) > 1 { + rolled = false + } + last = strings.Join(states, "; ") + switch { + case !rolled: + rolledSince = time.Time{} + case rolledSince.IsZero(): + rolledSince = time.Now() + } + if rolled && time.Since(rolledSince) >= stable { + t.Logf("WorkerPool %s/%s rolled: %s", namespace, name, last) + return + } + } + time.Sleep(pollInterval) + } + t.Fatalf("timed out after %v waiting for WorkerPool %s/%s to roll; pods: %s", rollTimeout, namespace, name, last) +} + +// workerTemplates lists, by pod-template-hash, the pod templates of the pool's +// worker pods that are not terminating. +func workerTemplates(ctx context.Context, t *testing.T, k8s kubernetes.Interface, pool types.NamespacedName) string { + t.Helper() + pods, err := k8s.CoreV1().Pods(pool.Namespace).List(ctx, metav1.ListOptions{LabelSelector: "ate.dev/worker-pool=" + pool.Name}) + if err != nil { + t.Fatalf("failed to list the worker pods of WorkerPool %s: %v", pool, err) + } + var hashes []string + for _, pod := range pods.Items { + if pod.DeletionTimestamp == nil { + hashes = append(hashes, pod.Labels["pod-template-hash"]) + } + } + slices.Sort(hashes) + return strings.Join(slices.Compact(hashes), ",") +} From 6c7303079f8ca6b4c747a7af33555e0868540b6c Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Thu, 15 Oct 2026 21:12:03 -0700 Subject: [PATCH 2/3] e2e: declare no preview gates and pin the old-worker actors CI requires every e2e run to pass --preview, so the upgrade workflow passes it empty: the previous release is installed with no preview gates. Without it the suite fails before running a test. The suite also expected evictions only for the actors that happened to be running after the atelet step. It now requires both running actors to still be running there, and expects the pool roll to evict exactly those two and the actor resumed onto the spare old worker. --- .github/workflows/upgrade-e2e.yaml | 4 +++- internal/e2e/upgrade/upgrade_test.go | 13 +++++-------- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/.github/workflows/upgrade-e2e.yaml b/.github/workflows/upgrade-e2e.yaml index 832880a784..dc19d70941 100644 --- a/.github/workflows/upgrade-e2e.yaml +++ b/.github/workflows/upgrade-e2e.yaml @@ -110,9 +110,11 @@ jobs: env: E2E_ATE_SETUP_CONFIG: ${{ github.workspace }}/new.yaml E2E_JUNIT_FILE: ${{ github.workspace }}/_artifacts/e2e-upgrade.xml + # The previous release is installed with no preview gates, and CI has to + # say so. run: | cp "${ATE_RECORD_DIR}"/installs/*.yaml "${E2E_ATE_SETUP_CONFIG}" - hack/run-e2e-kind.sh ./internal/e2e/upgrade -timeout 60m -v -args --no-color + hack/run-e2e-kind.sh ./internal/e2e/upgrade -timeout 60m -v -args --no-color --preview="" # A suite that stops matching would otherwise leave the run green and empty. - name: Require the suite to have run tests if: always() && steps.from.outputs.tag != '' diff --git a/internal/e2e/upgrade/upgrade_test.go b/internal/e2e/upgrade/upgrade_test.go index 1734a02f83..a3a13bfea8 100644 --- a/internal/e2e/upgrade/upgrade_test.go +++ b/internal/e2e/upgrade/upgrade_test.go @@ -165,19 +165,16 @@ func TestRollingUpgrade(t *testing.T) { // answer, and a suspended one resumes onto the spare old worker. router = newRouter(ctx, t) for _, a := range actors[:2] { - if state(ctx, t, api, a) == ateapipb.ActorState_ACTOR_STATE_RUNNING { - callCounter(ctx, t, router, a, "after "+stepAtelet) + if got := state(ctx, t, api, a); got != ateapipb.ActorState_ACTOR_STATE_RUNNING { + t.Fatalf("actor %q is %v after %s, want it still running", a.name, got, stepAtelet) } + callCounter(ctx, t, router, a, "after "+stepAtelet) } resumeActor(ctx, t, api, router, actors[3]) callCounter(ctx, t, router, actors[3], "after "+stepAtelet) - var onOldWorkers []*actor - for _, a := range actors { - if state(ctx, t, api, a) == ateapipb.ActorState_ACTOR_STATE_RUNNING { - onOldWorkers = append(onOldWorkers, a) - } - } + // These run on old workers, so the pool roll has to evict them. + onOldWorkers := []*actor{actors[0], actors[1], actors[3]} var newImage string ok := t.Run(stepPoolRoll, func(t *testing.T) { harness.setStep(stepPoolRoll) From 15c7829468b2d17a02537849b69292de09b60442 Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Sun, 18 Oct 2026 19:55:15 -0700 Subject: [PATCH 3/3] e2e: fail the upgrade if ate-controller rolls the pool, and suspend by mode The skew policy keeps the worker pod template the same within a major, so the suite now fails if the new ate-controller rolls the pool instead of allowing the roll and its evictions. PauseActor is gone and SuspendActor defaults to an asynchronous suspend. The paused actor is now suspended asynchronously, and the suspended actor and the final checks suspend synchronously. The imports move to the go.ate.dev/substrate module path. --- internal/e2e/upgrade/testmain_test.go | 2 +- internal/e2e/upgrade/upgrade_test.go | 56 +++++++++++++-------------- 2 files changed, 27 insertions(+), 31 deletions(-) diff --git a/internal/e2e/upgrade/testmain_test.go b/internal/e2e/upgrade/testmain_test.go index ecc2b4a129..85285a2524 100644 --- a/internal/e2e/upgrade/testmain_test.go +++ b/internal/e2e/upgrade/testmain_test.go @@ -18,7 +18,7 @@ import ( "os" "testing" - "github.com/agent-substrate/substrate/internal/e2e" + "go.ate.dev/substrate/internal/e2e" ) func TestMain(m *testing.M) { os.Exit(e2e.RunTestMain(m)) } diff --git a/internal/e2e/upgrade/upgrade_test.go b/internal/e2e/upgrade/upgrade_test.go index a3a13bfea8..135b7ec2fe 100644 --- a/internal/e2e/upgrade/upgrade_test.go +++ b/internal/e2e/upgrade/upgrade_test.go @@ -59,10 +59,10 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" - "github.com/agent-substrate/substrate/internal/e2e" - "github.com/agent-substrate/substrate/internal/portforward" - "github.com/agent-substrate/substrate/internal/resources" - "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "go.ate.dev/substrate/internal/e2e" + "go.ate.dev/substrate/internal/portforward" + "go.ate.dev/substrate/internal/resources" + "go.ate.dev/substrate/pkg/proto/ateapipb" ) const ( @@ -98,7 +98,9 @@ const podcertNamespace = "podcertificate-controller-system" var counterResponse = regexp.MustCompile(`preserved memory count: (\d+) \| preserved file counter: (\d+)`) // startState is how seedActor leaves an actor ahead of the upgrade. A paused -// actor is SUSPENDED with a node-local snapshot. +// actor is suspended asynchronously: SUSPENDED, with its snapshot on the node +// until atelet uploads it in the background. A suspended one waits for the +// upload. type startState string const ( @@ -201,13 +203,13 @@ func TestRollingUpgrade(t *testing.T) { runStep(ctx, t, api, harness, actors, pool, stepAPIServer) templates := workerTemplates(ctx, t, k8s, pool) runStep(ctx, t, api, harness, actors, pool, stepController) - // A new ate-controller rolls the pool again only if it renders a different - // pod template, which the runbook allows. Wait that out, then put the actor - // back on a worker for the steps that follow. + // The worker pod template stays the same within a major, so the new + // ate-controller must not roll the pool. It would start only once the + // controller reconciles, so give it a minute. waitForPoolRoll(ctx, t, api.get(), fixture.PoolNamespace, fixture.PoolName, newImage, time.Minute) - controllerRolled := workerTemplates(ctx, t, k8s, pool) != templates - t.Logf("%s rolled the pool: %t", stepController, controllerRolled) - resumeActor(ctx, t, api, router, actors[0]) + if got := workerTemplates(ctx, t, k8s, pool); got != templates { + t.Fatalf("%s rolled the pool: pod templates %s, then %s", stepController, templates, got) + } runStep(ctx, t, api, harness, actors, pool, stepAtenet) runStep(ctx, t, api, harness, actors, pool, stepSandboxConfig) runStep(ctx, t, api, harness, actors, pool, stepSystem) @@ -222,13 +224,8 @@ func TestRollingUpgrade(t *testing.T) { want = []string{stepPoolRoll} } } - // Only the pool roll may evict an actor, and ate-controller if it - // rolled the pool. - got := evictions - if controllerRolled { - got = slices.DeleteFunc(slices.Clone(evictions), func(step string) bool { return step == stepController }) - } - if strings.Join(got, ",") != strings.Join(want, ",") { + // Only the pool roll may evict an actor. + if strings.Join(evictions, ",") != strings.Join(want, ",") { t.Errorf("actor %q was evicted during %q, want %q", a.name, evictions, want) } } @@ -245,8 +242,9 @@ func TestRollingUpgrade(t *testing.T) { verifyActor(ctx, t, api, router, k8s, actors[1], newImage) } -// runStep runs one ate-setup step and checks the actors afterwards. No step -// but ate-controller may roll the workers; its caller checks that one. +// runStep runs one ate-setup step and checks the actors afterwards. No step may +// roll the workers. ate-controller would roll them only once it reconciles, so +// its caller checks that step. func runStep(ctx context.Context, t *testing.T, api *api, h *harness, actors []*actor, pool types.NamespacedName, step string) { t.Helper() ok := t.Run(step, func(t *testing.T) { @@ -302,7 +300,7 @@ func createActor(ctx context.Context, t *testing.T, api *api, fixture e2e.Substr t.Cleanup(func() { cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Minute) defer cancel() - _, _ = api.get().SubstrateAPI.DeleteActor(cleanupCtx, &ateapipb.DeleteActorRequest{Actor: a.ref(), AnyState: true}) + _, _ = api.get().SubstrateAPI.DeleteActor(cleanupCtx, &ateapipb.DeleteActorRequest{Actor: a.ref()}) }) waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) } @@ -315,12 +313,9 @@ func seedActor(ctx context.Context, t *testing.T, api *api, router *e2e.RouterCl switch a.before { case startRunning: case startPaused: - if _, err := api.get().SubstrateAPI.PauseActor(ctx, &ateapipb.PauseActorRequest{Actor: a.ref()}); err != nil { - t.Fatalf("failed to pause Actor %q: %v", a.name, err) - } - waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout) + suspendActor(ctx, t, api, a, ateapipb.DurabilityMode_DURABILITY_MODE_ASYNCHRONOUS) case startSuspended: - suspendActor(ctx, t, api, a) + suspendActor(ctx, t, api, a, ateapipb.DurabilityMode_DURABILITY_MODE_SYNCHRONOUS) default: t.Fatalf("actor %q: unsupported starting state %v", a.name, a.before) } @@ -340,7 +335,7 @@ func verifyActor(ctx context.Context, t *testing.T, api *api, router *e2e.Router t.Errorf("actor %q resumed on worker pod %s, which does not run %s", a.name, pod.Name, image) } callCounter(ctx, t, router, a, "after upgrade") - suspendActor(ctx, t, api, a) + suspendActor(ctx, t, api, a, ateapipb.DurabilityMode_DURABILITY_MODE_SYNCHRONOUS) } // resumeActor brings a to RUNNING, or returns it as is when it already is. @@ -361,10 +356,11 @@ func resumeActor(ctx context.Context, t *testing.T, api *api, router *e2e.Router return act } -func suspendActor(ctx context.Context, t *testing.T, api *api, a *actor) { +func suspendActor(ctx context.Context, t *testing.T, api *api, a *actor, mode ateapipb.DurabilityMode) { t.Helper() - t.Logf("Suspending Actor %q...", a.name) - if _, err := api.get().SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: a.ref()}); err != nil { + t.Logf("Suspending Actor %q (%v)...", a.name, mode) + req := &ateapipb.SuspendActorRequest{Actor: a.ref(), Options: &ateapipb.SuspendOptions{Mode: mode}} + if _, err := api.get().SubstrateAPI.SuspendActor(ctx, req); err != nil { t.Fatalf("failed to suspend Actor %q: %v", a.name, err) } waitForState(ctx, t, api, a, ateapipb.ActorState_ACTOR_STATE_SUSPENDED, stateTimeout)