Modify pod annotations update to safely allow concurrent writes

Author: Part-TimeWizard

pkg/daemon/daemon.go

@@ -209,8 +209,7 @@ func (d *daemon) getIbSriovNetwork(networkID string) (string, *utils.IbSriovCniS
 	// Check if this network's resource is managed by this daemon
 	resourceName := netAttInfo.Annotations["k8s.v1.cni.cncf.io/resourceName"]
 	if resourceName == "" || !d.config.IsManagedResource(resourceName) {
-		// TODO(Nik) dev qol, check if someone else manages this resource or if it is orphan
-		// checkResourceOwner(networkNamespace, networkName)
+		// TODO(Nik) qol, check if someone else manages this resource or if it is orphan
 		return "", nil, fmt.Errorf("network %s uses resource %s which is not managed by this daemon", networkName, resourceName)
 	}
 
@@ -254,8 +253,8 @@ func getPodNetworkInfo(netName string, pod *kapi.Pod, netMap networksMap) (*podN
 
 // addPodFinalizer adds the GUID cleanup finalizer to a pod
 func (d *daemon) addPodFinalizer(pod *kapi.Pod, networkName string) error {
+	podFinalizer := fmt.Sprintf("%s-%s", PodGUIDFinalizer, networkName)
 	return wait.ExponentialBackoff(backoffValues, func() (bool, error) {
-		podFinalizer := fmt.Sprintf("%s-%s", PodGUIDFinalizer, networkName)
 		if err := d.kubeClient.AddFinalizerToPod(pod, podFinalizer); err != nil {
 			log.Warn().Msgf("failed to add finalizer to pod %s/%s: %v",
 				pod.Namespace, pod.Name, err)
@@ -267,8 +266,8 @@ func (d *daemon) addPodFinalizer(pod *kapi.Pod, networkName string) error {
 
 // removePodFinalizer removes the GUID cleanup finalizer from a pod
 func (d *daemon) removePodFinalizer(pod *kapi.Pod, networkName string) error {
+	podFinalizer := fmt.Sprintf("%s-%s", PodGUIDFinalizer, networkName)
 	return wait.ExponentialBackoff(backoffValues, func() (bool, error) {
-		podFinalizer := fmt.Sprintf("%s-%s", PodGUIDFinalizer, networkName)
 		if err := d.kubeClient.RemoveFinalizerFromPod(pod, podFinalizer); err != nil {
 			log.Warn().Msgf("failed to remove finalizer from pod %s/%s: %v",
 				pod.Namespace, pod.Name, err)
@@ -480,44 +479,104 @@ func syncGUIDPool(smClient plugins.SubnetManagerClient, guidPool guid.Pool) erro
 
 // Update and set Pod's network annotation.
 // If failed to update annotation, pod's GUID added into the list to be removed from Pkey.
-func (d *daemon) updatePodNetworkAnnotation(pi *podNetworkInfo, removedList *[]net.HardwareAddr) error {
+func (d *daemon) updatePodNetworkAnnotation(pi *podNetworkInfo, removedList *[]net.HardwareAddr) {
 	if pi.ibNetwork.CNIArgs == nil {
 		pi.ibNetwork.CNIArgs = &map[string]interface{}{}
 	}
 
 	(*pi.ibNetwork.CNIArgs)[utils.InfiniBandAnnotation] = utils.ConfiguredInfiniBandPod
-	netAnnotations, err := json.Marshal(pi.networks)
-	if err != nil {
-		return fmt.Errorf("failed to dump networks %+v of pod into json with error: %v", pi.networks, err)
-	}
-
-	pi.pod.Annotations[v1.NetworkAttachmentAnnot] = string(netAnnotations)
 
 	// Try to set pod's annotations in backoff loop
-	if err = wait.ExponentialBackoff(backoffValues, func() (bool, error) {
-		log.Info().Msgf("updatePodNetworkAnnotation(): Updating pod annotation for pod: %s with anootation: %s", pi.pod.Name, pi.pod.Annotations)
+	if err := wait.ExponentialBackoff(backoffValues, func() (bool, error) {
+
+		// Get latest annotations state to avoid conflicts
+		latestPodAnnotations, networks, err := d.getLatestPodAnnotations(pi.pod)
+		if err != nil {
+			log.Warn().Msgf("failed to get latest pod annotations for %s/%s: %v", pi.pod.Namespace, pi.pod.Name, err)
+			return false, nil
+		}
+
+		targetNetwork, err := utils.GetPodNetwork(networks, pi.ibNetwork.Name)
+		if err != nil {
+			return false, fmt.Errorf("failed to locate network %s in pod %s/%s annotations: %v", pi.ibNetwork.Name, pi.pod.Namespace, pi.pod.Name, err)
+		}
+
+		err = updateInfiniBandNetwork(targetNetwork, pi.ibNetwork)
+		if err != nil {
+			return false, fmt.Errorf("failed to update infiniband network for pod %s/%s: %v", pi.pod.Namespace, pi.pod.Name, err)
+		}
+
+		netAnnotations, err := json.Marshal(networks)
+		if err != nil {
+			return false, fmt.Errorf("failed to marshal updated networks for pod %s/%s: %v", pi.pod.Namespace, pi.pod.Name, err)
+		}
+
+		if latestPodAnnotations == nil {
+			return false, fmt.Errorf("latestPodAnnotations is nil for pod %s/%s", pi.pod.Namespace, pi.pod.Name)
+		}
+
+		latestPodAnnotations[v1.NetworkAttachmentAnnot] = string(netAnnotations)
+		pi.pod.Annotations = latestPodAnnotations
+
+		log.Info().Msgf("updatePodNetworkAnnotation(): Updating pod annotation for pod: %s/%s", pi.pod.Namespace, pi.pod.Name)
 		if err = d.kubeClient.SetAnnotationsOnPod(pi.pod, pi.pod.Annotations); err != nil {
 			if kerrors.IsNotFound(err) {
 				return false, err
 			}
-			log.Warn().Msgf("failed to update pod annotations with err: %v", err)
+			if kerrors.IsConflict(err) {
+				log.Warn().Msgf("conflict while updating pod annotations for %s/%s, will retry", pi.pod.Namespace, pi.pod.Name)
+				return false, nil
+			}
+			log.Warn().Msgf("failed to update pod annotations for %s/%s with err: %v", pi.pod.Namespace, pi.pod.Name, err)
 			return false, nil
 		}
-		log.Info().Msgf("updatePodNetworkAnnotation(): Success on updating pod annotation for pod: %s with anootation: %s", pi.pod.Name, pi.pod.Annotations)
+
+		log.Info().Msgf("updatePodNetworkAnnotation(): Success on updating pod annotation for pod: %s/%s with annotations: %s", pi.pod.Namespace, pi.pod.Name, pi.pod.Annotations)
 		return true, nil
 	}); err != nil {
-		log.Error().Msgf("failed to update pod annotations")
+		log.Error().Msgf("failed to update pod annotations for %s/%s with error: %v", pi.pod.Namespace, pi.pod.Name, err)
 
 		if err = d.guidPool.ReleaseGUID(pi.addr.String()); err != nil {
-			log.Warn().Msgf("failed to release guid \"%s\" from removed pod \"%s\" in namespace "+
-				"\"%s\" with error: %v", pi.addr.String(), pi.pod.Name, pi.pod.Namespace, err)
+			log.Warn().Msgf("failed to release guid \"%s\" from removed pod \"%s\" in namespace \"%s\" with error: %v", pi.addr.String(), pi.pod.Name, pi.pod.Namespace, err)
 		} else {
 			delete(d.guidPodNetworkMap, pi.addr.String())
 		}
 
 		*removedList = append(*removedList, pi.addr)
 	}
 
+	return
+}
+
+// Retrieves the latest annotations for a pod and returns the annotations and the pod networks.
+func (d *daemon) getLatestPodAnnotations(pod *kapi.Pod) (map[string]string, []*v1.NetworkSelectionElement, error) {
+	latestPod, err := d.kubeClient.GetPod(pod.Namespace, pod.Name)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	networks, err := netAttUtils.ParsePodNetworkAnnotation(latestPod)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return latestPod.Annotations, networks, nil
+}
+
+// Replaces target network with source network, erroring if source is already configured.
+func updateInfiniBandNetwork(target *v1.NetworkSelectionElement, source *v1.NetworkSelectionElement) error {
+	if target == nil || source == nil {
+		return fmt.Errorf("target or source network is nil")
+	}
+
+	if target.CNIArgs != nil {
+		if (*target.CNIArgs)[utils.InfiniBandAnnotation] == utils.ConfiguredInfiniBandPod {
+			return fmt.Errorf("target network is already configured")
+		}
+	}
+
+	target.InfinibandGUIDRequest = source.InfinibandGUIDRequest
+	target.CNIArgs = source.CNIArgs
 	return nil
 }
 
@@ -609,10 +668,7 @@ func (d *daemon) AddPeriodicUpdate() {
 		var removedGUIDList []net.HardwareAddr
 		for _, pi := range passedPods {
 			log.Info().Msgf("Updating annotations for the pod %s, network %s", pi.pod.Name, pi.ibNetwork.Name)
-			err = d.updatePodNetworkAnnotation(pi, &removedGUIDList)
-			if err != nil {
-				log.Error().Msgf("%v", err)
-			}
+			d.updatePodNetworkAnnotation(pi, &removedGUIDList)
 		}
 
 		if ibCniSpec.PKey != "" && len(removedGUIDList) != 0 {

pkg/daemon/daemon_e2e_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -457,6 +456,15 @@ func (c *testK8sClient) GetPods(namespace string) (*corev1.PodList, error) {
 	return podList, err
 }
 
+func (c *testK8sClient) GetPod(namespace, name string) (*corev1.Pod, error) {
+	pod := &corev1.Pod{}
+	key := client.ObjectKey{Namespace: namespace, Name: name}
+	if err := c.client.Get(ctx, key, pod); err != nil {
+		return nil, err
+	}
+	return pod, nil
+}
+
 func (c *testK8sClient) SetAnnotationsOnPod(pod *corev1.Pod, annotations map[string]string) error {
 	pod.Annotations = annotations
 	return c.client.Update(ctx, pod)
@@ -533,9 +541,3 @@ func (c *testK8sClient) GetRestClient() rest.Interface {
 	}
 	return clientset.CoreV1().RESTClient()
 }
-
-func (c *testK8sClient) PatchPod(pod *corev1.Pod, patchType types.PatchType, patchData []byte) error {
-	// Use controller-runtime client patch
-	patch := client.RawPatch(patchType, patchData)
-	return c.client.Patch(ctx, pod, patch)
-}

pkg/daemon/daemon_test.go

@@ -209,7 +209,8 @@ var _ = Describe("Daemon Finalizer Tests", func() {
 				kubeClient.On("AddFinalizerToNetworkAttachmentDefinition", "default", "ib-network2", GUIDInUFMFinalizer).Return(nil)
 
 				// Mock pod annotation updates
-				kubeClient.On("SetAnnotationsOnPod", testPod, mock.AnythingOfType("map[string]string")).Return(nil)
+				kubeClient.On("GetPod", testPod.Namespace, testPod.Name).Return(testPod, nil)
+				kubeClient.On("SetAnnotationsOnPod", mock.AnythingOfType("*v1.Pod"), mock.AnythingOfType("map[string]string")).Return(nil)
 
 				// Mock SM client calls
 				smClient.On("AddGuidsToPKey", 1, mock.AnythingOfType("[]net.HardwareAddr")).Return(nil)

pkg/k8s-client/client.go

@@ -2,16 +2,15 @@ package k8sclient
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
+	"reflect"
 	"time"
 
 	netapi "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
 	netclient "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/client/clientset/versioned/typed/k8s.cni.cncf.io/v1"
 	"github.com/rs/zerolog/log"
 	kapi "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -20,8 +19,8 @@ import (
 
 type Client interface {
 	GetPods(namespace string) (*kapi.PodList, error)
+	GetPod(namespace, name string) (*kapi.Pod, error)
 	SetAnnotationsOnPod(pod *kapi.Pod, annotations map[string]string) error
-	PatchPod(pod *kapi.Pod, patchType types.PatchType, patchData []byte) error
 	GetNetworkAttachmentDefinition(namespace, name string) (*netapi.NetworkAttachmentDefinition, error)
 	GetRestClient() rest.Interface
 	AddFinalizerToNetworkAttachmentDefinition(namespace, name, finalizer string) error
@@ -65,33 +64,38 @@ func (c *client) GetPods(namespace string) (*kapi.PodList, error) {
 	return c.clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
 }
 
+// GetPod obtains a single Pod resource from the kubernetes api server
+func (c *client) GetPod(namespace, name string) (*kapi.Pod, error) {
+	log.Debug().Msgf("getting pod namespace: %s, name: %s", namespace, name)
+	return c.clientset.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{})
+}
+
 // SetAnnotationsOnPod takes the pod object and map of key/value string pairs to set as annotations
 func (c *client) SetAnnotationsOnPod(pod *kapi.Pod, annotations map[string]string) error {
 	log.Info().Msgf("Setting annotation on pod, namespace: %s, podName: %s, annotations: %v",
 		pod.Namespace, pod.Name, annotations)
-	var err error
-	var patchData []byte
-	patch := struct {
-		Metadata map[string]interface{} `json:"metadata"`
-	}{
-		Metadata: map[string]interface{}{
-			"annotations": annotations,
-		},
-	}
 
-	podDesc := pod.Namespace + "/" + pod.Name
-	patchData, err = json.Marshal(&patch)
+	// Get the latest version of the pod
+	currentPod, err := c.clientset.CoreV1().Pods(pod.Namespace).Get(context.TODO(), pod.Name, metav1.GetOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to set annotations on pod %s: %v", podDesc, err)
+		return err
 	}
-	return c.PatchPod(pod, types.MergePatchType, patchData)
-}
 
-// PatchPod applies the patch changes
-func (c *client) PatchPod(pod *kapi.Pod, patchType types.PatchType, patchData []byte) error {
-	log.Debug().Msgf("patch pod, namespace: %s, podName: %s", pod.Namespace, pod.Name)
-	_, err := c.clientset.CoreV1().Pods(pod.Namespace).Patch(
-		context.TODO(), pod.Name, patchType, patchData, metav1.PatchOptions{})
+	// Check if there are any conflicts with the current view of the pod's annotations and the latest version.
+	if currentPod.Annotations != nil {
+		if !reflect.DeepEqual(currentPod.Annotations, pod.Annotations) {
+			return fmt.Errorf("conflict with the current view of the pod's annotations and the latest version")
+		}
+	}
+
+	currentPod.Annotations = annotations
+
+	// Update the pod with retry and backoff
+	err = wait.ExponentialBackoff(backoffValues, func() (bool, error) {
+		_, err = c.clientset.CoreV1().Pods(pod.Namespace).Update(
+			context.Background(), currentPod, metav1.UpdateOptions{})
+		return err == nil, nil
+	})
 	return err
 }