[v3.5] Let Calico Nodes inherit labels from Kubernetes Nodes

lib/backend/k8s/resources/node.go

@@ -16,6 +16,7 @@ package resources
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 
@@ -38,6 +39,7 @@ const (
 	nodeBgpIpv6AddrAnnotation = "projectcalico.org/IPv6Address"
 	nodeBgpAsnAnnotation      = "projectcalico.org/ASNumber"
 	nodeBgpCIDAnnotation      = "projectcalico.org/RouteReflectorClusterID"
+	nodeK8sLabelAnnotation    = "projectcalico.org/kube-labels"
 )
 
 func NewNodeClient(c *kubernetes.Clientset) K8sResourceClient {
@@ -196,6 +198,13 @@ func K8sNodeToCalico(k8sNode *kapiv1.Node) (*model.KVPair, error) {
 	calicoNode.ObjectMeta.Name = k8sNode.Name
 	SetCalicoMetadataFromK8sAnnotations(calicoNode, k8sNode)
 
+	// Calico Nodes inherit labels from Kubernetes nodes, do that merge.
+	err := mergeCalicoAndK8sLabels(calicoNode, k8sNode)
+	if err != nil {
+		log.WithError(err).Error("Failed to merge Calico and Kubernetes labels.")
+		return nil, err
+	}
+
 	// Extract the BGP configuration stored in the annotations.
 	bgpSpec := &apiv3.NodeBGPSpec{}
 	annotations := k8sNode.ObjectMeta.Annotations
@@ -239,11 +248,19 @@ func K8sNodeToCalico(k8sNode *kapiv1.Node) (*model.KVPair, error) {
 	}, nil
 }
 
-// mergeCalicoNodeIntoK8sNode takes a k8s node and a Calico node and push the values from the Calico
+// mergeCalicoNodeIntoK8sNode takes a k8s node and a Calico node and puts the values from the Calico
 // node into the k8s node.
 func mergeCalicoNodeIntoK8sNode(calicoNode *apiv3.Node, k8sNode *kapiv1.Node) (*kapiv1.Node, error) {
-	// Set the k8s annotations from the Calico node metadata.  This ensures the k8s annotations
-	// is initialized.
+	// Nodes inherit labels from Kubernetes, but we also have our own set of labels that are stored in an annotation.
+	// For nodes that are being updated, we want to avoid writing k8s labels that we inherited into our annotation
+	// and we don't want to touch the k8s labels directly.  Take a copy of the node resource and update its labels
+	// to match what we want to store in our annotation only.
+	calicoNode, err := restoreCalicoLabels(calicoNode)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set the k8s annotations from the Calico node metadata.
 	SetK8sAnnotationsFromCalicoMetadata(k8sNode, calicoNode)
 
 	if calicoNode.Spec.BGP == nil {
@@ -282,6 +299,86 @@ func mergeCalicoNodeIntoK8sNode(calicoNode *apiv3.Node, k8sNode *kapiv1.Node) (*
 	return k8sNode, nil
 }
 
+// mergeCalicoAndK8sLabels merges the Kubernetes labels (from k8sNode.Labels) with those that are already present in
+// calicoNode (which were loaded from our annotation).  Kubernetes labels take precedence.  To make the operation
+// reversible (so that we can support write back of a Calico node that was read from Kubernetes), we also store the
+// complete set of Kubernetes labels in an annotation.
+//
+// Note: if a Kubernetes label shadows a Calico label, the Calico label will be lost when the resource is written
+// back to the datastore.  This is consistent with kube-controllers' behavior.
+func mergeCalicoAndK8sLabels(calicoNode *apiv3.Node, k8sNode *kapiv1.Node) error {
+	// Now, copy the Kubernetes Node labels over.  Note: this may overwrite Calico labels of the same name, but that's
+	// consistent with the kube-controllers behavior.
+	for k, v := range k8sNode.Labels {
+		if calicoNode.Labels == nil {
+			calicoNode.Labels = map[string]string{}
+		}
+		calicoNode.Labels[k] = v
+	}
+
+	// For consistency with kube-controllers, and so we can correctly round-trip labels, we stash the kubernetes labels
+	// in an annotation.
+	if calicoNode.Annotations == nil {
+		calicoNode.Annotations = map[string]string{}
+	}
+	bytes, err := json.Marshal(k8sNode.Labels)
+	if err != nil {
+		log.WithError(err).Errorf("Error marshalling node labels")
+		return err
+	}
+	calicoNode.Annotations[nodeK8sLabelAnnotation] = string(bytes)
+	return nil
+}
+
+// restoreCalicoLabels tries to undo the transformation done by mergeCalicoLabels.  If no changes are needed, it
+// returns the input value; otherwise, it returns a copy.
+func restoreCalicoLabels(calicoNode *apiv3.Node) (*apiv3.Node, error) {
+	rawLabels := calicoNode.Annotations[nodeK8sLabelAnnotation]
+	if rawLabels == "" {
+		return calicoNode, nil
+	}
+
+	// We're about to update the labels and annotations on the node, take a copy.
+	calicoNode = calicoNode.DeepCopy()
+
+	// We stashed the k8s labels in an annotation, extract them so we can compare with the combined labels.
+	k8sLabels := map[string]string{}
+	if err := json.Unmarshal([]byte(rawLabels), &k8sLabels); err != nil {
+		log.WithError(err).Error("Failed to unmarshal k8s node labels from " +
+			nodeK8sLabelAnnotation + " annotation")
+		return nil, err
+	}
+
+	// Now remove any labels that match the k8s ones.
+	if log.GetLevel() >= log.DebugLevel {
+		log.WithField("k8s", k8sLabels).Debug("Loaded label annotations")
+	}
+	for k, k8sVal := range k8sLabels {
+		if calVal, ok := calicoNode.Labels[k]; ok && calVal != k8sVal {
+			log.WithFields(log.Fields{
+				"label":    k,
+				"newValue": calVal,
+				"k8sValue": k8sVal,
+			}).Warn("Update to label that is shadowed by a Kubernetes label will be ignored.")
+		}
+
+		// The k8s value was inherited and there was no old Calico value, drop the label so that we don't copy
+		// it to the Calico annotation.
+		if log.GetLevel() >= log.DebugLevel {
+			log.WithField("key", k).Debug("Removing inherited k8s label")
+		}
+		delete(calicoNode.Labels, k)
+	}
+
+	// Filter out our bookkeeping annotation, which is only used for round-tripping labels correctly.
+	delete(calicoNode.Annotations, nodeK8sLabelAnnotation)
+	if len(calicoNode.Annotations) == 0 {
+		calicoNode.Annotations = nil
+	}
+
+	return calicoNode, nil
+}
+
 // Calculate the IPIP Tunnel IP address to use for a given Node.  We use the first IP in the
 // node CIDR for our tunnel address.  If an IPv4 address cannot be picked from the given
 // CIDR then an empty string will be returned.

lib/backend/k8s/resources/node_test.go

@@ -311,6 +311,87 @@ var _ = Describe("Test Node conversion", func() {
 		calicoNode.Spec.BGP.IPv4IPIPTunnelAddr = "172.100.0.1"
 		newCalicoNode, err := K8sNodeToCalico(newK8sNode)
 		Expect(err).NotTo(HaveOccurred())
-		Expect(newCalicoNode.Value).To(Equal(calicoNode))
+
+		calicoNodeWithMergedLabels := calicoNode.DeepCopy()
+		calicoNodeWithMergedLabels.Annotations[nodeK8sLabelAnnotation] = "{\"net.beta.kubernetes.io/role\":\"master\"}"
+		calicoNodeWithMergedLabels.Labels["net.beta.kubernetes.io/role"] = "master"
+		Expect(newCalicoNode.Value).To(Equal(calicoNodeWithMergedLabels))
+	})
+
+	It("Should shadow labels correctly", func() {
+		kl := map[string]string{
+			"net.beta.kubernetes.io/role": "master",
+			"shadowed":                    "k8s-value",
+		}
+		cl := map[string]string{
+			"label1":   "foo",
+			"label2":   "bar",
+			"shadowed": "calico-value",
+		}
+		k8sNode := &k8sapi.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            "TestNode",
+				Labels:          kl,
+				ResourceVersion: "1234",
+				Annotations:     make(map[string]string),
+			},
+			Spec: k8sapi.NodeSpec{},
+		}
+
+		By("Merging calico node config into the k8s node")
+		calicoNode := apiv3.NewNode()
+		calicoNode.Name = "TestNode"
+		calicoNode.ResourceVersion = "1234"
+		calicoNode.Labels = cl
+
+		newK8sNode, err := mergeCalicoNodeIntoK8sNode(calicoNode, k8sNode)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(newK8sNode.Annotations).To(Equal(map[string]string{
+			"projectcalico.org/labels": `{"label1":"foo","label2":"bar","shadowed":"calico-value"}`,
+		}))
+		Expect(newK8sNode.Labels).To(Equal(kl))
+
+		By("Converting the k8s node back into a calico node")
+		newCalicoNode, err := K8sNodeToCalico(newK8sNode)
+		Expect(err).NotTo(HaveOccurred())
+
+		// When we merge k8s into Calico, the k8s labels get stashed in an annotation along with the shadowed labels:
+		calicoNodeWithMergedLabels := calicoNode.DeepCopy()
+		calicoNodeWithMergedLabels.Annotations = map[string]string{}
+		calicoNodeWithMergedLabels.Annotations[nodeK8sLabelAnnotation] = "{\"net.beta.kubernetes.io/role\":\"master\",\"shadowed\":\"k8s-value\"}"
+		// And, the k8s labels get merged in...
+		calicoNodeWithMergedLabels.Labels["net.beta.kubernetes.io/role"] = "master"
+		calicoNodeWithMergedLabels.Labels["shadowed"] = "k8s-value"
+		Expect(newCalicoNode.Value).To(Equal(calicoNodeWithMergedLabels))
+
+		// restoreCalicoLabels should undo the merge, but the shadowed label will be lost.
+		calicoNodeNoShadow := calicoNode.DeepCopy()
+		delete(calicoNodeNoShadow.Labels, "shadowed")
+		calicoNodeRestored, err := restoreCalicoLabels(calicoNodeWithMergedLabels)
+		Expect(calicoNodeRestored).To(Equal(calicoNodeNoShadow))
+
+		// For coverage, make a change to the shadowed label, this will log a warning.
+		calicoNodeWithMergedLabels.Labels["shadowed"] = "some change"
+		calicoNodeRestored, err = restoreCalicoLabels(calicoNodeWithMergedLabels)
+		Expect(calicoNodeRestored).To(Equal(calicoNodeNoShadow))
+	})
+
+	It("restoreCalicoLabels should error if annotations are malformed", func() {
+		calicoNode := apiv3.NewNode()
+		calicoNode.Annotations = map[string]string{}
+		calicoNode.Annotations[nodeK8sLabelAnnotation] = "Garbage"
+		_, err := restoreCalicoLabels(calicoNode)
+		Expect(err).To(HaveOccurred())
+
+		k8sNode := &k8sapi.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            "TestNode",
+				ResourceVersion: "1234",
+				Annotations:     make(map[string]string),
+			},
+			Spec: k8sapi.NodeSpec{},
+		}
+		_, err = mergeCalicoNodeIntoK8sNode(calicoNode, k8sNode)
+		Expect(err).To(HaveOccurred())
 	})
 })