This commit is contained in:
Nikola Jokic
2026-07-14 19:57:17 +02:00
parent 3a6ce79672
commit 195f9a4e85
7 changed files with 60 additions and 164 deletions
@@ -292,18 +292,6 @@ func (r *AutoscalingRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl
}
if ephemeralRunnerSet.Annotations[annotationKeyIntegrityHash] != desired.Annotations[annotationKeyIntegrityHash] {
// When runners are actively processing jobs, defer the spec update:
// delete the listener to stop accepting new jobs, but leave the ERS
// (and its running pods) untouched until all jobs have drained.
if ephemeralRunnerSet.Status.RunningEphemeralRunners+ephemeralRunnerSet.Status.PendingEphemeralRunners > 0 {
log.Info("Ephemeral runner set spec changed but runners are still active; deleting listener to stop new jobs")
if _, err := r.cleanupListener(ctx, &autoscalingRunnerSet, log); err != nil {
log.Error(err, "Failed to clean up listener while waiting for runners to drain")
return ctrl.Result{}, err
}
return ctrl.Result{RequeueAfter: 1 * time.Second}, nil
}
original := ephemeralRunnerSet.DeepCopy()
ephemeralRunnerSet.Spec.EphemeralRunnerMetadata = desired.Spec.EphemeralRunnerMetadata
ephemeralRunnerSet.Spec.EphemeralRunnerSpec = desired.Spec.EphemeralRunnerSpec
@@ -438,9 +426,7 @@ func (r *AutoscalingRunnerSetReconciler) updateStatus(ctx context.Context, autos
}
original := autoscalingRunnerSet.DeepCopy()
if phaseDiff {
autoscalingRunnerSet.Status.Phase = phase
}
autoscalingRunnerSet.Status.Phase = phase
if err := r.Status().Patch(ctx, autoscalingRunnerSet, client.MergeFrom(original)); err != nil {
log.Error(err, "Failed to patch autoscaling runner set status")
@@ -837,111 +837,6 @@ var _ = Describe("Test AutoScalingRunnerSet controller", Ordered, func() {
})
})
Context("When updating an AutoscalingRunnerSet with running or pending jobs", func() {
It("It should wait for running and pending jobs to finish before applying the update.", func() {
// Wait till the listener is created
listener := new(v1alpha1.AutoscalingListener)
Eventually(
func() error {
return k8sClient.Get(ctx, client.ObjectKey{Name: scaleSetListenerName(autoscalingRunnerSet), Namespace: autoscalingRunnerSet.Namespace}, listener)
},
autoscalingRunnerSetTestTimeout,
autoscalingRunnerSetTestInterval,
).Should(Succeed(), "Listener should be created")
// Wait till the ephemeral runner set is created
Eventually(
func() (int, error) {
runnerSetList := new(v1alpha1.EphemeralRunnerSetList)
err := k8sClient.List(ctx, runnerSetList, client.InNamespace(autoscalingRunnerSet.Namespace))
if err != nil {
return 0, err
}
return len(runnerSetList.Items), nil
},
autoscalingRunnerSetTestTimeout,
autoscalingRunnerSetTestInterval,
).Should(BeEquivalentTo(1), "Only one EphemeralRunnerSet should be created")
runnerSetList := new(v1alpha1.EphemeralRunnerSetList)
err := k8sClient.List(ctx, runnerSetList, client.InNamespace(autoscalingRunnerSet.Namespace))
Expect(err).NotTo(HaveOccurred(), "failed to list EphemeralRunnerSet")
// Emulate running and pending jobs
runnerSet := runnerSetList.Items[0]
activeRunnerSet := runnerSet.DeepCopy()
activeRunnerSet.Status.CurrentReplicas = 6
activeRunnerSet.Status.FailedEphemeralRunners = 1
activeRunnerSet.Status.RunningEphemeralRunners = 2
activeRunnerSet.Status.PendingEphemeralRunners = 3
desiredStatus := v1alpha1.AutoscalingRunnerSetStatus{
CurrentRunners: activeRunnerSet.Status.CurrentReplicas,
Phase: v1alpha1.AutoscalingRunnerSetPhaseRunning,
PendingEphemeralRunners: activeRunnerSet.Status.PendingEphemeralRunners,
RunningEphemeralRunners: activeRunnerSet.Status.RunningEphemeralRunners,
FailedEphemeralRunners: activeRunnerSet.Status.FailedEphemeralRunners,
}
err = k8sClient.Status().Patch(ctx, activeRunnerSet, client.MergeFrom(&runnerSet))
Expect(err).NotTo(HaveOccurred(), "Failed to patch runner set status")
Eventually(
func() (v1alpha1.AutoscalingRunnerSetStatus, error) {
updated := new(v1alpha1.AutoscalingRunnerSet)
err := k8sClient.Get(ctx, client.ObjectKey{Name: autoscalingRunnerSet.Name, Namespace: autoscalingRunnerSet.Namespace}, updated)
if err != nil {
return v1alpha1.AutoscalingRunnerSetStatus{}, fmt.Errorf("failed to get AutoScalingRunnerSet: %w", err)
}
return updated.Status, nil
},
autoscalingRunnerSetTestTimeout,
autoscalingRunnerSetTestInterval,
).Should(BeEquivalentTo(desiredStatus), "AutoScalingRunnerSet status should be updated")
// Patch the AutoScalingRunnerSet image which should trigger
// the recreation of the Listener and EphemeralRunnerSet
patched := autoscalingRunnerSet.DeepCopy()
if patched.Annotations == nil {
patched.Annotations = make(map[string]string)
}
patched.Annotations[annotationKeyIntegrityHash] = "testgroup2"
patched.Spec.Template.Spec = corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "runner",
Image: "ghcr.io/actions/abcd:1.1.1",
},
},
}
err = k8sClient.Patch(ctx, patched, client.MergeFrom(autoscalingRunnerSet))
Expect(err).NotTo(HaveOccurred(), "failed to patch AutoScalingRunnerSet")
autoscalingRunnerSet = patched.DeepCopy()
// The EphemeralRunnerSet should not be recreated
Consistently(
func() (string, error) {
runnerSetList := new(v1alpha1.EphemeralRunnerSetList)
err := k8sClient.List(ctx, runnerSetList, client.InNamespace(autoscalingRunnerSet.Namespace))
Expect(err).NotTo(HaveOccurred(), "failed to fetch AutoScalingRunnerSet")
return runnerSetList.Items[0].Name, nil
},
autoscalingRunnerSetTestTimeout,
autoscalingRunnerSetTestInterval,
).Should(Equal(activeRunnerSet.Name), "The EphemeralRunnerSet should not be recreated")
// The listener should not be recreated
Consistently(
func() error {
return k8sClient.Get(ctx, client.ObjectKey{Name: scaleSetListenerName(autoscalingRunnerSet), Namespace: autoscalingRunnerSet.Namespace}, listener)
},
autoscalingRunnerSetTestTimeout,
autoscalingRunnerSetTestInterval,
).ShouldNot(Succeed(), "Listener should not be recreated")
})
})
It("Should update Status on EphemeralRunnerSet status Update", func() {
ars := new(v1alpha1.AutoscalingRunnerSet)
Eventually(
@@ -978,17 +873,10 @@ var _ = Describe("Test AutoScalingRunnerSet controller", Ordered, func() {
runnerSet := runnerSetList.Items[0]
statusUpdate := runnerSet.DeepCopy()
statusUpdate.Status.CurrentReplicas = 6
statusUpdate.Status.FailedEphemeralRunners = 1
statusUpdate.Status.RunningEphemeralRunners = 2
statusUpdate.Status.PendingEphemeralRunners = 3
statusUpdate.Status.Phase = v1alpha1.EphemeralRunnerSetPhaseRunning
desiredStatus := v1alpha1.AutoscalingRunnerSetStatus{
CurrentRunners: statusUpdate.Status.CurrentReplicas,
Phase: v1alpha1.AutoscalingRunnerSetPhaseRunning,
PendingEphemeralRunners: statusUpdate.Status.PendingEphemeralRunners,
RunningEphemeralRunners: statusUpdate.Status.RunningEphemeralRunners,
FailedEphemeralRunners: statusUpdate.Status.FailedEphemeralRunners,
Phase: v1alpha1.AutoscalingRunnerSetPhaseRunning,
}
err := k8sClient.Status().Patch(ctx, statusUpdate, client.MergeFrom(&runnerSet))
@@ -246,7 +246,6 @@ func (r *EphemeralRunnerSetReconciler) Reconcile(ctx context.Context, req ctrl.R
func (r *EphemeralRunnerSetReconciler) updateStatus(ctx context.Context, ephemeralRunnerSet *v1alpha1.EphemeralRunnerSet, state *ephemeralRunnersByState, log logr.Logger) error {
original := ephemeralRunnerSet.DeepCopy()
total := state.scaleTotal()
var phase v1alpha1.EphemeralRunnerSetPhase
switch {
case len(state.outdated) > 0:
@@ -257,11 +256,7 @@ func (r *EphemeralRunnerSetReconciler) updateStatus(ctx context.Context, ephemer
phase = ephemeralRunnerSet.Status.Phase
}
desiredStatus := v1alpha1.EphemeralRunnerSetStatus{
CurrentReplicas: total,
Phase: phase,
PendingEphemeralRunners: len(state.pending),
RunningEphemeralRunners: len(state.running),
FailedEphemeralRunners: len(state.failed),
Phase: phase,
}
// Update the status if needed.
@@ -222,20 +222,20 @@ var _ = Describe("Test EphemeralRunnerSet controller", func() {
ephemeralRunnerSetTestInterval,
).Should(BeEquivalentTo(0), "No EphemeralRunner should be created")
// Check if the status stay 0
// Check if the status is initialized
Consistently(
func() (int, error) {
func() (v1alpha1.EphemeralRunnerSetPhase, error) {
runnerSet := new(v1alpha1.EphemeralRunnerSet)
err := k8sClient.Get(ctx, client.ObjectKey{Name: ephemeralRunnerSet.Name, Namespace: ephemeralRunnerSet.Namespace}, runnerSet)
if err != nil {
return -1, err
return "", err
}
return int(runnerSet.Status.CurrentReplicas), nil
return runnerSet.Status.Phase, nil
},
ephemeralRunnerSetTestTimeout,
ephemeralRunnerSetTestInterval,
).Should(BeEquivalentTo(0), "EphemeralRunnerSet status should be 0")
).Should(BeEquivalentTo(v1alpha1.EphemeralRunnerSetPhaseRunning), "EphemeralRunnerSet status should be running")
// Scaling up the EphemeralRunnerSet
updated := created.DeepCopy()
@@ -1258,11 +1258,7 @@ var _ = Describe("Test EphemeralRunnerSet controller", func() {
).Should(BeTrue(), "Failed to eventually update to one pending, one running and one failed")
desiredStatus := v1alpha1.EphemeralRunnerSetStatus{
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
CurrentReplicas: 3,
PendingEphemeralRunners: 1,
RunningEphemeralRunners: 1,
FailedEphemeralRunners: 1,
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
}
Eventually(
func() (v1alpha1.EphemeralRunnerSetStatus, error) {
@@ -1301,11 +1297,7 @@ var _ = Describe("Test EphemeralRunnerSet controller", func() {
).Should(BeEquivalentTo(1), "Failed to eventually scale down")
desiredStatus = v1alpha1.EphemeralRunnerSetStatus{
CurrentReplicas: 1,
PendingEphemeralRunners: 0,
RunningEphemeralRunners: 0,
FailedEphemeralRunners: 1,
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
}
Eventually(
@@ -1325,11 +1317,7 @@ var _ = Describe("Test EphemeralRunnerSet controller", func() {
Expect(err).To(BeNil(), "Failed to delete failed ephemeral runner")
desiredStatus = v1alpha1.EphemeralRunnerSetStatus{
CurrentReplicas: 0,
PendingEphemeralRunners: 0,
RunningEphemeralRunners: 0,
FailedEphemeralRunners: 0,
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
}
Eventually(
func() (v1alpha1.EphemeralRunnerSetStatus, error) {