canary pod label (#52)

api/v1alpha1/const.go

@@ -10,6 +10,10 @@ const (
 	ExtendedDaemonSetNameLabelKey = "extendeddaemonset.datadoghq.com/name"
 	// ExtendedDaemonSetReplicaSetNameLabelKey label key use to link a Pod to a ExtendedDaemonSetReplicaSet
 	ExtendedDaemonSetReplicaSetNameLabelKey = "extendeddaemonsetreplicaset.datadoghq.com/name"
+	// ExtendedDaemonSetReplicaSetCanaryLabelKey label key used to identify canary Pods
+	ExtendedDaemonSetReplicaSetCanaryLabelKey = "extendeddaemonsetreplicaset.datadoghq.com/canary"
+	// ExtendedDaemonSetReplicaSetCanaryLabelValue label value used to identify canary Pods
+	ExtendedDaemonSetReplicaSetCanaryLabelValue = "true"
 	// MD5ExtendedDaemonSetAnnotationKey annotation key use on Pods in order to identify which PodTemplateSpec have been used to generate it.
 	MD5ExtendedDaemonSetAnnotationKey = "extendeddaemonset.datadoghq.com/templatehash"
 	// ExtendedDaemonSetCanaryValidAnnotationKey annotation key used on Pods in order to detect if a canary deployment is considered valid.

controllers/extendeddaemonset_test.go

@@ -123,6 +123,66 @@ var _ = Describe("ExtendedDaemonSet Controller", func() {
 				return eds.Status.Canary != nil && eds.Status.Canary.ReplicaSet != ""
 			}, timeout, interval).Should(BeTrue())
 		})
