Skip to content

Commit 987f3af

Browse files
L3n41cL3n41c
authored
[CASCL-1386] (5/11) Scale down the cluster-autoscaler before eviction (#3164)
* [CASCL-1386] Add evict-legacy-nodes command skeleton Part of a stack splitting #3026 (too large to review in one piece) into small pieces that each build and pass tests on their own. The command is fully functional only once the whole stack lands. * [CASCL-1386] Implement evict-legacy-nodes execution plan building Part of a stack splitting #3026 (too large to review in one piece) into small pieces that each build and pass tests on their own. The command is fully functional only once the whole stack lands. * [CASCL-1386] Implement evict-legacy-nodes plan display and confirmation Part of a stack splitting #3026 (too large to review in one piece) into small pieces that each build and pass tests on their own. The command is fully functional only once the whole stack lands. * [CASCL-1386] Implement evict-legacy-nodes preflight warnings Part of a stack splitting #3026 (too large to review in one piece) into small pieces that each build and pass tests on their own. The command is fully functional only once the whole stack lands. * [CASCL-1386] Implement cluster-autoscaler scale-down for evict-legacy-nodes Part of a stack splitting #3026 (too large to review in one piece) into small pieces that each build and pass tests on their own. The command is fully functional only once the whole stack lands.
1 parent 306e23a commit 987f3af

2 files changed

Lines changed: 324 additions & 1 deletion

File tree

‎cmd/kubectl-datadog/autoscaling/cluster/evict/clusterautoscaler.go‎

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,61 @@ package evict
22

33
import (
44
"context"
5+
"fmt"
6+
"log"
7+
"time"
58

9+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
10+
"k8s.io/apimachinery/pkg/util/wait"
611
"k8s.io/client-go/kubernetes"
12+
"k8s.io/client-go/util/retry"
713

814
"github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/clusterinfo"
915
)
1016

17+
// Poll cadence for confirming the cluster-autoscaler Deployment has actually
18+
// reached 0 replicas. Declared as vars (not consts) so tests can shrink them.
19+
var (
20+
caScaleDownPollInterval = 2 * time.Second
21+
caScaleDownPollTimeout = 2 * time.Minute
22+
)
23+
1124
func scaleDownClusterAutoscaler(ctx context.Context, clientset kubernetes.Interface, ca clusterinfo.ClusterAutoscaler, dryRun bool) error {
12-
panic("TODO: scaleDownClusterAutoscaler — implemented in PR https://github.com/DataDog/datadog-operator/pull/3164")
25+
if !ca.Present {
26+
return nil
27+
}
28+
if dryRun {
29+
log.Printf("[dry-run] would scale Deployment %s/%s to 0 replicas", ca.Namespace, ca.Name)
30+
return nil
31+
}
32+
33+
if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
34+
scale, err := clientset.AppsV1().Deployments(ca.Namespace).GetScale(ctx, ca.Name, metav1.GetOptions{})
35+
if err != nil {
36+
return fmt.Errorf("failed to get Deployment scale %s/%s: %w", ca.Namespace, ca.Name, err)
37+
}
38+
if scale.Spec.Replicas == 0 {
39+
return nil
40+
}
41+
scale.Spec.Replicas = 0
42+
if _, err = clientset.AppsV1().Deployments(ca.Namespace).UpdateScale(ctx, ca.Name, scale, metav1.UpdateOptions{}); err != nil {
43+
return err
44+
}
45+
log.Printf("Requested scale-down of Deployment %s/%s to 0 replicas.", ca.Namespace, ca.Name)
46+
return nil
47+
}); err != nil {
48+
return err
49+
}
50+
51+
if err := wait.PollUntilContextTimeout(ctx, caScaleDownPollInterval, caScaleDownPollTimeout, true, func(ctx context.Context) (bool, error) {
52+
if scale, err := clientset.AppsV1().Deployments(ca.Namespace).GetScale(ctx, ca.Name, metav1.GetOptions{}); err != nil {
53+
return false, fmt.Errorf("failed to get Deployment scale %s/%s: %w", ca.Namespace, ca.Name, err)
54+
} else {
55+
return scale.Status.Replicas == 0, nil
56+
}
57+
}); err != nil {
58+
return fmt.Errorf("cluster-autoscaler Deployment %s/%s did not reach 0 replicas: %w", ca.Namespace, ca.Name, err)
59+
}
60+
log.Printf("Cluster-autoscaler Deployment %s/%s is now at 0 replicas.", ca.Namespace, ca.Name)
61+
return nil
1362
}

