diff --git a/apis/actions.github.com/v1alpha1/ephemeralrunner_types.go b/apis/actions.github.com/v1alpha1/ephemeralrunner_types.go index 4c4a2acefd..f2903b511e 100644 --- a/apis/actions.github.com/v1alpha1/ephemeralrunner_types.go +++ b/apis/actions.github.com/v1alpha1/ephemeralrunner_types.go @@ -146,6 +146,10 @@ type EphemeralRunnerStatus struct { // // The PodSucceded phase should be set only when confirmed that EphemeralRunner // actually executed the job and has been removed from the service. + // + // The Running phase is owned by the listener and is set only when a job has + // been assigned to this EphemeralRunner. It does not mean the runner is merely + // online and waiting for work; an idle registered runner stays Pending. // +optional Phase EphemeralRunnerPhase `json:"phase,omitempty"` // +optional @@ -185,11 +189,12 @@ type EphemeralRunnerStatus struct { type EphemeralRunnerPhase string const ( - // EphemeralRunnerPhasePending is a phase set when the ephemeral runner is - // being provisioned and is not yet online. + // EphemeralRunnerPhasePending is a phase set while no job has been assigned to + // the ephemeral runner. It covers both a runner that is still being provisioned + // and one that is already online and registered but idle. EphemeralRunnerPhasePending EphemeralRunnerPhase = "Pending" - // EphemeralRunnerPhaseRunning is a phase set when the ephemeral runner is online and - // waiting for a job to execute. + // EphemeralRunnerPhaseRunning is a phase set by the listener when a job has been + // assigned to this ephemeral runner and the runner is executing it. EphemeralRunnerPhaseRunning EphemeralRunnerPhase = "Running" // EphemeralRunnerPhaseSucceeded is a phase set when the ephemeral runner // successfully executed the job and has been removed from the service. diff --git a/charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_ephemeralrunners.yaml b/charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_ephemeralrunners.yaml index a174d751d4..e6d671a139 100644 --- a/charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_ephemeralrunners.yaml +++ b/charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_ephemeralrunners.yaml @@ -8323,6 +8323,10 @@ spec: The PodSucceded phase should be set only when confirmed that EphemeralRunner actually executed the job and has been removed from the service. + + The Running phase is owned by the listener and is set only when a job has + been assigned to this EphemeralRunner. It does not mean the runner is merely + online and waiting for work; an idle registered runner stays Pending. type: string ready: description: Turns true only if the runner is online. diff --git a/charts/gha-runner-scale-set-controller/crds/actions.github.com_ephemeralrunners.yaml b/charts/gha-runner-scale-set-controller/crds/actions.github.com_ephemeralrunners.yaml index a174d751d4..e6d671a139 100644 --- a/charts/gha-runner-scale-set-controller/crds/actions.github.com_ephemeralrunners.yaml +++ b/charts/gha-runner-scale-set-controller/crds/actions.github.com_ephemeralrunners.yaml @@ -8323,6 +8323,10 @@ spec: The PodSucceded phase should be set only when confirmed that EphemeralRunner actually executed the job and has been removed from the service. + + The Running phase is owned by the listener and is set only when a job has + been assigned to this EphemeralRunner. It does not mean the runner is merely + online and waiting for work; an idle registered runner stays Pending. type: string ready: description: Turns true only if the runner is online. diff --git a/cmd/ghalistener/scaler/scaler.go b/cmd/ghalistener/scaler/scaler.go index 7c486f54f4..cf7dee3cda 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) @@ -123,6 +124,7 @@ func (w *Scaler) applyDefaults() error { // It takes a context and a jobInfo parameter which contains the details of the started job. // This update marks the ephemeral runner so that the controller would have more context // about the ephemeral runner that should not be deleted when scaling down. +// It also transitions the phase to Running if the runner is not in a terminal state. // It returns an error if there is any issue with updating the job information. func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStarted) error { w.logger.Info("Updating job info for the runner", @@ -137,23 +139,61 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar w.dirty = true + // The promotion to Running is guarded by an optimistic lock on the resource version + // observed by the GET below, so a terminal phase written between the read and the + // patch is never clobbered. Conflicts are retried against freshly read 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(). + Get(). + Prefix("apis", v1alpha1.GroupVersion.Group, v1alpha1.GroupVersion.Version). + Namespace(w.config.EphemeralRunnerSetNamespace). + Resource("ephemeralrunners"). + Name(jobInfo.RunnerName). + Do(ctx). + Into(currentRunner) + if err != nil { + if kerrors.IsNotFound(err) { + w.logger.Info("Ephemeral runner not found, skipping job info update", "runnerName", jobInfo.RunnerName) + return nil + } + return fmt.Errorf("failed to get ephemeral runner: %w", err) + } + original, err := json.Marshal(&v1alpha1.EphemeralRunner{}) if err != nil { return fmt.Errorf("failed to marshal empty ephemeral runner: %w", err) } - patch, err := json.Marshal( - &v1alpha1.EphemeralRunner{ - Status: v1alpha1.EphemeralRunnerStatus{ - JobRequestID: jobInfo.RunnerRequestID, - JobRepositoryName: fmt.Sprintf("%s/%s", jobInfo.OwnerName, jobInfo.RepositoryName), - JobID: jobInfo.JobID, - WorkflowRunID: jobInfo.WorkflowRunID, - JobWorkflowRef: jobInfo.JobWorkflowRef, - JobDisplayName: jobInfo.JobDisplayName, - }, + // Build patch with job fields + patchRunner := &v1alpha1.EphemeralRunner{ + Status: v1alpha1.EphemeralRunnerStatus{ + JobRequestID: jobInfo.RunnerRequestID, + JobRepositoryName: fmt.Sprintf("%s/%s", jobInfo.OwnerName, jobInfo.RepositoryName), + JobID: jobInfo.JobID, + WorkflowRunID: jobInfo.WorkflowRunID, + JobWorkflowRef: jobInfo.JobWorkflowRef, + JobDisplayName: jobInfo.JobDisplayName, }, - ) + } + + // Only set Running phase if current phase is not terminal/failure and deletion is not in progress + if currentRunner.DeletionTimestamp == nil && + currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseFailed && + currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseSucceeded && + currentRunner.Status.Phase != v1alpha1.EphemeralRunnerPhaseOutdated { + patchRunner.Status.Phase = v1alpha1.EphemeralRunnerPhaseRunning + // Optimistic lock: reject the promotion if the runner changed since the GET. + patchRunner.ResourceVersion = currentRunner.ResourceVersion + } + + patch, err := json.Marshal(patchRunner) if err != nil { return fmt.Errorf("failed to marshal ephemeral runner patch: %w", err) } @@ -170,7 +210,7 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar Patch(types.MergePatchType). Prefix("apis", v1alpha1.GroupVersion.Group, v1alpha1.GroupVersion.Version). Namespace(w.config.EphemeralRunnerSetNamespace). - Resource("EphemeralRunners"). + Resource("ephemeralrunners"). Name(jobInfo.RunnerName). SubResource("status"). Body(mergePatch). @@ -181,6 +221,10 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar w.logger.Info("Ephemeral runner not found, skipping patching of ephemeral runner status", "runnerName", jobInfo.RunnerName) return nil } + if kerrors.IsConflict(err) { + w.logger.Info("Ephemeral runner changed while patching job info, retrying", "runnerName", jobInfo.RunnerName) + return err + } return fmt.Errorf("could not patch ephemeral runner status, patch JSON: %s, error: %w", string(mergePatch), err) } diff --git a/cmd/ghalistener/scaler/scaler_test.go b/cmd/ghalistener/scaler/scaler_test.go index 2bf3105cd9..4de7812d1e 100644 --- a/cmd/ghalistener/scaler/scaler_test.go +++ b/cmd/ghalistener/scaler/scaler_test.go @@ -2,13 +2,23 @@ package scaler import ( "bytes" + "context" + "encoding/json" + "fmt" "log/slog" "math" + "net/http" + "net/http/httptest" "strconv" "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" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" ) var discardLogger = slog.New(slog.DiscardHandler) @@ -121,6 +131,196 @@ func TestEffectiveRateLimiterConfig_QuietAtInfoLevel(t *testing.T) { } } +func TestHandleJobStarted(t *testing.T) { + 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, + }, + } + + t.Run("patches job fields and running phase together", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, "") + scaler, shutdown := newTestScaler(t, runner) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + assertJobStartedStatus(t, runner, jobInfo) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseRunning, runner.Status.Phase) + }) + + t.Run("repeated assignment remains idempotent", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhaseRunning) + scaler, shutdown := newTestScaler(t, runner) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + firstStatus := runner.Status + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + assert.Equal(t, firstStatus, runner.Status) + assertJobStartedStatus(t, runner, jobInfo) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseRunning, runner.Status.Phase) + }) + + for _, phase := range []v1alpha1.EphemeralRunnerPhase{ + v1alpha1.EphemeralRunnerPhaseFailed, + v1alpha1.EphemeralRunnerPhaseSucceeded, + v1alpha1.EphemeralRunnerPhaseOutdated, + } { + t.Run("preserves "+string(phase)+" phase while patching job fields", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, phase) + scaler, shutdown := newTestScaler(t, runner) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + assertJobStartedStatus(t, runner, jobInfo) + assert.Equal(t, phase, runner.Status.Phase) + }) + } + + t.Run("retries against fresh state when a terminal write wins the race", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + // A terminal update lands between the scaler's GET and its first patch, so + // the patch carries a stale resource version and is rejected with 409. + raceTerminalWrite := func() { + runner.Status.Phase = v1alpha1.EphemeralRunnerPhaseFailed + runner.ResourceVersion = strconv.Itoa(mustAtoi(t, runner.ResourceVersion) + 1) + } + scaler, shutdown := newTestScaler(t, runner, raceTerminalWrite) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + // The retry re-reads the now-terminal runner, so the job fields are recorded + // while the promotion to Running is abandoned rather than clobbering Failed. + assertJobStartedStatus(t, runner, jobInfo) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseFailed, runner.Status.Phase) + }) + + t.Run("preserves deleting runner phase while patching job fields", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + deletionTimestamp := metav1.Now() + runner.DeletionTimestamp = &deletionTimestamp + scaler, shutdown := newTestScaler(t, runner) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + assertJobStartedStatus(t, runner, jobInfo) + assert.Equal(t, v1alpha1.EphemeralRunnerPhasePending, runner.Status.Phase) + }) +} + +func newTestEphemeralRunner(name string, phase v1alpha1.EphemeralRunnerPhase) *v1alpha1.EphemeralRunner { + return &v1alpha1.EphemeralRunner{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + ResourceVersion: "1", + }, + Status: v1alpha1.EphemeralRunnerStatus{ + Phase: phase, + }, + } +} + +// newTestScaler serves the runner over a stub API server that enforces the +// metadata.resourceVersion precondition the way the API server does, so that a +// patch carrying a stale resource version is rejected with 409 Conflict. +// Each onPatch hook runs before the corresponding patch is applied, which lets a +// test interleave a competing write between the scaler's GET and its patch. +func newTestScaler(t *testing.T, runner *v1alpha1.EphemeralRunner, onPatch ...func()) (*Scaler, func()) { + t.Helper() + + var patches int + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + + switch r.Method { + case http.MethodGet: + require.NoError(t, json.NewEncoder(w).Encode(runner)) + case http.MethodPatch: + var patch v1alpha1.EphemeralRunner + require.NoError(t, json.NewDecoder(r.Body).Decode(&patch)) + + if patches < len(onPatch) { + onPatch[patches]() + } + patches++ + + if patch.ResourceVersion != "" && patch.ResourceVersion != runner.ResourceVersion { + 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", + runner.Name), + })) + return + } + + runner.Status.JobRequestID = patch.Status.JobRequestID + runner.Status.JobRepositoryName = patch.Status.JobRepositoryName + runner.Status.JobID = patch.Status.JobID + runner.Status.WorkflowRunID = patch.Status.WorkflowRunID + runner.Status.JobWorkflowRef = patch.Status.JobWorkflowRef + runner.Status.JobDisplayName = patch.Status.JobDisplayName + if patch.Status.Phase != "" { + runner.Status.Phase = patch.Status.Phase + } + runner.ResourceVersion = strconv.Itoa(mustAtoi(t, runner.ResourceVersion) + 1) + + require.NoError(t, json.NewEncoder(w).Encode(runner)) + default: + http.Error(w, "unexpected method", http.StatusMethodNotAllowed) + } + })) + + clientset, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + require.NoError(t, err) + + return &Scaler{ + clientset: clientset, + config: Config{ + EphemeralRunnerSetNamespace: runner.Namespace, + }, + targetRunners: -1, + patchSeq: -1, + logger: discardLogger, + }, server.Close +} + +func mustAtoi(t *testing.T, s string) int { + t.Helper() + + n, err := strconv.Atoi(s) + require.NoError(t, err) + return n +} + +func assertJobStartedStatus(t *testing.T, runner *v1alpha1.EphemeralRunner, jobInfo *scaleset.JobStarted) { + t.Helper() + + assert.Equal(t, jobInfo.RunnerRequestID, runner.Status.JobRequestID) + assert.Equal(t, jobInfo.JobID, runner.Status.JobID) + assert.Equal(t, jobInfo.OwnerName+"/"+jobInfo.RepositoryName, runner.Status.JobRepositoryName) + assert.Equal(t, jobInfo.WorkflowRunID, runner.Status.WorkflowRunID) + assert.Equal(t, jobInfo.JobWorkflowRef, runner.Status.JobWorkflowRef) + assert.Equal(t, jobInfo.JobDisplayName, runner.Status.JobDisplayName) +} + func TestSetDesiredWorkerState_MinMaxDefaults(t *testing.T) { newEmptyWorker := func() *Scaler { return &Scaler{ @@ -443,3 +643,171 @@ func TestSetDesiredWorkerState_MinMaxSet(t *testing.T) { assert.Equal(t, 2, w.patchSeq) }) } + +// recordedRequest captures one request the scaler issued to the API server. +type recordedRequest struct { + method string + path string + body string +} + +func methodsOf(requests []recordedRequest) []string { + methods := make([]string, 0, len(requests)) + for _, request := range requests { + methods = append(methods, request.method) + } + return methods +} + +// newRecordingScaler serves runner over a stub API server that records every +// request and answers the verb named by notFoundFor with a 404 (empty serves +// both verbs normally). +// +// Recording the requests, rather than only the returned error, is what makes +// the NotFound paths observable at all: both log and return nil, so "no error" +// is equally consistent with the request having been skipped, having been +// issued and rejected, or having been retried. Only the request log tells those +// apart, and only a positive control proves an empty log is a real absence +// rather than a recorder that never worked. +func newRecordingScaler(t *testing.T, runner *v1alpha1.EphemeralRunner, notFoundFor string) (*Scaler, *[]recordedRequest, func()) { + t.Helper() + + requests := &[]recordedRequest{} + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body bytes.Buffer + _, err := body.ReadFrom(r.Body) + require.NoError(t, err) + + *requests = append(*requests, recordedRequest{ + method: r.Method, + path: r.URL.Path, + body: body.String(), + }) + + w.Header().Set("Content-Type", "application/json") + + if r.Method == notFoundFor { + w.WriteHeader(http.StatusNotFound) + require.NoError(t, json.NewEncoder(w).Encode(&metav1.Status{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Status"}, + Status: metav1.StatusFailure, + Code: http.StatusNotFound, + Reason: metav1.StatusReasonNotFound, + Message: fmt.Sprintf("ephemeralrunners.actions.github.com %q not found", + runner.Name), + })) + return + } + + switch r.Method { + case http.MethodGet: + require.NoError(t, json.NewEncoder(w).Encode(runner)) + case http.MethodPatch: + var patch v1alpha1.EphemeralRunner + require.NoError(t, json.Unmarshal(body.Bytes(), &patch)) + + runner.Status.JobRequestID = patch.Status.JobRequestID + runner.Status.JobRepositoryName = patch.Status.JobRepositoryName + runner.Status.JobID = patch.Status.JobID + runner.Status.WorkflowRunID = patch.Status.WorkflowRunID + runner.Status.JobWorkflowRef = patch.Status.JobWorkflowRef + runner.Status.JobDisplayName = patch.Status.JobDisplayName + if patch.Status.Phase != "" { + runner.Status.Phase = patch.Status.Phase + } + runner.ResourceVersion = strconv.Itoa(mustAtoi(t, runner.ResourceVersion) + 1) + + require.NoError(t, json.NewEncoder(w).Encode(runner)) + default: + http.Error(w, "unexpected method", http.StatusMethodNotAllowed) + } + })) + + clientset, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + require.NoError(t, err) + + return &Scaler{ + clientset: clientset, + config: Config{ + EphemeralRunnerSetNamespace: runner.Namespace, + }, + targetRunners: -1, + patchSeq: -1, + logger: discardLogger, + }, requests, server.Close +} + +// TestHandleJobStarted_NotFound covers the two paths that swallow a NotFound +// and return nil. A deleted runner is an expected race rather than an error -- +// the listener learns a job started for a runner the controller has already +// removed -- so the job info update is abandoned instead of failing the +// message handler and being redelivered forever. +// +// Neither path produces any observable state change, which is exactly why they +// had no coverage: there is nothing to assert on afterwards. Each case is +// therefore asserted against the request log and paired with a positive +// control, so an empty or short log is a measured absence rather than an +// unasked question. +func TestHandleJobStarted_NotFound(t *testing.T) { + 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, + }, + } + + t.Run("positive control: the recorder observes a successful promotion", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + scaler, requests, shutdown := newRecordingScaler(t, runner, "") + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + require.Equal(t, []string{http.MethodGet, http.MethodPatch}, methodsOf(*requests)) + // The recorded patch body is the load-bearing observation: it establishes + // that this recorder does capture a promotion when one is issued, which is + // what licenses reading its absence below as "no patch was sent". + assert.Contains(t, (*requests)[1].body, `"phase":"Running"`) + assert.Equal(t, "/apis/actions.github.com/v1alpha1/namespaces/default/ephemeralrunners/runner-1/status", (*requests)[1].path) + assert.Equal(t, v1alpha1.EphemeralRunnerPhaseRunning, runner.Status.Phase) + }) + + t.Run("get not found abandons the update without patching", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + scaler, requests, shutdown := newRecordingScaler(t, runner, http.MethodGet) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + // This swallow returns nil from inside the RetryOnConflict closure, so it + // ends the retry loop as a success. Pinning the exact request sequence is + // what distinguishes that from a silent retry or a patch against a runner + // that is known to be gone. + assert.Equal(t, []string{http.MethodGet}, methodsOf(*requests)) + assert.Equal(t, "/apis/actions.github.com/v1alpha1/namespaces/default/ephemeralrunners/runner-1", (*requests)[0].path) + assert.Equal(t, v1alpha1.EphemeralRunnerPhasePending, runner.Status.Phase) + }) + + t.Run("patch not found is swallowed and not retried", func(t *testing.T) { + runner := newTestEphemeralRunner(jobInfo.RunnerName, v1alpha1.EphemeralRunnerPhasePending) + scaler, requests, shutdown := newRecordingScaler(t, runner, http.MethodPatch) + defer shutdown() + + require.NoError(t, scaler.HandleJobStarted(context.Background(), jobInfo)) + + // Exactly one patch: a 404 must not be mistaken for a conflict and retried + // against state that will never come back. + require.Equal(t, []string{http.MethodGet, http.MethodPatch}, methodsOf(*requests)) + // The promotion really was attempted, so the unchanged phase below is the + // 404 being swallowed rather than the scaler declining to patch. + assert.Contains(t, (*requests)[1].body, `"phase":"Running"`) + assert.Equal(t, v1alpha1.EphemeralRunnerPhasePending, runner.Status.Phase) + }) +} diff --git a/config/crd/bases/actions.github.com_ephemeralrunners.yaml b/config/crd/bases/actions.github.com_ephemeralrunners.yaml index a174d751d4..e6d671a139 100644 --- a/config/crd/bases/actions.github.com_ephemeralrunners.yaml +++ b/config/crd/bases/actions.github.com_ephemeralrunners.yaml @@ -8323,6 +8323,10 @@ spec: The PodSucceded phase should be set only when confirmed that EphemeralRunner actually executed the job and has been removed from the service. + + The Running phase is owned by the listener and is set only when a job has + been assigned to this EphemeralRunner. It does not mean the runner is merely + online and waiting for work; an idle registered runner stays Pending. type: string ready: description: Turns true only if the runner is online. diff --git a/controllers/actions.github.com/ephemeralrunner_controller.go b/controllers/actions.github.com/ephemeralrunner_controller.go index 7608ef8bf2..9998793c37 100644 --- a/controllers/actions.github.com/ephemeralrunner_controller.go +++ b/controllers/actions.github.com/ephemeralrunner_controller.go @@ -834,6 +834,7 @@ func (r *EphemeralRunnerReconciler) createSecret(ctx context.Context, runner *v1 // updateRunStatusFromPod is responsible for updating non-exiting statuses. // It should never update phase to Failed or Succeeded +// It should never update phase to Running (the listener owns that transition) // // The event should not be re-queued since the termination status should be set // before proceeding with reconciliation logic @@ -851,8 +852,25 @@ func (r *EphemeralRunnerReconciler) updateRunStatusFromPod(ctx context.Context, } } - phase := v1alpha1.EphemeralRunnerPhase(pod.Status.Phase) - phaseChanged := ephemeralRunner.Status.Phase != phase + // Publish Pending as soon as the runner is observed non-terminal, regardless of + // the pod phase. The controller only reaches this point once the runner + // container status exists, and by then the pod has usually already advanced to + // Running, so keying the initial phase off PodPending would leave a runner + // phase-empty for its whole life -- omitted from the phase metrics, and in + // breach of the documented contract that Pending means "created, no job yet". + // Guarding on the empty phase alone is sufficient: every terminal phase, and + // Running itself, is non-empty, so this can never overwrite one. + phase := ephemeralRunner.Status.Phase + if phase == "" { + phase = v1alpha1.EphemeralRunnerPhasePending + } + + // The controller no longer promotes the runner to Running. The listener owns that + // transition and applies it when a job is assigned to this runner. The controller + // still publishes the initial Pending phase while the runner pod is starting. + // The patch below is optimistically locked so a stale cached copy of this runner + // cannot undo the listener's transition to Running. + phaseChanged := phase != ephemeralRunner.Status.Phase readyChanged := ready != ephemeralRunner.Status.Ready if !phaseChanged && !readyChanged { @@ -872,7 +890,7 @@ func (r *EphemeralRunnerReconciler) updateRunStatusFromPod(ctx context.Context, ephemeralRunner.Status.Reason = pod.Status.Reason ephemeralRunner.Status.Message = pod.Status.Message - if err := r.Status().Patch(ctx, ephemeralRunner, client.MergeFrom(original)); err != nil { + if err := r.Status().Patch(ctx, ephemeralRunner, client.MergeFromWithOptions(original, client.MergeFromWithOptimisticLock{})); err != nil { return fmt.Errorf("failed to update runner status for Phase/Reason/Message/Ready: %w", err) } r.publishEphemeralRunnerPhaseMetric(ephemeralRunner, ephemeralRunner.Status.Phase, log) diff --git a/controllers/actions.github.com/ephemeralrunner_controller_test.go b/controllers/actions.github.com/ephemeralrunner_controller_test.go index 74f9fe9923..e295a55812 100644 --- a/controllers/actions.github.com/ephemeralrunner_controller_test.go +++ b/controllers/actions.github.com/ephemeralrunner_controller_test.go @@ -825,32 +825,47 @@ var _ = Describe("EphemeralRunner", func() { ephemeralRunnerInterval, ).Should(BeEquivalentTo(true)) - for _, phase := range []corev1.PodPhase{corev1.PodRunning, corev1.PodPending} { - podCopy := pod.DeepCopy() - pod.Status.Phase = phase - // set container state to force status update - pod.Status.ContainerStatuses = append(pod.Status.ContainerStatuses, corev1.ContainerStatus{ - Name: v1alpha1.EphemeralRunnerContainerName, - State: corev1.ContainerState{}, - }) + podCopy := pod.DeepCopy() + pod.Status.Phase = corev1.PodPending + // set container state to force status update + pod.Status.ContainerStatuses = append(pod.Status.ContainerStatuses, corev1.ContainerStatus{ + Name: v1alpha1.EphemeralRunnerContainerName, + State: corev1.ContainerState{}, + }) - err := k8sClient.Status().Patch(ctx, pod, client.MergeFrom(podCopy)) - Expect(err).To(BeNil(), "failed to patch pod status") + err := k8sClient.Status().Patch(ctx, pod, client.MergeFrom(podCopy)) + Expect(err).To(BeNil(), "failed to patch pod status") - var updated *v1alpha1.EphemeralRunner - Eventually( - func() (v1alpha1.EphemeralRunnerPhase, error) { - updated = new(v1alpha1.EphemeralRunner) - err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated) - if err != nil { - return "", err - } - return updated.Status.Phase, nil - }, - ephemeralRunnerTimeout, - ephemeralRunnerInterval, - ).Should(BeEquivalentTo(phase)) - } + Eventually( + func() (v1alpha1.EphemeralRunnerPhase, error) { + updated := new(v1alpha1.EphemeralRunner) + err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated) + if err != nil { + return "", err + } + return updated.Status.Phase, nil + }, + ephemeralRunnerTimeout, + ephemeralRunnerInterval, + ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhasePending)) + + podCopy = pod.DeepCopy() + pod.Status.Phase = corev1.PodRunning + err = k8sClient.Status().Patch(ctx, pod, client.MergeFrom(podCopy)) + Expect(err).To(BeNil(), "failed to patch pod status") + + Consistently( + func() (v1alpha1.EphemeralRunnerPhase, error) { + updated := new(v1alpha1.EphemeralRunner) + err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated) + if err != nil { + return "", err + } + return updated.Status.Phase, nil + }, + ephemeralRunnerInterval*3, + ephemeralRunnerInterval, + ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhasePending), "controller should not set Running from pod status") }) It("It should update ready based on the latest condition", func() { @@ -1173,7 +1188,6 @@ var _ = Describe("EphemeralRunner", func() { ephemeralRunnerInterval, ).Should(BeEquivalentTo(true)) - // first set phase to running pod.Status.ContainerStatuses = append(pod.Status.ContainerStatuses, corev1.ContainerStatus{ Name: v1alpha1.EphemeralRunnerContainerName, State: corev1.ContainerState{ @@ -1186,7 +1200,20 @@ var _ = Describe("EphemeralRunner", func() { err := k8sClient.Status().Update(ctx, pod) Expect(err).To(BeNil()) - Eventually( + updated := new(v1alpha1.EphemeralRunner) + err = k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated) + Expect(err).To(BeNil()) + + original := updated.DeepCopy() + updated.Status.Phase = v1alpha1.EphemeralRunnerPhaseRunning + err = k8sClient.Status().Patch(ctx, updated, client.MergeFrom(original)) + Expect(err).To(BeNil()) + + pod.Status.Phase = corev1.PodSucceeded + err = k8sClient.Status().Update(ctx, pod) + Expect(err).To(BeNil()) + + Consistently( func() (v1alpha1.EphemeralRunnerPhase, error) { updated := new(v1alpha1.EphemeralRunner) if err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated); err != nil { @@ -1195,24 +1222,79 @@ var _ = Describe("EphemeralRunner", func() { return updated.Status.Phase, nil }, ephemeralRunnerTimeout, - ephemeralRunnerInterval, ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhaseRunning)) + }) - // set phase to succeeded - pod.Status.Phase = corev1.PodSucceeded - err = k8sClient.Status().Update(ctx, pod) + It("Controller should not set Running phase from pod status - listener owns Running transition", func() { + pod := new(corev1.Pod) + Eventually( + func() (bool, error) { + if err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, pod); err != nil { + return false, err + } + return true, nil + }, + ephemeralRunnerTimeout, + ephemeralRunnerInterval, + ).Should(BeEquivalentTo(true)) + + pod.Status.ContainerStatuses = append(pod.Status.ContainerStatuses, corev1.ContainerStatus{ + Name: v1alpha1.EphemeralRunnerContainerName, + State: corev1.ContainerState{ + Running: &corev1.ContainerStateRunning{ + StartedAt: metav1.Now(), + }, + }, + }) + pod.Status.Phase = corev1.PodRunning + pod.Status.Conditions = append(pod.Status.Conditions, corev1.PodCondition{ + Type: corev1.PodReady, + Status: corev1.ConditionTrue, + LastTransitionTime: metav1.Now(), + }) + err := k8sClient.Status().Update(ctx, pod) Expect(err).To(BeNil()) + // Two-stage on purpose. Eventually establishes that the controller does + // publish Pending even though the pod was first observed already Running + // -- the common case once the image is cached, and the only chance the + // controller gets to publish an initial phase. Consistently then holds + // that it never advances to Running, which is the listener's transition + // to make. Asserting Pending is strictly stronger than asserting empty, + // because empty is also what a controller that never ran would leave. + updated := new(v1alpha1.EphemeralRunner) + Eventually( + func() (v1alpha1.EphemeralRunnerPhase, error) { + if err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated); err != nil { + return "Unknown", err + } + return updated.Status.Phase, nil + }, + ephemeralRunnerTimeout, + ephemeralRunnerInterval, + ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhasePending), "controller must publish the initial Pending phase") + Consistently( func() (v1alpha1.EphemeralRunnerPhase, error) { updated := new(v1alpha1.EphemeralRunner) if err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated); err != nil { - return "", err + return "Unknown", err } return updated.Status.Phase, nil }, ephemeralRunnerTimeout, - ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhaseRunning)) + ).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerPhasePending), "controller must not set Running from pod status") + + Eventually( + func() (bool, error) { + if err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, updated); err != nil { + return false, err + } + return updated.Status.Ready, nil + }, + ephemeralRunnerTimeout, + ephemeralRunnerInterval, + ).Should(BeEquivalentTo(true)) }) }) diff --git a/controllers/actions.github.com/ephemeralrunnerset_controller_test.go b/controllers/actions.github.com/ephemeralrunnerset_controller_test.go index 993c62e38d..9de0ad2f49 100644 --- a/controllers/actions.github.com/ephemeralrunnerset_controller_test.go +++ b/controllers/actions.github.com/ephemeralrunnerset_controller_test.go @@ -1840,6 +1840,14 @@ var _ = Describe("EphemeralRunner phase metrics", func() { err = k8sClient.Status().Patch(ctx, podRunning, client.MergeFrom(podPending)) Expect(err).NotTo(HaveOccurred(), "failed to patch pod to running") + runnerRunning := new(v1alpha1.EphemeralRunner) + err = k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunner.Name, Namespace: ephemeralRunner.Namespace}, runnerRunning) + Expect(err).NotTo(HaveOccurred(), "failed to get ephemeral runner before listener-owned running patch") + runnerRunningOriginal := runnerRunning.DeepCopy() + runnerRunning.Status.Phase = v1alpha1.EphemeralRunnerPhaseRunning + err = k8sClient.Status().Patch(ctx, runnerRunning, client.MergeFrom(runnerRunningOriginal)) + Expect(err).NotTo(HaveOccurred(), "failed to simulate listener running phase patch") + _, err = controller.Reconcile(ctx, request) Expect(err).NotTo(HaveOccurred(), "failed to reconcile running pod") expectEphemeralRunnerPhase(ctx, ephemeralRunner, v1alpha1.EphemeralRunnerPhaseRunning) diff --git a/controllers/actions.github.com/metrics/metrics.go b/controllers/actions.github.com/metrics/metrics.go index 1d1f28670d..d759ff3fcb 100644 --- a/controllers/actions.github.com/metrics/metrics.go +++ b/controllers/actions.github.com/metrics/metrics.go @@ -39,7 +39,7 @@ var ( prometheus.GaugeOpts{ Subsystem: githubScaleSetControllerSubsystem, Name: "pending_ephemeral_runners", - Help: "Number of ephemeral runners in a pending state.", + Help: "Number of ephemeral runners that have not been assigned a job yet.", }, labels, ) @@ -47,7 +47,7 @@ var ( prometheus.GaugeOpts{ Subsystem: githubScaleSetControllerSubsystem, Name: "running_ephemeral_runners", - Help: "Number of ephemeral runners in a running state.", + Help: "Number of ephemeral runners that have been assigned a job.", }, labels, ) diff --git a/controllers/actions.github.com/resourcebuilder.go b/controllers/actions.github.com/resourcebuilder.go index ea092aab09..4dc13f5b89 100644 --- a/controllers/actions.github.com/resourcebuilder.go +++ b/controllers/actions.github.com/resourcebuilder.go @@ -1022,7 +1022,12 @@ func rulesForListenerRole(resourceNames []string) []rbacv1.PolicyRule { }, { APIGroups: []string{"actions.github.com"}, - Resources: []string{"ephemeralrunners", "ephemeralrunners/status"}, + Resources: []string{"ephemeralrunners"}, + Verbs: []string{"get", "patch"}, + }, + { + APIGroups: []string{"actions.github.com"}, + Resources: []string{"ephemeralrunners/status"}, Verbs: []string{"patch"}, }, } diff --git a/docs/adrs/2023-05-08-exposing-metrics.md b/docs/adrs/2023-05-08-exposing-metrics.md index 6dc2fd7eee..16b6d2f7d1 100644 --- a/docs/adrs/2023-05-08-exposing-metrics.md +++ b/docs/adrs/2023-05-08-exposing-metrics.md @@ -158,13 +158,14 @@ started. To get a better understanding of health and workings of the cluster resources, we need to expose the following metrics: -- `pending_ephemeral_runners` - Number of ephemeral runners in a pending state. - This information can show the latency between creating an `EphemeralRunner` - resource, and having an ephemeral runner pod started and ready to receive a - job. -- `running_ephemeral_runners` - Number of ephemeral runners currently running. - This information is helpful to see how many ephemeral runner pods are running - at any given time. +- `pending_ephemeral_runners` - Number of ephemeral runners that have not been + assigned a job yet. This covers both runners whose pod has not finished + starting and runners that are registered and idle, so with a non-zero + `minRunners` it does not drop to zero. +- `running_ephemeral_runners` - Number of ephemeral runners that have been + assigned a job. This information is helpful to see how many ephemeral runners + are executing a workflow job at any given time. It reflects job assignment, + not pod liveness. - `failed_ephemeral_runners` - Number of ephemeral runners in a `Failed` state. This information is helpful to catch the faulty image, or some underlying problem. When the ephemeral runner controller is not able to start the diff --git a/docs/gha-runner-scale-set-controller/samples/grafana-dashboard/README.md b/docs/gha-runner-scale-set-controller/samples/grafana-dashboard/README.md index d9965aeced..4edc435dde 100644 --- a/docs/gha-runner-scale-set-controller/samples/grafana-dashboard/README.md +++ b/docs/gha-runner-scale-set-controller/samples/grafana-dashboard/README.md @@ -66,7 +66,7 @@ The dashboard includes the following metrics: | Running Jobs | The number of runners that are currently processing jobs. | | Failed Runners | The total number of ephemeral runners that have failed to properly start. This may require reviewing the custom resource and logs to identify and resolve the root causes. Common causes include resource issues and failure to pull the required image. | | Listeners | The number of listeners currently running and attempting to manage jobs for the scale set. This should match the number of scale sets deployed. | -| Pending Runners | The total number of ephemeral runners that ARC has requested and is waiting for Kubernetes to provide in a running state. If the Kubernetes API server is responsive, this will typically match the number of runner pods that are in a pending state. This number includes requests for runner pods that have not yet been scheduled. When this number is higher than the number of runner pods in a pending state, it can indicate performance issues. | +| Pending Runners | The total number of ephemeral runners that have not been assigned a job. This covers runners that Kubernetes has not finished starting as well as runners that are registered and idle waiting for work, so with a non-zero `minRunners` it does not fall to zero. On its own it is therefore not a signal of scheduling trouble; compare it against the number of runner pods actually in a pending state, and treat a persistent excess of pending runners over pending pods as the indicator of performance issues. | | Registered Runners | The total number of ephemeral runners that have been successfully registered. | | Active Runners | The total number of runners that are active and either available or processing jobs. | | Out of Memory | The number of containers that have been terminated by the OOMKiller. This can indicate that the requests/ limits for one or more pods on the node were configured improperly, allowing pods to request more memory than the node had available. |