+
+		It("Should add canary labels", func() {
+			Eventually(func() bool {
+				canaryPods := &corev1.PodList{}
+				listOptions := []client.ListOption{
+					client.MatchingLabels{
+						datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelKey: datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelValue,
+					},
+				}
+
+				err = k8sClient.List(context.Background(), canaryPods, listOptions...)
+				if err != nil {
+					fmt.Fprint(GinkgoWriter, err)
+					return false
+				}
+
+				return len(canaryPods.Items) == 2
+			}, timeout, interval).Should(BeTrue())
+		})
+
+		It("Should remove canary labels", func() {
+			Eventually(func() bool {
+				eds := &datadoghqv1alpha1.ExtendedDaemonSet{}
+				err = k8sClient.Get(context.Background(), key, eds)
+				if err != nil {
+					fmt.Fprint(GinkgoWriter, err)
+					return false
+				}
+
+				if eds.Status.Canary == nil {
+					return false
+				}
+
+				eds.Status.ActiveReplicaSet = eds.Status.Canary.ReplicaSet
+
+				if err = k8sClient.Update(context.Background(), eds); err != nil {
+					fmt.Fprint(GinkgoWriter, err)
+					return false
+				}
+
+				return true
+			}, timeout, interval).Should(BeTrue())
+
+			Eventually(func() bool {
+				canaryPods := &corev1.PodList{}
+				listOptions := []client.ListOption{
+					client.MatchingLabels{
+						datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelKey: datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelValue,
+					},
+				}
+
+				err = k8sClient.List(context.Background(), canaryPods, listOptions...)
+				if err != nil {
+					fmt.Fprint(GinkgoWriter, err)
+					return false
+				}
+
+				return len(canaryPods.Items) == 0
+			}, timeout, interval).Should(BeTrue())
+		})
 	})
 
 	Context("Using ExtendedDaemonsetSetting", func() {

controllers/extendeddaemonsetreplicaset/strategy/canary.go

@@ -6,6 +6,7 @@
 package strategy
 
 import (
+	"fmt"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -36,7 +37,12 @@ func ManageCanaryDeployment(client client.Client, daemonset *v1alpha1.ExtendedDa
 		if pod, ok := params.PodByNodeName[node]; ok {
 			if pod == nil {
 				result.PodsToCreate = append(result.PodsToCreate, node)
+				needRequeue = true
 			} else {
+				if err = addPodLabel(client, pod, v1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelKey, v1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelValue); err != nil {
+					params.Logger.Error(err, fmt.Sprintf("Couldn't add the canary label for pod '%s/%s', will retry later", pod.GetNamespace(), pod.GetName()))
+					needRequeue = true
+				}
 				if pod.DeletionTimestamp != nil {
 					needRequeue = true
 					continue

controllers/extendeddaemonsetreplicaset/strategy/rollingupdate.go

@@ -6,13 +6,15 @@
 package strategy
 
 import (
+	"context"
+	"fmt"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	intstrutil "k8s.io/apimachinery/pkg/util/intstr"
-	"sigs.k8s.io/controller-runtime/pkg/client"
+	runtimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	datadoghqv1alpha1 "github.com/DataDog/extendeddaemonset/api/v1alpha1"
 	"github.com/DataDog/extendeddaemonset/controllers/extendeddaemonsetreplicaset/conditions"
@@ -21,8 +23,12 @@ import (
 	podutils "github.com/DataDog/extendeddaemonset/pkg/controller/utils/pod"
 )
 
+// cleanCanaryLabelsThreshold is the duration since the last transition to a rolling update of a replicaset
+// during which we keep retrying cleaning up the canary labels that were added to the canary pods during the canary phase
+const cleanCanaryLabelsThreshold = 5 * time.Minute
+
 // ManageDeployment used to manage ReplicaSet in rollingupdate state
-func ManageDeployment(client client.Client, params *Parameters) (*Result, error) {
+func ManageDeployment(client runtimeclient.Client, params *Parameters) (*Result, error) {
 	result := &Result{}
 
 	// remove canary node if define
@@ -125,6 +131,29 @@ func ManageDeployment(client client.Client, params *Parameters) (*Result, error)
 		result.Result.Requeue = true
 	}
 
+	// Remove canary labels from canary pods (if they exist)
+	// We keep retrying these operations only for the first X minutes after starting the rolling update to avoid Listing pods endlessly.
+	if time.Since(rollingUpdateStartTime) < cleanCanaryLabelsThreshold {
+		canaryPods := &corev1.PodList{}
+		listOptions := []runtimeclient.ListOption{
+			runtimeclient.MatchingLabels{
+				datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelKey: datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelValue,
+				datadoghqv1alpha1.ExtendedDaemonSetReplicaSetNameLabelKey:   params.Replicaset.GetName(),
+			},
+		}
+		if err = client.List(context.TODO(), canaryPods, listOptions...); err != nil {
+			params.Logger.Error(err, "Couldn't get canary pods")
+			result.Result.Requeue = true
+		} else {
+			for _, pod := range canaryPods.Items {
+				if err = deletePodLabel(client, &pod, datadoghqv1alpha1.ExtendedDaemonSetReplicaSetCanaryLabelKey); err != nil {
+					params.Logger.Error(err, fmt.Sprintf("Couldn't remove canary label from pod '%s/%s'", pod.GetNamespace(), pod.GetName()))
+					result.Result.Requeue = true
+				}
+			}
+		}
+	}
+
 	return result, err
 }
 

controllers/extendeddaemonsetreplicaset/strategy/utils.go

@@ -158,3 +158,29 @@ func pauseCanaryDeployment(client client.Client, eds *datadoghqv1alpha1.Extended
 	}
 	return nil
 }
+
+// addPodLabel adds a given label to a pod, no-op if the pod is nil or if the label exists
+func addPodLabel(c client.Client, pod *corev1.Pod, k, v string) error {
+	if pod == nil {
+		return nil
+	}
+	if label, found := pod.GetLabels()[k]; found && label == v {
+		// The label is there, nothing to do
+		return nil
+	}
+	pod.Labels[k] = v
+	return c.Update(context.TODO(), pod)
+}
+
+// deletePodLabel deletes a given pod label, no-op if the pod is nil or if the label doesn't exists
+func deletePodLabel(c client.Client, pod *corev1.Pod, k string) error {
+	if pod == nil {
+		return nil
+	}
+	if _, found := pod.GetLabels()[k]; !found {
+		// The label is not there, nothing to do
+		return nil
+	}
+	delete(pod.Labels, k)
+	return c.Update(context.TODO(), pod)
+}