Skip to content

Commit 071f5d7

Browse files
committed
[CASCL-1386] Create temporary PodDisruptionBudgets during eviction
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 72d803c commit 071f5d7

3 files changed

Lines changed: 755 additions & 2 deletions

File tree

cmd/kubectl-datadog/autoscaling/cluster/evict/pdb.go

Lines changed: 322 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,335 @@ package evict
22

33
import (
44
"context"
5+
"errors"
6+
"fmt"
7+
"log"
8+
"reflect"
9+
"slices"
10+
"strings"
511

12+
"github.com/samber/lo"
13+
appsv1 "k8s.io/api/apps/v1"
14+
corev1 "k8s.io/api/core/v1"
15+
policyv1 "k8s.io/api/policy/v1"
16+
apierrors "k8s.io/apimachinery/pkg/api/errors"
17+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
18+
"k8s.io/apimachinery/pkg/runtime"
19+
"k8s.io/apimachinery/pkg/util/intstr"
20+
"k8s.io/apimachinery/pkg/util/validation"
621
"k8s.io/client-go/kubernetes"
22+
"k8s.io/client-go/tools/pager"
723
"sigs.k8s.io/controller-runtime/pkg/client"
24+
25+
commonk8s "github.com/DataDog/datadog-operator/cmd/kubectl-datadog/autoscaling/cluster/common/k8s"
26+
)
27+
28+
// Label keys used to mark PodDisruptionBudgets created by this command. Both
29+
// must be present for cleanup to consider the PDB ours — this makes the
30+
// cleanup safe against accidentally removing a user PDB with a colliding name.
31+
const (
32+
pdbManagedByLabelKey = "app.kubernetes.io/managed-by"
33+
pdbManagedByLabelValue = "kubectl-datadog"
34+
pdbTempLabelKey = "autoscaling.datadoghq.com/temporary-pdb"
35+
pdbTempLabelValue = "true"
36+
// pdbNameSuffix is appended to the controller name to form the temp PDB
37+
// name.
38+
pdbNameSuffix = "-evict-legacy"
839
)
940

1041
func ensureTempPDBs(ctx context.Context, clientset kubernetes.Interface, ctrlClient client.Client, targets []Target, dryRun bool) error {
11-
panic("TODO: ensureTempPDBs — implemented in PR https://github.com/DataDog/datadog-operator/pull/3178")
42+
allNodes := lo.FlatMap(targets, func(t Target, _ int) []string { return t.Nodes })
43+
nodeSet := lo.SliceToMap(allNodes, func(n string) (string, struct{}) { return n, struct{}{} })
44+
if len(nodeSet) == 0 {
45+
return nil
46+
}
47+
48+
controllers, err := discoverControllers(ctx, clientset, nodeSet)
49+
if err != nil {
50+
return fmt.Errorf("failed to discover controllers: %w", err)
51+
}
52+
if len(controllers) == 0 {
53+
return nil
54+
}
55+
56+
// Group controllers by namespace to amortize the per-namespace PDB list.
57+
byNamespace := make(map[string][]controllerInfo)
58+
for _, c := range controllers {
59+
byNamespace[c.Namespace] = append(byNamespace[c.Namespace], c)
60+
}
61+
62+
var errs []error
63+
for ns, ctrls := range byNamespace {
64+
existing, err := listNamespacePDBs(ctx, clientset, ns)
65+
if err != nil {
66+
errs = append(errs, fmt.Errorf("namespace %s: failed to list PDBs: %w", ns, err))
67+
continue
68+
}
69+
for _, c := range ctrls {
70+
if hasUserPDB(existing, c.Selector) {
71+
continue
72+
}
73+
if err := createTempPDB(ctx, ctrlClient, c, dryRun); err != nil {
74+
errs = append(errs, fmt.Errorf("controller %s/%s/%s: %w", c.Namespace, c.Kind, c.Name, err))
75+
}
76+
}
77+
}
78+
return errors.Join(errs...)
1279
}
1380

