fix: clean up runners after terminal jobs

Author: robinbraemer

apis/actions.github.com/v1alpha1/ephemeralrunner_types.go

@@ -183,6 +183,21 @@ type EphemeralRunnerStatus struct {
 
 	// +optional
 	JobDisplayName string `json:"jobDisplayName,omitempty"`
+
+	// JobCompletion records the terminal job event received for this runner.
+	// +optional
+	JobCompletion *EphemeralRunnerJobCompletion `json:"jobCompletion,omitempty"`
+}
+
+// EphemeralRunnerJobCompletion identifies the terminal job event received for
+// an ephemeral runner. The controller uses all identity fields before deleting
+// a runner whose process did not exit after the job became terminal.
+type EphemeralRunnerJobCompletion struct {
+	Result        string      `json:"result"`
+	RunnerID      int         `json:"runnerId"`
+	JobID         string      `json:"jobId"`
+	WorkflowRunID int64       `json:"workflowRunId"`
+	FinishedAt    metav1.Time `json:"finishedAt"`
 }
 
 // EphemeralRunnerPhase is the phase of the ephemeral runner.

apis/actions.github.com/v1alpha1/zz_generated.deepcopy.go

@@ -12,6 +12,7 @@ import (
 	"github.com/actions/scaleset/listener"
 	jsonpatch "github.com/evanphx/json-patch"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -156,6 +157,59 @@ func (w *Scaler) HandleJobStarted(ctx context.Context, jobInfo *scaleset.JobStar
 
 func (w *Scaler) HandleJobCompleted(ctx context.Context, msg *scaleset.JobCompleted) error {
 	w.dirty = true
+
+	w.logger.Info("Recording completed job for the runner",
+		"runnerName", msg.RunnerName,
+		"runnerId", msg.RunnerID,
+		"jobId", msg.JobID,
+		"workflowRunId", msg.WorkflowRunID,
+		"result", msg.Result,
+		"finishedAt", msg.FinishTime)
+
+	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{
+			JobCompletion: &v1alpha1.EphemeralRunnerJobCompletion{
+				Result:        msg.Result,
+				RunnerID:      msg.RunnerID,
+				JobID:         msg.JobID,
+				WorkflowRunID: msg.WorkflowRunID,
+				FinishedAt:    metav1.NewTime(msg.FinishTime),
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal ephemeral runner completion patch: %w", err)
+	}
+
+	mergePatch, err := jsonpatch.CreateMergePatch(original, patch)
+	if err != nil {
+		return fmt.Errorf("failed to create completion merge patch for ephemeral runner: %w", err)
+	}
+
+	patchedStatus := &v1alpha1.EphemeralRunner{}
+	err = w.clientset.RESTClient().
+		Patch(types.MergePatchType).
+		Prefix("apis", v1alpha1.GroupVersion.Group, v1alpha1.GroupVersion.Version).
+		Namespace(w.config.EphemeralRunnerSetNamespace).
+		Resource("EphemeralRunners").
+		Name(msg.RunnerName).
+		SubResource("status").
+		Body(mergePatch).
+		Do(ctx).
+		Into(patchedStatus)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			w.logger.Info("Ephemeral runner not found, skipping completed job status patch", "runnerName", msg.RunnerName)
+			return nil
+		}
+		return fmt.Errorf("could not patch completed job status, patch JSON: %s, error: %w", string(mergePatch), err)
+	}
+
 	return nil
 }
 

charts/gha-runner-scale-set-controller-experimental/crds/actions.github.com_ephemeralrunners.yaml

@@ -44,6 +44,7 @@ import (
 const (
 	ephemeralRunnerFinalizerName        = "ephemeralrunner.actions.github.com/finalizer"
 	ephemeralRunnerActionsFinalizerName = "ephemeralrunner.actions.github.com/runner-registration-finalizer"
+	completedJobRunnerGracePeriod       = 30 * time.Second
 )
 
 // EphemeralRunnerReconciler reconciles a EphemeralRunner object
@@ -186,6 +187,29 @@ func (r *EphemeralRunnerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		log.Info("Successfully added finalizers")
 	}
 
+	if completion := ephemeralRunner.Status.JobCompletion; completion != nil {
+		if !jobCompletionMatchesRunner(&ephemeralRunner) {
+			log.Info("Ignoring completed job event that does not match the ephemeral runner status",
+				"completionRunnerId", completion.RunnerID,
+				"completionJobId", completion.JobID,
+				"completionWorkflowRunId", completion.WorkflowRunID)
+		} else {
+			deleteAfter := completion.FinishedAt.Add(completedJobRunnerGracePeriod)
+			if wait := time.Until(deleteAfter); wait > 0 {
+				log.Info("Waiting briefly for the runner process to exit after job completion", "requeueAfter", wait)
+				return ctrl.Result{RequeueAfter: wait}, nil
+			}
+
+			log.Info("Job is terminal but runner is still present; issuing delete",
+				"result", completion.Result,
+				"finishedAt", completion.FinishedAt)
+			if err := r.Delete(ctx, &ephemeralRunner); client.IgnoreNotFound(err) != nil {
+				return ctrl.Result{}, fmt.Errorf("failed to delete ephemeral runner after terminal job: %w", err)
+			}
+			return ctrl.Result{}, nil
+		}
+	}
+
 	secret := new(corev1.Secret)
 	if err := r.Get(ctx, req.NamespacedName, secret); err != nil {
 		if !kerrors.IsNotFound(err) {
@@ -971,6 +995,18 @@ func runnerContainerStatus(pod *corev1.Pod) *corev1.ContainerStatus {
 	return nil
 }
 
+func jobCompletionMatchesRunner(runner *v1alpha1.EphemeralRunner) bool {
+	completion := runner.Status.JobCompletion
+	if completion == nil || completion.Result == "" || completion.JobID == "" || completion.FinishedAt.IsZero() {
+		return false
+	}
+
+	return completion.RunnerID == runner.Status.RunnerID &&
+		runner.Status.RunnerName == runner.Name &&
+		completion.JobID == runner.Status.JobID &&
+		completion.WorkflowRunID == runner.Status.WorkflowRunID
+}
+
 func initContainerFailed(pod *corev1.Pod) bool {
 	for i := range pod.Status.InitContainerStatuses {
 		cs := &pod.Status.InitContainerStatuses[i]

charts/gha-runner-scale-set-controller/crds/actions.github.com_ephemeralrunners.yaml

@@ -185,6 +185,35 @@ var _ = Describe("EphemeralRunner", func() {
 			).Should(BeEquivalentTo(ephemeralRunner.Name))
 		})
 
+		It("deletes a live runner after receiving its exact terminal job event", func() {
+			created := new(v1alpha1.EphemeralRunner)
+			Eventually(func(g Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(ephemeralRunner), created)).To(Succeed())
+				g.Expect(created.Finalizers).To(ConsistOf(ephemeralRunnerFinalizerName, ephemeralRunnerActionsFinalizerName))
+
+				pod := new(corev1.Pod)
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(ephemeralRunner), pod)).To(Succeed())
+			}, ephemeralRunnerTimeout, ephemeralRunnerInterval).Should(Succeed())
+
+			created.Status.RunnerID = 2402
+			created.Status.RunnerName = created.Name
+			created.Status.JobID = "85cb98eb-2919-5766-872c-cc997f618c1f"
+			created.Status.WorkflowRunID = 31753891150
+			created.Status.JobCompletion = &v1alpha1.EphemeralRunnerJobCompletion{
+				Result:        "failed",
+				RunnerID:      created.Status.RunnerID,
+				JobID:         created.Status.JobID,
+				WorkflowRunID: created.Status.WorkflowRunID,
+				FinishedAt:    metav1.NewTime(time.Now().Add(-completedJobRunnerGracePeriod - time.Second)),
+			}
+			Expect(k8sClient.Status().Update(ctx, created)).To(Succeed())
+
+			Eventually(func() bool {
+				err := k8sClient.Get(ctx, client.ObjectKeyFromObject(ephemeralRunner), new(v1alpha1.EphemeralRunner))
+				return kerrors.IsNotFound(err)
+			}, ephemeralRunnerTimeout, ephemeralRunnerInterval).Should(BeTrue())
+		})
+
 		It("It should re-create pod on failure and no job assigned", func() {
 			pod := new(corev1.Pod)
 			Eventually(func() (bool, error) {

cmd/ghalistener/scaler/scaler.go

@@ -0,0 +1,46 @@
+package actionsgithubcom
+
+import (
+	"testing"
+	"time"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestJobCompletionMatchesRunner(t *testing.T) {
+	finishedAt := metav1.NewTime(time.Now())
+	runner := &v1alpha1.EphemeralRunner{
+		ObjectMeta: metav1.ObjectMeta{Name: "runner-1"},
+		Status: v1alpha1.EphemeralRunnerStatus{
+			RunnerID:      42,
+			RunnerName:    "runner-1",
+			JobID:         "job-1",
+			WorkflowRunID: 100,
+			JobCompletion: &v1alpha1.EphemeralRunnerJobCompletion{
+				Result:        "cancelled",
+				RunnerID:      42,
+				JobID:         "job-1",
+				WorkflowRunID: 100,
+				FinishedAt:    finishedAt,
+			},
+		},
+	}
+
+	assert.True(t, jobCompletionMatchesRunner(runner))
+
+	tests := map[string]func(*v1alpha1.EphemeralRunner){
+		"runner id":       func(r *v1alpha1.EphemeralRunner) { r.Status.JobCompletion.RunnerID++ },
+		"runner name":     func(r *v1alpha1.EphemeralRunner) { r.Status.RunnerName = "another-runner" },
+		"job id":          func(r *v1alpha1.EphemeralRunner) { r.Status.JobCompletion.JobID = "another-job" },
+		"workflow run id": func(r *v1alpha1.EphemeralRunner) { r.Status.JobCompletion.WorkflowRunID++ },
+	}
+	for name, mutate := range tests {
+		t.Run("rejects mismatched "+name, func(t *testing.T) {
+			candidate := runner.DeepCopy()
+			mutate(candidate)
+			assert.False(t, jobCompletionMatchesRunner(candidate))
+		})
+	}
+}