17156871947

Committed 22 Aug 2025 01:39PM UTC coverage: 78.824% (-0.2%) from 79.059%

Build # 17156871947

Build Type

Pull #72

github

Committed by

mikkeloscar

Commit Message

Add dependabot config

Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de>

Pull Request Pull Request #72: Add dependabot config

Run Details

335 of 425 relevant lines covered (78.82%)

20.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.2

/controller.go

package main

import (
        "context"
        "crypto/sha1"
        "encoding/hex"
        "time"

        log "github.com/sirupsen/logrus"
        v1 "k8s.io/api/core/v1"
        pv1 "k8s.io/api/policy/v1"
        "k8s.io/apimachinery/pkg/api/equality"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/util/intstr"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/util/retry"
)

const (
        heritageLabel               = "heritage"
        pdbController               = "pdb-controller"
        nonReadyTTLAnnotationName   = "pdb-controller.zalando.org/non-ready-ttl"
        nonReadySinceAnnotationName = "pdb-controller.zalando.org/non-ready-since"
        parentResourceHashLabel     = "parent-resource-hash"
)

var (
        ownerLabels = map[string]string{heritageLabel: pdbController}
)

// PDBController creates PodDisruptionBudgets for deployments and StatefulSets
// if missing.
type PDBController struct {
        kubernetes.Interface
        interval           time.Duration
        pdbNameSuffix      string
        nonReadyTTL        time.Duration
        parentResourceHash bool
        maxUnavailable     intstr.IntOrString
}

// NewPDBController initializes a new PDBController.
func NewPDBController(interval time.Duration, client kubernetes.Interface, pdbNameSuffix string, nonReadyTTL time.Duration, parentResourceHash bool, maxUnavailable intstr.IntOrString) *PDBController {
        return &PDBController{
                Interface:          client,
                interval:           interval,
                pdbNameSuffix:      pdbNameSuffix,
                nonReadyTTL:        nonReadyTTL,
                parentResourceHash: parentResourceHash,
                maxUnavailable:     maxUnavailable,
        }
}

// Run runs the controller loop until it receives a stop signal over the stop
// channel.
func (n *PDBController) Run(ctx context.Context) {
        for {
                log.Debug("Running main control loop.")
                err := n.runOnce(ctx)
                if err != nil {
                        log.Error(err)
                }

                select {
                case <-time.After(n.interval):
                case <-ctx.Done():
                        log.Info("Terminating main controller loop.")
                        return
                }
        }
}

// runOnce runs the main reconcilation loop of the controller.
func (n *PDBController) runOnce(ctx context.Context) error {
        allPDBs, err := n.PolicyV1().PodDisruptionBudgets(v1.NamespaceAll).List(ctx, metav1.ListOptions{})
        if err != nil {
                return err
        }

        managedPDBs, unmanagedPDBs := filterPDBs(allPDBs.Items)

        deployments, err := n.AppsV1().Deployments(v1.NamespaceAll).List(ctx, metav1.ListOptions{})
        if err != nil {
                return err
        }

        statefulSets, err := n.AppsV1().StatefulSets(v1.NamespaceAll).List(ctx, metav1.ListOptions{})
        if err != nil {
                return err
        }

        resources := make([]kubeResource, 0, len(deployments.Items)+len(statefulSets.Items))

        for _, d := range deployments.Items {
                // manually set Kind and APIVersion because of a bug in
                // client-go
                // https://github.com/kubernetes/client-go/issues/308
                d.Kind = "Deployment"
                d.APIVersion = "apps/v1"
                resources = append(resources, deployment{d})
        }

        for _, s := range statefulSets.Items {
                // manually set Kind and APIVersion because of a bug in
                // client-go
                // https://github.com/kubernetes/client-go/issues/308
                s.Kind = "StatefulSet"
                s.APIVersion = "apps/v1"
                resources = append(resources, statefulSet{s})
        }

        desiredPDBs := n.generateDesiredPDBs(resources, managedPDBs, unmanagedPDBs)
        n.reconcilePDBs(ctx, desiredPDBs, managedPDBs)
        return nil
}

