Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions docs/how-it-works.md
Original file line number Diff line number Diff line change
Expand Up @@ -109,8 +109,9 @@ relaxed before the eviction is admitted.
Voluntary disruption is held until the stand-in is ready. An involuntary
deadline is compared with `minSurgeTimeSeconds`; if too little time remains,
the hold is relaxed immediately while the surge continues. A hold is also
relaxed after `readinessDeadlineSeconds`, ensuring an unschedulable stand-in
cannot wedge a drain forever.
relaxed after `readinessDeadlineSeconds`, ensuring that neither an
unschedulable stand-in nor a surge the API server refuses (`SurgeReady=False`,
reason `SurgeFailed`) can wedge a drain forever.

`holdMode` chooses which of those disruptions are held in the first
place. `always` holds both classes, `voluntary-only` holds the unbounded
Expand Down
22 changes: 15 additions & 7 deletions internal/controller/understudy_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -280,12 +280,13 @@ func (r *UnderstudyReconciler) reconcileDoomed(ctx context.Context, cr *appsv1al
}
metrics.TrackBlocked(req.String(), status.BlockedSince.Time)

// A failed surge still reaches the status patch below: the hold must be
// recorded for readinessDeadlineSeconds to be able to relax it.
var surgeErr error
if !surgeActive || cr.Spec.BaselineReplicas != nil && target.Replicas() != base+cr.SurgeReplicasOrDefault() {
if err := target.PatchReplicas(ctx, r.Client, base+cr.SurgeReplicasOrDefault(), &base); err != nil {
return ctrl.Result{}, err
}
surgeErr = target.PatchReplicas(ctx, r.Client, base+cr.SurgeReplicasOrDefault(), &base)
}
if !surgeActive {
if surgeErr == nil && !surgeActive {
now := metav1.Now()
status.LastSurgeTime = &now
class := signal.VoluntaryUnbounded
Expand All @@ -297,9 +298,11 @@ func (r *UnderstudyReconciler) reconcileDoomed(ctx context.Context, cr *appsv1al
fmt.Sprintf("node doom (%s) for pod(s) of %s; surging %d -> %d",
doomSource(a.worstDoom), target.GetName(), base, base+cr.SurgeReplicasOrDefault()))
}

if err := r.stampDeletionCosts(ctx, a.doomed); err != nil {
return ctrl.Result{}, err
if surgeErr == nil {
surgeErr = r.stampDeletionCosts(ctx, a.doomed)
}
if surgeErr != nil {
logf.FromContext(ctx).Error(surgeErr, "Failed to surge workload", "kind", target.Kind(), "name", target.GetName())
}

previousPhase := status.Phase
Expand All @@ -323,6 +326,11 @@ func (r *UnderstudyReconciler) reconcileDoomed(ctx context.Context, cr *appsv1al
}
r.eventOnce(cr, &status, corev1.EventTypeWarning, "HoldRelaxed", "Relax",
fmt.Sprintf("eviction hold for %s relaxed (%s); eviction can proceed without a ready understudy", target.GetName(), relaxReason))
case surgeErr != nil:
status.Phase = appsv1alpha1.PhaseSurging
message := fmt.Sprintf("cannot surge %s: %v", target.GetName(), surgeErr)
setCond(&status, cr, appsv1alpha1.ConditionSurgeReady, metav1.ConditionFalse, "SurgeFailed", message)
r.eventOnce(cr, &status, corev1.EventTypeWarning, "SurgeFailed", "Surge", message)
default:
status.Phase = appsv1alpha1.PhaseSurging
setCond(&status, cr, appsv1alpha1.ConditionSurgeReady, metav1.ConditionFalse,
Expand Down
73 changes: 73 additions & 0 deletions internal/controller/understudy_controller_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"

admissionv1 "k8s.io/api/admissionregistration/v1"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
policyv1 "k8s.io/api/policy/v1"
Expand Down Expand Up @@ -184,6 +185,56 @@ func taintNode(node *corev1.Node, key string) {
}).Should(Succeed())
}

// freezeReplicas makes the API server refuse every replica change to the
// fixture's Deployment, as a policy engine or a missing permission would.
func freezeReplicas(f *fixture) {
policy := &admissionv1.ValidatingAdmissionPolicy{
ObjectMeta: metav1.ObjectMeta{Name: f.ns + "-freeze-replicas"},
Spec: admissionv1.ValidatingAdmissionPolicySpec{
FailurePolicy: ptr.To(admissionv1.Fail),
MatchConstraints: &admissionv1.MatchResources{
ResourceRules: []admissionv1.NamedRuleWithOperations{{
RuleWithOperations: admissionv1.RuleWithOperations{
Operations: []admissionv1.OperationType{admissionv1.Update},
Rule: admissionv1.Rule{
APIGroups: []string{"apps"},
APIVersions: []string{"v1"},
Resources: []string{"deployments"},
},
},
}},
},
Validations: []admissionv1.Validation{{
Expression: "object.spec.replicas == oldObject.spec.replicas",
Message: "replicas are frozen",
}},
},
}
binding := &admissionv1.ValidatingAdmissionPolicyBinding{
ObjectMeta: metav1.ObjectMeta{Name: policy.Name},
Spec: admissionv1.ValidatingAdmissionPolicyBindingSpec{
PolicyName: policy.Name,
ValidationActions: []admissionv1.ValidationAction{admissionv1.Deny},
MatchResources: &admissionv1.MatchResources{
NamespaceSelector: &metav1.LabelSelector{
MatchLabels: map[string]string{corev1.LabelMetadataName: f.ns},
},
},
},
}
Expect(k8sClient.Create(ctx, policy)).To(Succeed())
Expect(k8sClient.Create(ctx, binding)).To(Succeed())
DeferCleanup(func() {
Expect(k8sClient.Delete(ctx, binding)).To(Succeed())
Expect(k8sClient.Delete(ctx, policy)).To(Succeed())
})
Eventually(func() bool {
dep := getDeployment(f)
dep.Spec.Replicas = ptr.To(*dep.Spec.Replicas + 1)
return apierrors.IsInvalid(k8sClient.Update(ctx, dep, client.DryRunAll))
}).Should(BeTrue(), "the API server should enforce the policy before the test relies on it")
}

func getDeployment(f *fixture) *appsv1.Deployment {
dep := &appsv1.Deployment{}
ExpectWithOffset(1, k8sClient.Get(ctx, client.ObjectKeyFromObject(f.dep), dep)).To(Succeed())
Expand Down Expand Up @@ -374,6 +425,28 @@ var _ = Describe("Understudy controller", func() {
Eventually(func() error { return evict(f.pod, false) }).Should(Succeed())
})

It("fires the never-wedge TTL even when the surge itself is refused", func() {
f := newFixture(func(cr *appsv1alpha1.Understudy) {
cr.Spec.ReadinessDeadlineSeconds = ptr.To(int32(3))
})
freezeReplicas(f)
cordonNode(f.node)

By("recording the hold while the surge is refused")
Eventually(func() string {
if cond := findCond(getCR(f), appsv1alpha1.ConditionSurgeReady); cond != nil {
return cond.Reason
}
return ""
}).Should(Equal("SurgeFailed"))
Expect(getCR(f).Status.BlockedSince).NotTo(BeNil())

By("relaxing at the readiness deadline and admitting the eviction")
Eventually(func() appsv1alpha1.Phase { return getCR(f).Status.Phase }).Should(Equal(appsv1alpha1.PhaseRelaxed))
Eventually(func() error { return evict(f.pod, false) }).Should(Succeed())
Expect(*getDeployment(f).Spec.Replicas).To(Equal(int32(1)))
})

It("releases immediately when an involuntary deadline is shorter than minSurgeTimeSeconds", func() {
f := newFixture(func(cr *appsv1alpha1.Understudy) {
cr.Spec.MinSurgeTimeSeconds = ptr.To(int32(300))
Expand Down
Loading