1481
func cleanupTempPDBs(ctx context.Context, ctrlClient client.Client, dryRun bool) error {
15-
panic("TODO: cleanupTempPDBs — implemented in PR https://github.com/DataDog/datadog-operator/pull/3178")
82+
list := &policyv1.PodDisruptionBudgetList{}
83+
if err := ctrlClient.List(ctx, list, client.MatchingLabels{
84+
pdbManagedByLabelKey: pdbManagedByLabelValue,
85+
pdbTempLabelKey: pdbTempLabelValue,
86+
}); err != nil {
87+
return fmt.Errorf("failed to list temporary PDBs: %w", err)
88+
}
89+
if len(list.Items) == 0 {
90+
return nil
91+
}
92+
var errs []error
93+
for i := range list.Items {
94+
pdb := &list.Items[i]
95+
if dryRun {
96+
log.Printf("[dry-run] would delete PDB %s/%s", pdb.Namespace, pdb.Name)
97+
continue
98+
}
99+
if err := commonk8s.Delete(ctx, ctrlClient, pdb); err != nil {
100+
errs = append(errs, fmt.Errorf("PDB %s/%s: %w", pdb.Namespace, pdb.Name, err))
101+
}
102+
}
103+
return errors.Join(errs...)
104+
}
105+
106+
// controllerKey identifies a top-level controller that owns evictable pods on
107+
// our target nodes. It is the dedup key in the seen map and the identity half
108+
// of controllerInfo.
109+
type controllerKey struct {
110+
Namespace string
111+
Kind string // "Deployment", "StatefulSet", "ReplicaSet"
112+
Name string
113+
}
114+
115+
// controllerInfo is a controllerKey plus the controller's pod selector — what a
116+
// PDB would match on.
117+
type controllerInfo struct {
118+
controllerKey
119+
Selector *metav1.LabelSelector
120+
}
121+
122+
// discoverControllers lists every Pod cluster-wide once (paginated) and, for
123+
// each Pod scheduled on one of the target nodes, resolves the top-level
124+
// controller. Listing once and filtering client-side avoids the N-API-calls
125+
// problem of doing one List per node, which on a large legacy fleet would
126+
// dominate the command's wall-clock time. The resulting slice contains each
127+
// controller at most once.
128+
func discoverControllers(ctx context.Context, clientset kubernetes.Interface, nodeSet map[string]struct{}) ([]controllerInfo, error) {
129+
seen := make(map[controllerKey]*metav1.LabelSelector)
130+
depCache := make(map[client.ObjectKey]*appsv1.Deployment)
131+
rsCache := make(map[client.ObjectKey]*appsv1.ReplicaSet)
132+
stsCache := make(map[client.ObjectKey]*appsv1.StatefulSet)
133+
134+
p := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
135+
return clientset.CoreV1().Pods(metav1.NamespaceAll).List(ctx, opts)
136+
})
137+
if err := p.EachListItem(ctx, metav1.ListOptions{}, func(obj runtime.Object) error {
138+
pod := obj.(*corev1.Pod)
139+
if _, onTarget := nodeSet[pod.Spec.NodeName]; !onTarget {
140+
return nil
141+
}
142+
if shouldSkipEviction(pod) {
143+
return nil
144+
}
145+
info, err := resolveTopLevelController(ctx, clientset, pod, depCache, rsCache, stsCache)
146+
if err != nil {
147+
log.Printf("Warning: cannot resolve controller for pod %s/%s: %v", pod.Namespace, pod.Name, err)
148+
return nil
149+
}
150+
if info == nil {
151+
return nil
152+
}
153+
seen[info.controllerKey] = info.Selector
154+
return nil
155+
}); err != nil {
156+
return nil, fmt.Errorf("list pods: %w", err)
157+
}
158+
159+
return lo.MapToSlice(seen, func(key controllerKey, selector *metav1.LabelSelector) controllerInfo {
160+
return controllerInfo{controllerKey: key, Selector: selector}
161+
}), nil
162+
}
163+
164+
// resolveTopLevelController walks a Pod's owner chain up to the workload
165+
// controller (Deployment > ReplicaSet > Pod, StatefulSet > Pod). Returns nil
166+
// for Pods whose top-level owner is not a stable workload — Jobs (TTL-managed),
167+
// DaemonSets (already skipped at eviction), or bare Pods.
168+
func resolveTopLevelController(
169+
ctx context.Context,
170+
clientset kubernetes.Interface,
171+
pod *corev1.Pod,
172+
depCache map[client.ObjectKey]*appsv1.Deployment,
173+
rsCache map[client.ObjectKey]*appsv1.ReplicaSet,
174+
stsCache map[client.ObjectKey]*appsv1.StatefulSet,
175+
) (*controllerInfo, error) {
176+
owner := metav1.GetControllerOf(pod)
177+
if owner == nil {
178+
return nil, nil
179+
}
180+
switch owner.Kind {
181+
case "ReplicaSet":
182+
rs, err := getCached(ctx, rsCache, pod.Namespace, owner.Name, clientset.AppsV1().ReplicaSets(pod.Namespace).Get)
183+
if err != nil {
184+
return nil, err
185+
}
186+
rsOwner := metav1.GetControllerOf(rs)
187+
if rsOwner != nil && rsOwner.Kind == "Deployment" {
188+
dep, err := getCached(ctx, depCache, pod.Namespace, rsOwner.Name, clientset.AppsV1().Deployments(pod.Namespace).Get)
189+
if err != nil {
190+
return nil, err
191+
}
192+
return &controllerInfo{
193+
controllerKey: controllerKey{Namespace: pod.Namespace, Kind: "Deployment", Name: dep.Name},
194+
Selector: dep.Spec.Selector,
195+
}, nil
196+
}
197+
return &controllerInfo{
198+
controllerKey: controllerKey{Namespace: pod.Namespace, Kind: "ReplicaSet", Name: rs.Name},
199+
Selector: rs.Spec.Selector,
200+
}, nil
201+
case "StatefulSet":
202+
sts, err := getCached(ctx, stsCache, pod.Namespace, owner.Name, clientset.AppsV1().StatefulSets(pod.Namespace).Get)
203+
if err != nil {
204+
return nil, err
205+
}
206+
return &controllerInfo{
207+
controllerKey: controllerKey{Namespace: pod.Namespace, Kind: "StatefulSet", Name: sts.Name},
208+
Selector: sts.Spec.Selector,
209+
}, nil
210+
default:
211+
// DaemonSet (skipped before reaching here), Job (TTL), CronJob,
212+
// custom controllers — none get a temporary PDB.
213+
return nil, nil
214+
}
215+
}
216+
217+
// getCached returns the object identified by (ns, name) from cache, fetching it
218+
// via get — the namespace-bound typed clientset accessor, e.g.
219+
// clientset.AppsV1().ReplicaSets(ns).Get — and populating the cache on a miss.
220+
// T is inferred from the cache's value type, collapsing the per-kind getters
221+
// into a single generic lookup.
222+
func getCached[T any](
223+
ctx context.Context,
224+
cache map[client.ObjectKey]*T,
225+
ns, name string,
226+
get func(ctx context.Context, name string, opts metav1.GetOptions) (*T, error),
227+
) (*T, error) {
228+
key := client.ObjectKey{Namespace: ns, Name: name}
229+
if obj, ok := cache[key]; ok {
230+
return obj, nil
231+
}
232+
obj, err := get(ctx, name, metav1.GetOptions{})
233+
if err != nil {
234+
return nil, err
235+
}
236+
cache[key] = obj
237+
return obj, nil
238+
}
239+
240+
// listNamespacePDBs returns every PDB in the namespace. Used to detect
241+
// pre-existing user PDBs covering a controller we'd otherwise PDB-protect.
242+
func listNamespacePDBs(ctx context.Context, clientset kubernetes.Interface, namespace string) ([]policyv1.PodDisruptionBudget, error) {
243+
list, err := clientset.PolicyV1().PodDisruptionBudgets(namespace).List(ctx, metav1.ListOptions{})
244+
if err != nil {
245+
return nil, err
246+
}
247+
return list.Items, nil
248+
}
249+
250+
// hasUserPDB returns true when an existing non-temporary PDB has the same
251+
// selector as the controller's pod selector. This is a conservative
252+
// equality check: a broader user PDB will NOT be detected, and we'll create
253+
// our own. Eviction will then respect the most restrictive of the two,
254+
// preserving the user's intent.
255+
func hasUserPDB(existing []policyv1.PodDisruptionBudget, controllerSelector *metav1.LabelSelector) bool {
256+
if controllerSelector == nil {
257+
return false
258+
}
259+
return slices.ContainsFunc(existing, func(pdb policyv1.PodDisruptionBudget) bool {
260+
return !isTemporaryPDB(&pdb) && reflect.DeepEqual(pdb.Spec.Selector, controllerSelector)
261+
})
262+
}
263+
264+
func isTemporaryPDB(pdb *policyv1.PodDisruptionBudget) bool {
265+
return pdb.Labels[pdbManagedByLabelKey] == pdbManagedByLabelValue &&
266+
pdb.Labels[pdbTempLabelKey] == pdbTempLabelValue
267+
}
268+
269+
// createTempPDB writes (or no-ops if our PDB already exists) a temporary
270+
// PodDisruptionBudget with maxUnavailable: 1. Existing PDBs that aren't ours
271+
// at the same name are left alone with a logged warning — that's a name
272+
// collision the user must resolve.
273+
func createTempPDB(ctx context.Context, ctrlClient client.Client, c controllerInfo, dryRun bool) error {
274+
name := tempPDBName(c.Kind, c.Name)
275+
if dryRun {
276+
log.Printf("[dry-run] would create PDB %s/%s (maxUnavailable: 1, selector: %s)", c.Namespace, name, formatSelector(c.Selector))
277+
return nil
278+
}
279+
280+
existing := &policyv1.PodDisruptionBudget{}
281+
err := ctrlClient.Get(ctx, client.ObjectKey{Namespace: c.Namespace, Name: name}, existing)
282+
switch {
283+
case err == nil:
284+
if !isTemporaryPDB(existing) {
285+
log.Printf("Warning: PDB %s/%s exists but is not labelled as temporary; leaving it untouched", c.Namespace, name)
286+
return nil
287+
}
288+
// Our PDB from a previous (possibly crashed) run. Leave as-is; the
289+
// cleanup step will remove it at the end of the current run.
290+
return nil
291+
case !apierrors.IsNotFound(err):
292+
return fmt.Errorf("failed to get PDB %s/%s: %w", c.Namespace, name, err)
293+
}
294+
295+
maxUnavailable := intstr.FromInt(1)
296+
pdb := &policyv1.PodDisruptionBudget{
297+
TypeMeta: metav1.TypeMeta{APIVersion: "policy/v1", Kind: "PodDisruptionBudget"},
298+
ObjectMeta: metav1.ObjectMeta{
299+
Name: name,
300+
Namespace: c.Namespace,
301+
Labels: map[string]string{
302+
pdbManagedByLabelKey: pdbManagedByLabelValue,
303+
pdbTempLabelKey: pdbTempLabelValue,
304+
},
305+
},
306+
Spec: policyv1.PodDisruptionBudgetSpec{
307+
Selector: c.Selector.DeepCopy(),
308+
MaxUnavailable: &maxUnavailable,
309+
},
310+
}
311+
if err := ctrlClient.Create(ctx, pdb); err != nil {
312+
return fmt.Errorf("failed to create PDB %s/%s: %w", c.Namespace, name, err)
313+
}
314+
log.Printf("Created temporary PDB %s/%s for %s/%s (maxUnavailable: 1).", c.Namespace, name, c.Kind, c.Name)
315+
return nil
316+
}
317+
318+
// tempPDBName builds a DNS-label-safe PDB name. Long controller names are
319+
// truncated so the final name (including the suffix) fits the 63-char limit.
320+
func tempPDBName(kind, controllerName string) string {
321+
prefix := strings.ToLower(kind) + "-" + controllerName
322+
if len(prefix)+len(pdbNameSuffix) > validation.DNS1123LabelMaxLength {
323+
prefix = prefix[:validation.DNS1123LabelMaxLength-len(pdbNameSuffix)]
324+
}
325+
return prefix + pdbNameSuffix
326+
}
327+
328+
func formatSelector(s *metav1.LabelSelector) string {
329+
if s == nil {
330+
return "<nil>"
331+
}
332+
parts := lo.MapToSlice(s.MatchLabels, func(k, v string) string {
333+
return k + "=" + v
334+
})
335+
return strings.Join(parts, ",")
16336
}

0 commit comments

Comments
 (0)