func (n *PDBController) generateDesiredPDBs(resources []kubeResource, managedPDBs, unmanagedPDBs map[string]pv1.PodDisruptionBudget) map[string]pv1.PodDisruptionBudget {
        desiredPDBs := make(map[string]pv1.PodDisruptionBudget, len(managedPDBs))

        nonReadyTTL := time.Time{}
        if n.nonReadyTTL > 0 {
                nonReadyTTL = time.Now().UTC().Add(-n.nonReadyTTL)
        }

        for _, resource := range resources {
                matchedPDBs := getMatchedPDBs(resource.TemplateLabels(), unmanagedPDBs)

                // don't create managed PDB if there is already unmanaged and
                // matched PDBs
                if len(matchedPDBs) > 0 {
                        continue
                }

                // don't create PDB if the resource has 1 or less replicas
                if resource.Replicas() <= 1 {
                        continue
                }

                // ensure PDB if the resource has more than one replica and all
                // of them are ready
                if resource.StatusReadyReplicas() >= resource.Replicas() {
                        pdb := n.generatePDB(resource, time.Time{})
                        desiredPDBs[pdb.Namespace+"/"+pdb.Name] = pdb
                        continue
                }

                ownedPDBs := getOwnedPDBs(managedPDBs, resource)
                validPDBs := make([]pv1.PodDisruptionBudget, 0, len(ownedPDBs))
                // only consider valid PDBs. If they're invalid they'll be
                // recreated on the next iteration
                for _, pdb := range ownedPDBs {
                        if pdbSpecValid(pdb) {
                                validPDBs = append(validPDBs, pdb)
                        }
                }

                if len(validPDBs) > 0 {
                        // it's unlikely that we will have more than a single
                        // valid owned PDB. If we do simply pick the first one
                        // and check if it's still valid. Any other PDBs will
                        // automatically get dropped.
                        pdb := validPDBs[0]
                        if pdb.Annotations == nil {
                                pdb.Annotations = make(map[string]string)
                        }

                        ttl, err := overrideNonReadyTTL(resource.Annotations(), nonReadyTTL)
                        if err != nil {
                                log.Errorf("Failed to override PDB Delete TTL: %s", err)
                        }

                        var nonReadySince time.Time
                        if nonReadySinceStr, ok := pdb.Annotations[nonReadySinceAnnotationName]; ok {
                                nonReadySince, err = time.Parse(time.RFC3339, nonReadySinceStr)
                                if err != nil {
                                        log.Errorf("Failed to parse non-ready-since annotation '%s': %v", nonReadySinceStr, err)
                                }
                        }

                        if !nonReadySince.IsZero() {
                                if !ttl.IsZero() && nonReadySince.Before(ttl) {
                                        continue
                                }
                        } else {
                                nonReadySince = time.Now().UTC()
                        }

                        generatedPDB := n.generatePDB(resource, nonReadySince)
                        desiredPDBs[generatedPDB.Namespace+"/"+generatedPDB.Name] = generatedPDB
                }
        }

        return desiredPDBs
}

// mergeActualAndDesiredPDB takes the current definition of a PDB as it is in cluster and a PDB
// with our desired configurations and does a merge between them. We also return a boolean to tell the
// caller if any change actually had to be made to achieve our desired state or not
func mergeActualAndDesiredPDB(managedPDB, desiredPDB pv1.PodDisruptionBudget) (pv1.PodDisruptionBudget, bool) {

        needsUpdate := false

        // check if PDBs are equal an only update if not
        if !equality.Semantic.DeepEqual(managedPDB.Spec, desiredPDB.Spec) ||
                !equality.Semantic.DeepEqual(managedPDB.Labels, desiredPDB.Labels) ||
                !equality.Semantic.DeepEqual(managedPDB.Annotations, desiredPDB.Annotations) {
                managedPDB.Annotations = desiredPDB.Annotations
                managedPDB.Labels = desiredPDB.Labels
                managedPDB.Spec = desiredPDB.Spec

                needsUpdate = true
        }

        return managedPDB, needsUpdate
}

