mirror of
https://github.com/actions-runner-controller/actions-runner-controller.git
synced 2026-09-30 10:11:39 +02:00
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>
This commit is contained in:
Nikola Jokic
co-authored by
Copilot App
Copilot App
parent
e2768dddbf
commit
9c9cc51a9b
@@ -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)
|
||||
|
||||
@@ -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))
|
||||
})
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user