‎cmd/kubectl-datadog/autoscaling/cluster/evict/clusterautoscaler_test.go‎

Lines changed: 274 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,274 @@
1+
package evict
2+
3+
import (
4+
"errors"
5+
"testing"
6+
"time"
7+
8+
"github.com/stretchr/testify/assert"
9+
"github.com/stretchr/testify/require"
10+
appsv1 "k8s.io/api/apps/v1"
11+
autoscalingv1 "k8s.io/api/autoscaling/v1"
12+
apierrors "k8s.io/apimachinery/pkg/api/errors"
13+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14+
"k8s.io/apimachinery/pkg/runtime"
15+
"k8s.io/apimachinery/pkg/runtime/schema"
16+
"k8s.io/client-go/kubernetes/fake"
17+
clienttesting "k8s.io/client-go/testing"
18+
"k8s.io/utils/ptr"
19+
20+
"github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/clusterinfo"
21+
)
22+
23+
// deploymentsGVR is the GroupVersionResource used to talk to the fake
24+
// clientset's object tracker for Deployment objects.
25+
var deploymentsGVR = schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
26+
27+
// scaleReactor wires GetScale/UpdateScale on a fake clientset so the test
28+
// observes the same read-modify-write surface as production. The fake doesn't
29+
// natively support the Scale subresource, so we synthesize Scale objects from
30+
// the underlying Deployment. The reactor talks to the underlying tracker
31+
// directly (rather than via client.AppsV1()) because the fake holds an
32+
// internal mutex while a reactor runs — calling back through the typed client
33+
// would deadlock.
34+
func scaleReactor(t *testing.T, client *fake.Clientset, conflictFirstUpdate bool) *int {
35+
t.Helper()
36+
updateCalls := 0
37+
tracker := client.Tracker()
38+
client.PrependReactor("get", "deployments", func(action clienttesting.Action) (bool, runtime.Object, error) {
39+
ga, ok := action.(clienttesting.GetAction)
40+
if !ok || ga.GetSubresource() != "scale" {
41+
return false, nil, nil
42+
}
43+
obj, err := tracker.Get(deploymentsGVR, ga.GetNamespace(), ga.GetName())
44+
if err != nil {
45+
return true, nil, err
46+
}
47+
dep := obj.(*appsv1.Deployment)
48+
var replicas int32
49+
if dep.Spec.Replicas != nil {
50+
replicas = *dep.Spec.Replicas
51+
}
52+
// Status.Replicas mirrors the observed pod count: the fake has no pod
53+
// lifecycle, so we model it as instantly converging to the desired
54+
// spec. This is what scaleDownClusterAutoscaler polls after the
55+
// scale-down to confirm the autoscaler has actually stopped.
56+
return true, &autoscalingv1.Scale{
57+
ObjectMeta: metav1.ObjectMeta{Name: dep.Name, Namespace: dep.Namespace, ResourceVersion: dep.ResourceVersion},
58+
Spec: autoscalingv1.ScaleSpec{Replicas: replicas},
59+
Status: autoscalingv1.ScaleStatus{Replicas: replicas},
60+
}, nil
61+
})
62+
client.PrependReactor("update", "deployments", func(action clienttesting.Action) (bool, runtime.Object, error) {
63+
ua, ok := action.(clienttesting.UpdateAction)
64+
if !ok || ua.GetSubresource() != "scale" {
65+
return false, nil, nil
66+
}
67+
scale := ua.GetObject().(*autoscalingv1.Scale)
68+
updateCalls++
69+
if conflictFirstUpdate && updateCalls == 1 {
70+
return true, nil, apierrors.NewConflict(
71+
schema.GroupResource{Group: "apps", Resource: "deployments"},
72+
scale.Name,
73+
errors.New("forced conflict"),
74+
)
75+
}
76+
obj, err := tracker.Get(deploymentsGVR, ua.GetNamespace(), scale.Name)
77+
if err != nil {
78+
return true, nil, err
79+
}
80+
dep := obj.(*appsv1.Deployment).DeepCopy()
81+
dep.Spec.Replicas = ptr.To(scale.Spec.Replicas)
82+
if err := tracker.Update(deploymentsGVR, dep, ua.GetNamespace()); err != nil {
83+
return true, nil, err
84+
}
85+
return true, scale, nil
86+
})
87+
return &updateCalls
88+
}
89+
90+
func TestScaleDownClusterAutoscaler(t *testing.T) {
91+
caDep := func(replicas int32) *appsv1.Deployment {
92+
return &appsv1.Deployment{
93+
ObjectMeta: metav1.ObjectMeta{Name: "cluster-autoscaler", Namespace: "kube-system"},
94+
Spec: appsv1.DeploymentSpec{Replicas: ptr.To(replicas)},
95+
}
96+
}
97+
caPresent := clusterinfo.ClusterAutoscaler{Present: true, Namespace: "kube-system", Name: "cluster-autoscaler"}
98+
99+
for _, tc := range []struct {
100+
name string
101+
// dep, when non-nil, is pre-loaded into the fake clientset.
102+
dep *appsv1.Deployment
103+
// ca is what Run would have passed in from clusterinfo.Classify.
104+
ca clusterinfo.ClusterAutoscaler
105+
// conflictFirstUpdate forces the scaleReactor to return Conflict on
106+
// the first UpdateScale call so RetryOnConflict has to refetch.
107+
conflictFirstUpdate bool
108+
dryRun bool
109+
// wantUpdateCalls is the minimum number of UpdateScale invocations
110+
// the test expects (>= because retries may happen).
111+
wantUpdateCalls int
112+
// wantReplicas, when non-nil, is the final value of the Deployment's
113+
// Spec.Replicas after the call.
114+
wantReplicas *int32
115+
}{
116+
{
117+
name: "CA absent is a no-op",
118+
ca: clusterinfo.ClusterAutoscaler{Present: false},
119+
},
120+
{
121+
name: "replicas already 0 skips Update",
122+
dep: caDep(0),
123+
ca: caPresent,
124+
wantUpdateCalls: 0,
125+
wantReplicas: ptr.To(int32(0)),
126+
},
127+
{
128+
name: "scales from 3 to 0",
129+
dep: caDep(3),
130+
ca: caPresent,
131+
wantUpdateCalls: 1,
132+
wantReplicas: ptr.To(int32(0)),
133+
},
134+
{
135+
name: "retries on Conflict",
136+
dep: caDep(2),
137+
ca: caPresent,
138+
conflictFirstUpdate: true,
139+
wantUpdateCalls: 2,
140+
wantReplicas: ptr.To(int32(0)),
141+
},
142+
{
143+
name: "dry-run touches nothing",
144+
dep: caDep(3),
145+
ca: caPresent,
146+
dryRun: true,
147+
wantUpdateCalls: 0,
148+
wantReplicas: ptr.To(int32(3)),
149+
},
150+
} {
151+
t.Run(tc.name, func(t *testing.T) {
152+
var objs []runtime.Object
153+
if tc.dep != nil {
154+
objs = append(objs, tc.dep)
155+
}
156+
client := fake.NewClientset(objs...)
157+
calls := scaleReactor(t, client, tc.conflictFirstUpdate)
158+
159+
require.NoError(t, scaleDownClusterAutoscaler(t.Context(), client, tc.ca, tc.dryRun))
160+
161+
assert.GreaterOrEqual(t, *calls, tc.wantUpdateCalls, "minimum UpdateScale calls")
162+
if tc.wantReplicas == nil {
163+
return
164+
}
165+
got, err := client.AppsV1().Deployments(tc.ca.Namespace).Get(t.Context(), tc.ca.Name, metav1.GetOptions{})
166+
require.NoError(t, err)
167+
require.NotNil(t, got.Spec.Replicas)
168+
assert.Equal(t, *tc.wantReplicas, *got.Spec.Replicas)
169+
})
170+
}
171+
}
172+
173+
// withFastPoll shrinks the scale-down poll cadence so the wait-loop tests run
174+
// in milliseconds. It returns a restore func to defer.
175+
func withFastPoll(t *testing.T) {
176+
t.Helper()
177+
origInterval, origTimeout := caScaleDownPollInterval, caScaleDownPollTimeout
178+
caScaleDownPollInterval = time.Millisecond
179+
caScaleDownPollTimeout = 100 * time.Millisecond
180+
t.Cleanup(func() {
181+
caScaleDownPollInterval = origInterval
182+
caScaleDownPollTimeout = origTimeout
183+
})
184+
}
185+
186+
// lingeringStatusReactor installs a Scale subresource that reports Spec
187+
// faithfully but holds Status.Replicas at a non-zero "lingering" value for the
188+
// first lingerPolls scale-reads taken after the Deployment is scaled to 0 —
189+
// modelling cluster-autoscaler pods that keep running through their termination
190+
// grace period. zeroPolls is incremented every time Status.Replicas is reported
191+
// as 0, so the caller can assert the wait actually polled. When lingerForever
192+
// is set, Status never reaches 0 (drives the timeout path).
193+
func lingeringStatusReactor(t *testing.T, client *fake.Clientset, lingerPolls int, lingerForever bool) *int {
194+
t.Helper()
195+
tracker := client.Tracker()
196+
remaining := lingerPolls
197+
zeroPolls := 0
198+
client.PrependReactor("get", "deployments", func(action clienttesting.Action) (bool, runtime.Object, error) {
199+
ga, ok := action.(clienttesting.GetAction)
200+
if !ok || ga.GetSubresource() != "scale" {
201+
return false, nil, nil
202+
}
203+
obj, err := tracker.Get(deploymentsGVR, ga.GetNamespace(), ga.GetName())
204+
if err != nil {
205+
return true, nil, err
206+
}
207+
dep := obj.(*appsv1.Deployment)
208+
var spec int32
209+
if dep.Spec.Replicas != nil {
210+
spec = *dep.Spec.Replicas
211+
}
212+
status := spec
213+
if spec == 0 {
214+
if lingerForever || remaining > 0 {
215+
status = 1
216+
remaining--
217+
} else {
218+
zeroPolls++
219+
}
220+
}
221+
return true, &autoscalingv1.Scale{
222+
ObjectMeta: metav1.ObjectMeta{Name: dep.Name, Namespace: dep.Namespace, ResourceVersion: dep.ResourceVersion},
223+
Spec: autoscalingv1.ScaleSpec{Replicas: spec},
224+
Status: autoscalingv1.ScaleStatus{Replicas: status},
225+
}, nil
226+
})
227+
client.PrependReactor("update", "deployments", func(action clienttesting.Action) (bool, runtime.Object, error) {
228+
ua, ok := action.(clienttesting.UpdateAction)
229+
if !ok || ua.GetSubresource() != "scale" {
230+
return false, nil, nil
231+
}
232+
scale := ua.GetObject().(*autoscalingv1.Scale)
233+
obj, err := tracker.Get(deploymentsGVR, ua.GetNamespace(), scale.Name)
234+
if err != nil {
235+
return true, nil, err
236+
}
237+
dep := obj.(*appsv1.Deployment).DeepCopy()
238+
dep.Spec.Replicas = ptr.To(scale.Spec.Replicas)
239+
if err := tracker.Update(deploymentsGVR, dep, ua.GetNamespace()); err != nil {
240+
return true, nil, err
241+
}
242+
return true, scale, nil
243+
})
244+
return &zeroPolls
245+
}
246+
247+
func TestScaleDownClusterAutoscalerWaitsForReplicasZero(t *testing.T) {
248+
caPresent := clusterinfo.ClusterAutoscaler{Present: true, Namespace: "kube-system", Name: "cluster-autoscaler"}
249+
caDep := func() *appsv1.Deployment {
250+
return &appsv1.Deployment{
251+
ObjectMeta: metav1.ObjectMeta{Name: caPresent.Name, Namespace: caPresent.Namespace},
252+
Spec: appsv1.DeploymentSpec{Replicas: ptr.To(int32(3))},
253+
}
254+
}
255+
256+
t.Run("blocks until Status.Replicas reaches 0", func(t *testing.T) {
257+
withFastPoll(t)
258+
client := fake.NewClientset(caDep())
259+
zeroPolls := lingeringStatusReactor(t, client, 2, false)
260+
261+
require.NoError(t, scaleDownClusterAutoscaler(t.Context(), client, caPresent, false))
262+
assert.Positive(t, *zeroPolls, "wait loop should have observed Status.Replicas == 0 before returning")
263+
})
264+
265+
t.Run("times out when Status.Replicas never reaches 0", func(t *testing.T) {
266+
withFastPoll(t)
267+
client := fake.NewClientset(caDep())
268+
lingeringStatusReactor(t, client, 0, true)
269+
270+
err := scaleDownClusterAutoscaler(t.Context(), client, caPresent, false)
271+
require.Error(t, err)
272+
assert.Contains(t, err.Error(), "did not reach 0 replicas")
273+
})
274+
}

0 commit comments

Comments
 (0)