func (n *PDBController) reconcilePDBs(ctx context.Context, desiredPDBs, managedPDBs map[string]pv1.PodDisruptionBudget) {
        for key, managedPDB := range managedPDBs {
                desiredPDB, ok := desiredPDBs[key]
                if !ok {
                        err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Delete(ctx, managedPDB.Name, metav1.DeleteOptions{})
                        if err != nil {
                                log.Errorf("Failed to delete PDB: %v", err)
                                continue
                        }

                        log.WithFields(log.Fields{
                                "action":    "removed",
                                "pdb":       managedPDB.Name,
                                "namespace": managedPDB.Namespace,
                                "selector":  managedPDB.Spec.Selector.String(),
                        }).Info("")

                        // If we delete a PDB then we don't want to attempt to update it later since this will
                        // result in a `StorageError` since we can't find the PDB to make an update to it.
                        continue
                }

                // check if PDBs are equal an only update if not
                updatedPDB, needsUpdate := mergeActualAndDesiredPDB(managedPDB, desiredPDB)
                if needsUpdate {
                        err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
                                // Technically the updatedPDB and managedPDB namespace should never be different
                                // but just to be **certain** we're updating the correct namespace we'll just use
                                // the one that was given to us and not potentially modified
                                _, err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Update(ctx, &updatedPDB, metav1.UpdateOptions{})

                                // If the update failed that likely means that our definition of what was on the cluster
                                // has become out of date. To resolve this we'll need to get a more up to date copy of
                                // the object we're attempting to modify
                                if err != nil {
                                        currentPDB, err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Get(ctx, managedPDB.Name, metav1.GetOptions{})

                                        // This err is locally scoped to this if block and will not cause our `RetryOnConflict`
                                        // to pass if it is nil. If we're in this block then we will get another Retry
                                        if err != nil {
                                                return err
                                        }

                                        updatedPDB, _ = mergeActualAndDesiredPDB(
                                                *currentPDB,
                                                desiredPDB,
                                        )
                                }

                                // If this err != nil then the current block will be re-run by `RetryOnConflict`
                                // on an exponential backoff schedule to see if we can fix the problem by trying again
                                return err
                        })
                        if err != nil {
                                log.Errorf("Failed to update PDB: %v", err)
                                continue
                        }

                        log.WithFields(log.Fields{
                                "action":    "updated",
                                "pdb":       desiredPDB.Name,
                                "namespace": desiredPDB.Namespace,
                                "selector":  desiredPDB.Spec.Selector.String(),
                        }).Info("")
                }
        }

        for key, desiredPDB := range desiredPDBs {
                if _, ok := managedPDBs[key]; !ok {
                        _, err := n.PolicyV1().PodDisruptionBudgets(desiredPDB.Namespace).Create(ctx, &desiredPDB, metav1.CreateOptions{})
                        if err != nil {
                                log.Errorf("Failed to create PDB: %v", err)
                                continue
                        }

                        log.WithFields(log.Fields{
                                "action":    "added",
                                "pdb":       desiredPDB.Name,
                                "namespace": desiredPDB.Namespace,
                                "selector":  desiredPDB.Spec.Selector.String(),
                        }).Info("")
                }
        }
}

func overrideNonReadyTTL(annotations map[string]string, nonReadyTTL time.Time) (time.Time, error) {
        if ttlVal, ok := annotations[nonReadyTTLAnnotationName]; ok {
                duration, err := time.ParseDuration(ttlVal)
                if err != nil {
                        return time.Time{}, err
                }
                return time.Now().UTC().Add(-duration), nil
        }
        return nonReadyTTL, nil
}

// pdbSpecValid returns true if the PDB spec is up-to-date
func pdbSpecValid(pdb pv1.PodDisruptionBudget) bool {
        return pdb.Spec.MinAvailable == nil
}

// getMatchedPDBs gets matching PodDisruptionBudgets.
func getMatchedPDBs(labels map[string]string, pdbs map[string]pv1.PodDisruptionBudget) []pv1.PodDisruptionBudget {
        matchedPDBs := make([]pv1.PodDisruptionBudget, 0)
        for _, pdb := range pdbs {
                if labelsIntersect(labels, pdb.Spec.Selector.MatchLabels) {
                        matchedPDBs = append(matchedPDBs, pdb)
                }
        }
        return matchedPDBs
}

func getOwnedPDBs(pdbs map[string]pv1.PodDisruptionBudget, owner kubeResource) []pv1.PodDisruptionBudget {
        ownedPDBs := make([]pv1.PodDisruptionBudget, 0, len(pdbs))
        for _, pdb := range pdbs {
                if isOwnedReference(owner, pdb.ObjectMeta) {
                        ownedPDBs = append(ownedPDBs, pdb)
                }
        }
        return ownedPDBs
}

// isOwnedReference returns true if the dependent object is owned by the owner
// object.
func isOwnedReference(owner kubeResource, dependent metav1.ObjectMeta) bool {
        for _, ref := range dependent.OwnerReferences {
                if ref.APIVersion == owner.APIVersion() &&
                        ref.Kind == owner.Kind() &&
                        ref.UID == owner.UID() &&
                        ref.Name == owner.Name() {
                        return true
                }
        }
        return false
}

// containLabels reports whether expectedLabels are in labels.
func containLabels(labels, expectedLabels map[string]string) bool {
        for key, val := range expectedLabels {
                if v, ok := labels[key]; !ok || v != val {
                        return false
                }
        }
        return true
}

// labelsIntersect checks whether two maps a and b intersects. Intersection is
// defined as at least one identical key value pair must exist in both maps and
// there must be no keys which match where the values doesn't match.
func labelsIntersect(a, b map[string]string) bool {
        intersect := false
        for key, val := range a {
                v, ok := b[key]
                if ok {
                        if v == val {
                                intersect = true
                        } else { // if the key exists but the values doesn't match, don't consider it an intersection
                                return false
                        }
                }
        }

        return intersect
}

