From 9c9cc51a9b7e22daf19dc2e3f14fa1622d2160e3 Mon Sep 17 00:00:00 2001 From: Nikola Jokic Date: Thu, 10 Sep 2026 18:32:39 +0200 Subject: [PATCH] Guard the listener's Running phase write with resourceVersion HandleJobStarted reads the ephemeral runner to decide whether the runner is still eligible for the Running phase, then applies that decision with a separate merge patch. The patch carried no precondition, so nothing tied it to the state the decision was made on: if the ephemeral runner controller wrote a terminal phase in between, the listener silently overwrote it and resurrected a runner that had already finished. Attach the observed resourceVersion to the patch whenever the phase is transitioned, and retry on conflict so the decision is re-made against the fresh state. The job detail fields stay unguarded: they are write-once metadata that the runner set only consults for runners that are neither done nor being deleted, so patching them cannot change a scaling decision. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- cmd/ghalistener/scaler/scaler.go | 22 ++- .../scaler/scaler_apiserver_test.go | 158 ++++++++++++++++ cmd/ghalistener/scaler/scaler_test.go | 168 +++++++++++++++++- 3 files changed, 345 insertions(+), 3 deletions(-) create mode 100644 cmd/ghalistener/scaler/scaler_apiserver_test.go diff --git a/cmd/ghalistener/scaler/scaler.go b/cmd/ghalistener/scaler/scaler.go index 94c87f2b..235a8422 100644 --- a/cmd/ghalistener/scaler/scaler.go +++ b/cmd/ghalistener/scaler/scaler.go @@ -15,6 +15,7 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" + "k8s.io/client-go/util/retry" ) type Option func(*Scaler) @@ -138,6 +139,16 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar w.dirty = true + // The phase transition below is decided from the runner state observed by a + // read, so the write has to be guarded against a concurrent update by the + // ephemeral runner controller. On conflict the runner is re-read and the + // decision is made again against the fresh state. + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + return w.patchJobStarted(ctx, jobInfo) + }) +} + +func (w *Scaler) patchJobStarted(ctx context.Context, jobInfo *scaleset.JobStarted) error { // Fetch current EphemeralRunner to check phase and deletion status currentRunner := &v1alpha1.EphemeralRunner{} err := w.clientset.RESTClient(). @@ -173,12 +184,21 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar }, } - // Only set Running phase if current phase is not terminal/failure and deletion is not in progress + // Only set Running phase if current phase is not terminal/failure and deletion is not in progress. + // + // The phase is the only field derived from the state read above, so the observed + // resourceVersion is attached to the patch as a precondition. Without it, a terminal + // phase written between the read and the patch would be silently overwritten with + // Running, resurrecting a runner that already finished. The job fields carry no such + // precondition: they are write-once metadata that the runner set only consults for + // runners that are neither done nor being deleted, so patching them unconditionally + // cannot change any scaling decision. if currentRunner.DeletionTimestamp == nil && currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseFailed && currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseSucceeded && currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseOutdated { patchRunner.Status.Phase = v1alpha1.EphemeralRunnerPhaseRunning + patchRunner.ResourceVersion = currentRunner.ResourceVersion } patch, err := json.Marshal(patchRunner) diff --git a/cmd/ghalistener/scaler/scaler_apiserver_test.go b/cmd/ghalistener/scaler/scaler_apiserver_test.go new file mode 100644 index 00000000..96b6771f --- /dev/null +++ b/cmd/ghalistener/scaler/scaler_apiserver_test.go @@ -0,0 +1,158 @@ +package scaler + +import ( + "context" + "net/http" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1" + "github.com/actions/scaleset" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" +) + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +// TestHandleJobStartedAgainstAPIServer exercises HandleJobStarted against a real +// API server. The unit tests above emulate the optimistic concurrency check that +// kube-apiserver performs when a merge patch carries metadata.resourceVersion; +// this test pins that emulation to the real behaviour. +// +// The race is made deterministic by writing the terminal phase from inside the +// client transport, right before the scaler's patch reaches the API server. +func TestHandleJobStartedAgainstAPIServer(t *testing.T) { + if os.Getenv("KUBEBUILDER_ASSETS") == "" { + t.Skip("KUBEBUILDER_ASSETS is not set; run via `make test`") + } + + env := &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "config", "crd", "bases")}, + ErrorIfCRDPathMissing: true, + } + cfg, err := env.Start() + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, env.Stop()) + }) + + sch := runtime.NewScheme() + require.NoError(t, scheme.AddToScheme(sch)) + require.NoError(t, v1alpha1.AddToScheme(sch)) + k8sClient, err := client.New(cfg, client.Options{Scheme: sch}) + require.NoError(t, err) + + ctx := context.Background() + + jobInfo := &scaleset.JobStarted{ + RunnerName: "runner-1", + JobMessageBase: scaleset.JobMessageBase{ + OwnerName: "actions", + RepositoryName: "actions-runner-controller", + JobID: "job-1", + WorkflowRunID: 456, + JobWorkflowRef: "actions/actions-runner-controller/.github/workflows/ci.yaml@refs/heads/main", + JobDisplayName: "build", + RunnerRequestID: 123, + }, + } + + newRunner := func(t *testing.T, name string) *v1alpha1.EphemeralRunner { + t.Helper() + + runner := &v1alpha1.EphemeralRunner{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: v1alpha1.EphemeralRunnerSpec{ + GitHubConfigURL: "https://github.com/actions", + GitHubConfigSecret: "secret", + RunnerScaleSetID: 1, + PodTemplateSpec: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "runner", Image: "ghcr.io/actions/runner"}}, + }, + }, + }, + } + require.NoError(t, k8sClient.Create(ctx, runner)) + return runner + } + + newScaler := func(t *testing.T, beforePatch func()) *Scaler { + t.Helper() + + conf := rest.CopyConfig(cfg) + if beforePatch != nil { + var once sync.Once + conf.Wrap(func(rt http.RoundTripper) http.RoundTripper { + return roundTripperFunc(func(req *http.Request) (*http.Response, error) { + if req.Method == http.MethodPatch { + once.Do(beforePatch) + } + return rt.RoundTrip(req) + }) + }) + } + + clientset, err := kubernetes.NewForConfig(conf) + require.NoError(t, err) + + return &Scaler{ + clientset: clientset, + config: Config{EphemeralRunnerSetNamespace: "default"}, + targetRunners: -1, + patchSeq: -1, + logger: discardLogger, + } + } + + t.Run("transitions an idle runner to Running", func(t *testing.T) { + runner := newRunner(t, "runner-running") + jobInfo := *jobInfo + jobInfo.RunnerName = runner.Name + + require.NoError(t, newScaler(t, nil).HandleJobStarted(ctx, &jobInfo)) + + require.NoError(t, k8sClient.Get(ctx, client.ObjectKeyFromObject(runner), runner)) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseRunning, runner.Status.Phase) + assert.Equal(t, jobInfo.JobID, runner.Status.JobID) + }) + + t.Run("does not resurrect a runner that failed after the read", func(t *testing.T) { + runner := newRunner(t, "runner-raced") + jobInfo := *jobInfo + jobInfo.RunnerName = runner.Name + + scaler := newScaler(t, func() { + failed := runner.DeepCopy() + failed.Status.Phase = v1alpha1.EphemeralRunnerPhaseFailed + require.NoError(t, k8sClient.Status().Patch(ctx, failed, client.MergeFrom(runner))) + }) + + require.NoError(t, scaler.HandleJobStarted(ctx, &jobInfo)) + + require.NoError(t, k8sClient.Get(ctx, client.ObjectKeyFromObject(runner), runner)) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseFailed, runner.Status.Phase) + assert.Equal(t, jobInfo.JobID, runner.Status.JobID, "job details are still recorded") + }) + + t.Run("ignores a runner that no longer exists", func(t *testing.T) { + jobInfo := *jobInfo + jobInfo.RunnerName = "runner-missing" + + assert.NoError(t, newScaler(t, nil).HandleJobStarted(ctx, &jobInfo)) + }) +} diff --git a/cmd/ghalistener/scaler/scaler_test.go b/cmd/ghalistener/scaler/scaler_test.go index 14d01b45..eabd1c98 100644 --- a/cmd/ghalistener/scaler/scaler_test.go +++ b/cmd/ghalistener/scaler/scaler_test.go @@ -4,20 +4,25 @@ import ( "bytes" "context" "encoding/json" + "fmt" + "io" "log/slog" "math" "net/http" "net/http/httptest" "strconv" + "sync" "testing" "github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1" "github.com/actions/scaleset" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" + "k8s.io/client-go/util/retry" ) var discardLogger = slog.New(slog.DiscardHandler) @@ -198,6 +203,77 @@ func TestHandleJobStarted(t *testing.T) { assertJobStartedStatus(t, runner, jobInfo) assert.Equal(t, v1alpha1.EphemeralRunnerPhasePending, runner.Status.Phase) }) + + t.Run("guards the phase transition with the observed resourceVersion", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhaseRunning) + scaler, api, shutdown := newTestScalerWithAPI(t, runner, nil) + defer shutdown() + + observedResourceVersion := runner.ResourceVersion + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + meta, ok := api.patch()["metadata"].(map[string]any) + require.True(t, ok, "patch must carry a metadata precondition, got %v", api.patch()) + assert.Equal(t, observedResourceVersion, meta["resourceVersion"]) + }) + + t.Run("omits the precondition when the phase is not transitioned", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhaseFailed) + scaler, api, shutdown := newTestScalerWithAPI(t, runner, nil) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + _, hasMetadata := api.patch()["metadata"] + assert.False(t, hasMetadata, "job fields alone must not be guarded, got %v", api.patch()) + }) + + // A runner can reach a terminal phase between the read that decides the + // transition and the patch that applies it. Without a precondition the + // listener would resurrect it back into Running. + for _, phase := range []v1alpha1.EphemeralRunnerPhase{ + v1alpha1.EphemeralRunnerPhaseFailed, + v1alpha1.EphemeralRunnerPhaseSucceeded, + v1alpha1.EphemeralRunnerPhaseOutdated, + } { + t.Run("does not resurrect a runner that became "+string(phase)+" concurrently", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + + var raced bool + scaler, api, shutdown := newTestScalerWithAPI(t, runner, func(runner *v1alpha1.EphemeralRunner) { + if raced { + return + } + raced = true + runner.Status.Phase = phase + }) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + assert.Equal(t, phase, runner.Status.Phase) + assertJobStartedStatus(t, runner, jobInfo) + + gets, patches, conflicts := api.counts() + assert.Equal(t, 1, conflicts, "the stale patch must be rejected") + assert.Equal(t, 2, gets, "the runner must be re-read after the conflict") + assert.Equal(t, 2, patches) + }) + } + + t.Run("gives up when the runner keeps changing", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + scaler, api, shutdown := newTestScalerWithAPI(t, runner, func(runner *v1alpha1.EphemeralRunner) {}) + defer shutdown() + + err := scaler.HandleJobStarted(context.Background(), jobInfo) + require.Error(t, err) + assert.True(t, kerrors.IsConflict(err), "expected a conflict error, got %v", err) + + _, _, conflicts := api.counts() + assert.Equal(t, retry.DefaultRetry.Steps, conflicts) + assert.Equal(t, v1alpha1.EphemeralRunnerPhasePending, runner.Status.Phase) + }) } func newTestEphemeralRunner(name string, phase v1alpha1.EphemeralRunnerPhase) *v1alpha1.EphemeralRunner { @@ -215,15 +291,89 @@ func newTestEphemeralRunner(name string, phase v1alpha1.EphemeralRunnerPhase) *v func newTestScaler(t *testing.T, runner *v1alpha1.EphemeralRunner) (*Scaler, func()) { t.Helper() + scaler, _, shutdown := newTestScalerWithAPI(t, runner, nil) + return scaler, shutdown +} + +// fakeRunnerAPI records what the scaler sent to the API server. +type fakeRunnerAPI struct { + mu sync.Mutex + gets int + patches int + conflicts int + lastPatch map[string]any + patchedRVs []string +} + +func (f *fakeRunnerAPI) counts() (gets, patches, conflicts int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.gets, f.patches, f.conflicts +} + +func (f *fakeRunnerAPI) patch() map[string]any { + f.mu.Lock() + defer f.mu.Unlock() + return f.lastPatch +} + +// newTestScalerWithAPI serves runner over a fake API server that emulates the +// resourceVersion precondition enforced by kube-apiserver on merge patches. +// afterGet, when set, runs after every read is served and simulates another +// writer mutating the runner before the scaler's patch lands. +func newTestScalerWithAPI(t *testing.T, runner *v1alpha1.EphemeralRunner, afterGet func(runner *v1alpha1.EphemeralRunner)) (*Scaler, *fakeRunnerAPI, func()) { + t.Helper() + + api := &fakeRunnerAPI{} + if runner.ResourceVersion == "" { + runner.ResourceVersion = "1" + } + + bumpResourceVersion := func() { + rv, err := strconv.Atoi(runner.ResourceVersion) + require.NoError(t, err) + runner.ResourceVersion = strconv.Itoa(rv + 1) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + api.mu.Lock() + defer api.mu.Unlock() + w.Header().Set("Content-Type", "application/json") switch r.Method { case http.MethodGet: + api.gets++ require.NoError(t, json.NewEncoder(w).Encode(runner)) + if afterGet != nil { + afterGet(runner) + bumpResourceVersion() + } case http.MethodPatch: + api.patches++ + + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + + var raw map[string]any + require.NoError(t, json.Unmarshal(body, &raw)) + api.lastPatch = raw + + // Emulate the optimistic concurrency check performed by the API + // server when metadata.resourceVersion is present in the patch. + if meta, ok := raw["metadata"].(map[string]any); ok { + if rv, ok := meta["resourceVersion"].(string); ok { + api.patchedRVs = append(api.patchedRVs, rv) + if rv != runner.ResourceVersion { + api.conflicts++ + writeConflict(t, w, runner.Name) + return + } + } + } + var patch v1alpha1.EphemeralRunner - require.NoError(t, json.NewDecoder(r.Body).Decode(&patch)) + require.NoError(t, json.Unmarshal(body, &patch)) runner.Status.JobRequestID = patch.Status.JobRequestID runner.Status.JobRepositoryName = patch.Status.JobRepositoryName @@ -234,6 +384,7 @@ func newTestScaler(t *testing.T, runner *v1alpha1.EphemeralRunner) (*Scaler, fun if patch.Status.Phase != "" { runner.Status.Phase = patch.Status.Phase } + bumpResourceVersion() require.NoError(t, json.NewEncoder(w).Encode(runner)) default: @@ -252,7 +403,20 @@ func newTestScaler(t *testing.T, runner *v1alpha1.EphemeralRunner) (*Scaler, fun targetRunners: -1, patchSeq: -1, logger: discardLogger, - }, server.Close + }, api, server.Close +} + +func writeConflict(t *testing.T, w http.ResponseWriter, name string) { + t.Helper() + + w.WriteHeader(http.StatusConflict) + require.NoError(t, json.NewEncoder(w).Encode(&metav1.Status{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Status"}, + Status: metav1.StatusFailure, + Code: http.StatusConflict, + Reason: metav1.StatusReasonConflict, + Message: fmt.Sprintf("Operation cannot be fulfilled on ephemeralrunners.actions.github.com %q: the object has been modified", name), + })) } func assertJobStartedStatus(t *testing.T, runner *v1alpha1.EphemeralRunner, jobInfo *scaleset.JobStarted) {