func filterPDBs(pdbs []pv1.PodDisruptionBudget) (map[string]pv1.PodDisruptionBudget, map[string]pv1.PodDisruptionBudget) {
        managed := make(map[string]pv1.PodDisruptionBudget, len(pdbs))
        unmanaged := make(map[string]pv1.PodDisruptionBudget, len(pdbs))
        for _, pdb := range pdbs {
                if containLabels(pdb.Labels, ownerLabels) {
                        managed[pdb.Namespace+"/"+pdb.Name] = pdb
                        continue
                }
                unmanaged[pdb.Namespace+"/"+pdb.Name] = pdb
        }
        return managed, unmanaged
}

func (n *PDBController) generatePDB(owner kubeResource, ttl time.Time) pv1.PodDisruptionBudget {
        var suffix string
        if n.pdbNameSuffix != "" {
                suffix = "-" + n.pdbNameSuffix
        }

        pdb := pv1.PodDisruptionBudget{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      owner.Name() + suffix,
                        Namespace: owner.Namespace(),
                        OwnerReferences: []metav1.OwnerReference{
                                {
                                        APIVersion: owner.APIVersion(),
                                        Kind:       owner.Kind(),
                                        Name:       owner.Name(),
                                        UID:        owner.UID(),
                                },
                        },
                        Labels:      owner.Labels(),
                        Annotations: make(map[string]string),
                },
                Spec: pv1.PodDisruptionBudgetSpec{
                        MaxUnavailable: &n.maxUnavailable,
                        Selector:       owner.Selector(),
                },
        }

        if n.parentResourceHash {
                // if we fail to generate the hash simply fall back to using
                // the existing selector
                hash, err := resourceHash(owner.Kind(), owner.Name())
                if err == nil {
                        pdb.Spec.Selector = &metav1.LabelSelector{
                                MatchLabels: map[string]string{
                                        parentResourceHashLabel: hash,
                                },
                        }
                }
        }

        if pdb.Labels == nil {
                pdb.Labels = make(map[string]string)
        }
        pdb.Labels[heritageLabel] = pdbController

        if !ttl.IsZero() {
                pdb.Annotations[nonReadySinceAnnotationName] = ttl.Format(time.RFC3339)
        }
        return pdb
}

func resourceHash(kind, name string) (string, error) {
        h := sha1.New()
        _, err := h.Write([]byte(kind + "-" + name))
        if err != nil {
                return "", err
        }
        return hex.EncodeToString(h.Sum(nil)), nil
}

1	package main
2
3	import (
4	"context"
5	"crypto/sha1"
6	"encoding/hex"
7	"time"
8
9	log "github.com/sirupsen/logrus"
10	v1 "k8s.io/api/core/v1"
11	pv1 "k8s.io/api/policy/v1"
12	"k8s.io/apimachinery/pkg/api/equality"
13	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14	"k8s.io/apimachinery/pkg/util/intstr"
15	"k8s.io/client-go/kubernetes"
16	"k8s.io/client-go/util/retry"
17	)
18
19	const (
20	heritageLabel = "heritage"
21	pdbController = "pdb-controller"
22	nonReadyTTLAnnotationName = "pdb-controller.zalando.org/non-ready-ttl"
23	nonReadySinceAnnotationName = "pdb-controller.zalando.org/non-ready-since"
24	parentResourceHashLabel = "parent-resource-hash"
25	)
26
27	var (
28	ownerLabels = map[string]string{heritageLabel: pdbController}
29	)
30
31	// PDBController creates PodDisruptionBudgets for deployments and StatefulSets
32	// if missing.
33	type PDBController struct {
34	kubernetes.Interface
35	interval time.Duration
36	pdbNameSuffix string
37	nonReadyTTL time.Duration
38	parentResourceHash bool
39	maxUnavailable intstr.IntOrString
40	}
41
42	// NewPDBController initializes a new PDBController.
43	func NewPDBController(interval time.Duration, client kubernetes.Interface, pdbNameSuffix string, nonReadyTTL time.Duration, parentResourceHash bool, maxUnavailable intstr.IntOrString) *PDBController {	24✔
44	return &PDBController{	24✔
45	Interface: client,	24✔
46	interval: interval,	24✔
47	pdbNameSuffix: pdbNameSuffix,	24✔
48	nonReadyTTL: nonReadyTTL,	24✔
49	parentResourceHash: parentResourceHash,	24✔
50	maxUnavailable: maxUnavailable,	24✔
51	}	24✔
52	}	24✔
53
54	// Run runs the controller loop until it receives a stop signal over the stop
55	// channel.
56	func (n *PDBController) Run(ctx context.Context) {	1✔
57	for {	2✔
58	log.Debug("Running main control loop.")	1✔
59	err := n.runOnce(ctx)	1✔
60	if err != nil {	1✔
61	log.Error(err)	×
62	}	×
63
64	select {	1✔
65	case <-time.After(n.interval):	×
66	case <-ctx.Done():	1✔
67	log.Info("Terminating main controller loop.")	1✔
68	return	1✔
69	}
70	}
71	}
72
73	// runOnce runs the main reconcilation loop of the controller.
74	func (n *PDBController) runOnce(ctx context.Context) error {	26✔
75	allPDBs, err := n.PolicyV1().PodDisruptionBudgets(v1.NamespaceAll).List(ctx, metav1.ListOptions{})	26✔
76	if err != nil {	26✔
77	return err	×
78	}	×
79
80	managedPDBs, unmanagedPDBs := filterPDBs(allPDBs.Items)	26✔
81		26✔
82	deployments, err := n.AppsV1().Deployments(v1.NamespaceAll).List(ctx, metav1.ListOptions{})	26✔
83	if err != nil {	26✔
84	return err	×
85	}	×
86
87	statefulSets, err := n.AppsV1().StatefulSets(v1.NamespaceAll).List(ctx, metav1.ListOptions{})	26✔
88	if err != nil {	26✔
89	return err	×
90	}	×
91
92	resources := make([]kubeResource, 0, len(deployments.Items)+len(statefulSets.Items))	26✔
93		26✔
94	for _, d := range deployments.Items {	50✔
95	// manually set Kind and APIVersion because of a bug in	24✔
96	// client-go	24✔
97	// https://github.com/kubernetes/client-go/issues/308	24✔
98	d.Kind = "Deployment"	24✔
99	d.APIVersion = "apps/v1"	24✔
100	resources = append(resources, deployment{d})	24✔
101	}	24✔
102
103	for _, s := range statefulSets.Items {	50✔
104	// manually set Kind and APIVersion because of a bug in	24✔
105	// client-go	24✔
106	// https://github.com/kubernetes/client-go/issues/308	24✔
107	s.Kind = "StatefulSet"	24✔
108	s.APIVersion = "apps/v1"	24✔
109	resources = append(resources, statefulSet{s})	24✔
110	}	24✔
111
112	desiredPDBs := n.generateDesiredPDBs(resources, managedPDBs, unmanagedPDBs)	26✔
113	n.reconcilePDBs(ctx, desiredPDBs, managedPDBs)	26✔
114	return nil	26✔
115	}
116
117	func (n *PDBController) generateDesiredPDBs(resources []kubeResource, managedPDBs, unmanagedPDBs map[string]pv1.PodDisruptionBudget) map[string]pv1.PodDisruptionBudget {	26✔
118	desiredPDBs := make(map[string]pv1.PodDisruptionBudget, len(managedPDBs))	26✔
119		26✔
120	nonReadyTTL := time.Time{}	26✔
121	if n.nonReadyTTL > 0 {	50✔
122	nonReadyTTL = time.Now().UTC().Add(-n.nonReadyTTL)	24✔
123	}	24✔
124
125	for _, resource := range resources {	74✔
126	matchedPDBs := getMatchedPDBs(resource.TemplateLabels(), unmanagedPDBs)	48✔
127		48✔
128	// don't create managed PDB if there is already unmanaged and	48✔
129	// matched PDBs	48✔
130	if len(matchedPDBs) > 0 {	52✔
131	continue	4✔
132	}
133
134	// don't create PDB if the resource has 1 or less replicas
135	if resource.Replicas() <= 1 {	48✔
136	continue	4✔
137	}
138
139	// ensure PDB if the resource has more than one replica and all
140	// of them are ready
141	if resource.StatusReadyReplicas() >= resource.Replicas() {	60✔
142	pdb := n.generatePDB(resource, time.Time{})	20✔
143	desiredPDBs[pdb.Namespace+"/"+pdb.Name] = pdb	20✔
144	continue	20✔
145	}
146
147	ownedPDBs := getOwnedPDBs(managedPDBs, resource)	20✔
148	validPDBs := make([]pv1.PodDisruptionBudget, 0, len(ownedPDBs))	20✔
149	// only consider valid PDBs. If they're invalid they'll be	20✔
150	// recreated on the next iteration	20✔
151	for _, pdb := range ownedPDBs {	40✔
152	if pdbSpecValid(pdb) {	40✔
153	validPDBs = append(validPDBs, pdb)	20✔
154	}	20✔
155	}
156
157	if len(validPDBs) > 0 {	40✔
158	// it's unlikely that we will have more than a single	20✔
159	// valid owned PDB. If we do simply pick the first one	20✔
160	// and check if it's still valid. Any other PDBs will	20✔
161	// automatically get dropped.	20✔
162	pdb := validPDBs[0]	20✔
163	if pdb.Annotations == nil {	20✔
164	pdb.Annotations = make(map[string]string)	×
165	}	×
166
167	ttl, err := overrideNonReadyTTL(resource.Annotations(), nonReadyTTL)	20✔
168	if err != nil {	20✔
169	log.Errorf("Failed to override PDB Delete TTL: %s", err)	×
170	}	×
171
172	var nonReadySince time.Time	20✔
173	if nonReadySinceStr, ok := pdb.Annotations[nonReadySinceAnnotationName]; ok {	36✔
174	nonReadySince, err = time.Parse(time.RFC3339, nonReadySinceStr)	16✔
175	if err != nil {	16✔
176	log.Errorf("Failed to parse non-ready-since annotation '%s': %v", nonReadySinceStr, err)	×
177	}	×
178	}
179
180	if !nonReadySince.IsZero() {	36✔
181	if !ttl.IsZero() && nonReadySince.Before(ttl) {	24✔
182	continue	8✔
183	}
184	} else {	4✔
185	nonReadySince = time.Now().UTC()	4✔
186	}	4✔
187
188	generatedPDB := n.generatePDB(resource, nonReadySince)	12✔
189	desiredPDBs[generatedPDB.Namespace+"/"+generatedPDB.Name] = generatedPDB	12✔
190	}
191	}
192
193	return desiredPDBs	26✔
194	}
195
196	// mergeActualAndDesiredPDB takes the current definition of a PDB as it is in cluster and a PDB
197	// with our desired configurations and does a merge between them. We also return a boolean to tell the
198	// caller if any change actually had to be made to achieve our desired state or not
199	func mergeActualAndDesiredPDB(managedPDB, desiredPDB pv1.PodDisruptionBudget) (pv1.PodDisruptionBudget, bool) {	28✔
200		28✔
201	needsUpdate := false	28✔
202		28✔
203	// check if PDBs are equal an only update if not	28✔
204	if !equality.Semantic.DeepEqual(managedPDB.Spec, desiredPDB.Spec) \|\|	28✔
205	!equality.Semantic.DeepEqual(managedPDB.Labels, desiredPDB.Labels) \|\|	28✔
206	!equality.Semantic.DeepEqual(managedPDB.Annotations, desiredPDB.Annotations) {	44✔
207	managedPDB.Annotations = desiredPDB.Annotations	16✔
208	managedPDB.Labels = desiredPDB.Labels	16✔
209	managedPDB.Spec = desiredPDB.Spec	16✔
210		16✔
211	needsUpdate = true	16✔
212	}	16✔
213
214	return managedPDB, needsUpdate	28✔
215	}
216
217	func (n *PDBController) reconcilePDBs(ctx context.Context, desiredPDBs, managedPDBs map[string]pv1.PodDisruptionBudget) {	26✔
218	for key, managedPDB := range managedPDBs {	70✔
219	desiredPDB, ok := desiredPDBs[key]	44✔
220	if !ok {	60✔
221	err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Delete(ctx, managedPDB.Name, metav1.DeleteOptions{})	16✔
222	if err != nil {	16✔
223	log.Errorf("Failed to delete PDB: %v", err)	×
224	continue	×
225	}
226
227	log.WithFields(log.Fields{	16✔
228	"action": "removed",	16✔
229	"pdb": managedPDB.Name,	16✔
230	"namespace": managedPDB.Namespace,	16✔
231	"selector": managedPDB.Spec.Selector.String(),	16✔
232	}).Info("")	16✔
233		16✔
234	// If we delete a PDB then we don't want to attempt to update it later since this will	16✔
235	// result in a `StorageError` since we can't find the PDB to make an update to it.	16✔
236	continue	16✔
237	}
238
239	// check if PDBs are equal an only update if not
240	updatedPDB, needsUpdate := mergeActualAndDesiredPDB(managedPDB, desiredPDB)	28✔
241	if needsUpdate {	44✔
242	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {	32✔
243	// Technically the updatedPDB and managedPDB namespace should never be different	16✔
244	// but just to be certain we're updating the correct namespace we'll just use	16✔
245	// the one that was given to us and not potentially modified	16✔
246	_, err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Update(ctx, &updatedPDB, metav1.UpdateOptions{})	16✔
247		16✔
248	// If the update failed that likely means that our definition of what was on the cluster	16✔
249	// has become out of date. To resolve this we'll need to get a more up to date copy of	16✔
250	// the object we're attempting to modify	16✔
251	if err != nil {	16✔
252	currentPDB, err := n.PolicyV1().PodDisruptionBudgets(managedPDB.Namespace).Get(ctx, managedPDB.Name, metav1.GetOptions{})	×
253		×
254	// This err is locally scoped to this if block and will not cause our `RetryOnConflict`	×
255	// to pass if it is nil. If we're in this block then we will get another Retry	×
256	if err != nil {	×
257	return err	×
258	}	×
259
260	updatedPDB, _ = mergeActualAndDesiredPDB(	×
261	*currentPDB,	×
262	desiredPDB,	×
263	)	×
264	}
265
266	// If this err != nil then the current block will be re-run by `RetryOnConflict`
267	// on an exponential backoff schedule to see if we can fix the problem by trying again
268	return err	16✔
269	})
270	if err != nil {	16✔
271	log.Errorf("Failed to update PDB: %v", err)	×
272	continue	×
273	}
274
275	log.WithFields(log.Fields{	16✔
276	"action": "updated",	16✔
277	"pdb": desiredPDB.Name,	16✔
278	"namespace": desiredPDB.Namespace,	16✔
279	"selector": desiredPDB.Spec.Selector.String(),	16✔
280	}).Info("")	16✔
281	}
282	}
283
284	for key, desiredPDB := range desiredPDBs {	58✔
285	if _, ok := managedPDBs[key]; !ok {	36✔
286	_, err := n.PolicyV1().PodDisruptionBudgets(desiredPDB.Namespace).Create(ctx, &desiredPDB, metav1.CreateOptions{})	4✔
287	if err != nil {	4✔
288	log.Errorf("Failed to create PDB: %v", err)	×
289	continue	×
290	}
291
292	log.WithFields(log.Fields{	4✔
293	"action": "added",	4✔
294	"pdb": desiredPDB.Name,	4✔
295	"namespace": desiredPDB.Namespace,	4✔
296	"selector": desiredPDB.Spec.Selector.String(),	4✔
297	}).Info("")	4✔
298	}
299	}
300	}
301
302	func overrideNonReadyTTL(annotations map[string]string, nonReadyTTL time.Time) (time.Time, error) {	20✔
303	if ttlVal, ok := annotations[nonReadyTTLAnnotationName]; ok {	28✔
304	duration, err := time.ParseDuration(ttlVal)	8✔
305	if err != nil {	8✔
306	return time.Time{}, err	×
307	}	×
308	return time.Now().UTC().Add(-duration), nil	8✔
309	}
310	return nonReadyTTL, nil	12✔
311	}
312
313	// pdbSpecValid returns true if the PDB spec is up-to-date
314	func pdbSpecValid(pdb pv1.PodDisruptionBudget) bool {	20✔
315	return pdb.Spec.MinAvailable == nil	20✔
316	}	20✔
317
318	// getMatchedPDBs gets matching PodDisruptionBudgets.
319	func getMatchedPDBs(labels map[string]string, pdbs map[string]pv1.PodDisruptionBudget) []pv1.PodDisruptionBudget {	48✔
320	matchedPDBs := make([]pv1.PodDisruptionBudget, 0)	48✔
321	for _, pdb := range pdbs {	56✔
322	if labelsIntersect(labels, pdb.Spec.Selector.MatchLabels) {	12✔
323	matchedPDBs = append(matchedPDBs, pdb)	4✔
324	}	4✔
325	}
326	return matchedPDBs	48✔
327	}
328
329	func getOwnedPDBs(pdbs map[string]pv1.PodDisruptionBudget, owner kubeResource) []pv1.PodDisruptionBudget {	20✔
330	ownedPDBs := make([]pv1.PodDisruptionBudget, 0, len(pdbs))	20✔
331	for _, pdb := range pdbs {	60✔
332	if isOwnedReference(owner, pdb.ObjectMeta) {	60✔
333	ownedPDBs = append(ownedPDBs, pdb)	20✔
334	}	20✔
335	}
336	return ownedPDBs	20✔
337	}
338
339	// isOwnedReference returns true if the dependent object is owned by the owner
340	// object.
341	func isOwnedReference(owner kubeResource, dependent metav1.ObjectMeta) bool {	40✔
342	for _, ref := range dependent.OwnerReferences {	80✔
343	if ref.APIVersion == owner.APIVersion() &&	40✔
344	ref.Kind == owner.Kind() &&	40✔
345	ref.UID == owner.UID() &&	40✔
346	ref.Name == owner.Name() {	60✔
347	return true	20✔
348	}	20✔
349	}
350	return false	20✔
351	}
352
353	// containLabels reports whether expectedLabels are in labels.
354	func containLabels(labels, expectedLabels map[string]string) bool {	50✔
355	for key, val := range expectedLabels {	100✔
356	if v, ok := labels[key]; !ok \|\| v != val {	55✔
357	return false	5✔
358	}	5✔
359	}
360	return true	45✔
361	}
362
363	// labelsIntersect checks whether two maps a and b intersects. Intersection is
364	// defined as at least one identical key value pair must exist in both maps and
365	// there must be no keys which match where the values doesn't match.
366	func labelsIntersect(a, b map[string]string) bool {	11✔
367	intersect := false	11✔
368	for key, val := range a {	22✔
369	v, ok := b[key]	11✔
370	if ok {	22✔
371	if v == val {	17✔
372	intersect = true	6✔
373	} else { // if the key exists but the values doesn't match, don't consider it an intersection	11✔
374	return false	5✔
375	}	5✔
376	}
377	}
378
379	return intersect	6✔
380	}
381
382	func filterPDBs(pdbs []pv1.PodDisruptionBudget) (map[string]pv1.PodDisruptionBudget, map[string]pv1.PodDisruptionBudget) {	26✔
383	managed := make(map[string]pv1.PodDisruptionBudget, len(pdbs))	26✔
384	unmanaged := make(map[string]pv1.PodDisruptionBudget, len(pdbs))	26✔
385	for _, pdb := range pdbs {	74✔
386	if containLabels(pdb.Labels, ownerLabels) {	92✔
387	managed[pdb.Namespace+"/"+pdb.Name] = pdb	44✔
388	continue	44✔
389	}
390	unmanaged[pdb.Namespace+"/"+pdb.Name] = pdb	4✔
391	}
392	return managed, unmanaged	26✔
393	}
394
395	func (n *PDBController) generatePDB(owner kubeResource, ttl time.Time) pv1.PodDisruptionBudget {	32✔
396	var suffix string	32✔
397	if n.pdbNameSuffix != "" {	64✔
398	suffix = "-" + n.pdbNameSuffix	32✔
399	}	32✔
400
401	pdb := pv1.PodDisruptionBudget{	32✔
402	ObjectMeta: metav1.ObjectMeta{	32✔
403	Name: owner.Name() + suffix,	32✔
404	Namespace: owner.Namespace(),	32✔
405	OwnerReferences: []metav1.OwnerReference{	32✔
406	{	32✔
407	APIVersion: owner.APIVersion(),	32✔
408	Kind: owner.Kind(),	32✔
409	Name: owner.Name(),	32✔
410	UID: owner.UID(),	32✔
411	},	32✔
412	},	32✔
413	Labels: owner.Labels(),	32✔
414	Annotations: make(map[string]string),	32✔
415	},	32✔
416	Spec: pv1.PodDisruptionBudgetSpec{	32✔
417	MaxUnavailable: &n.maxUnavailable,	32✔
418	Selector: owner.Selector(),	32✔
419	},	32✔
420	}	32✔
421		32✔
422	if n.parentResourceHash {	48✔
423	// if we fail to generate the hash simply fall back to using	16✔
424	// the existing selector	16✔
425	hash, err := resourceHash(owner.Kind(), owner.Name())	16✔
426	if err == nil {	32✔
427	pdb.Spec.Selector = &metav1.LabelSelector{	16✔
428	MatchLabels: map[string]string{	16✔
429	parentResourceHashLabel: hash,	16✔
430	},	16✔
431	}	16✔
432	}	16✔
433	}
434
435	if pdb.Labels == nil {	64✔
436	pdb.Labels = make(map[string]string)	32✔
437	}	32✔
438	pdb.Labels[heritageLabel] = pdbController	32✔
439		32✔
440	if !ttl.IsZero() {	44✔
441	pdb.Annotations[nonReadySinceAnnotationName] = ttl.Format(time.RFC3339)	12✔
442	}	12✔
443	return pdb	32✔
444	}
445
446	func resourceHash(kind, name string) (string, error) {	18✔
447	h := sha1.New()	18✔
448	_, err := h.Write([]byte(kind + "-" + name))	18✔
449	if err != nil {	18✔
450	return "", err	×
451	}	×
452	return hex.EncodeToString(h.Sum(nil)), nil	18✔
453	}

mikkeloscar / pdb-controller / 17156871947

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous