Merge pull request #488 from ykulazhenkov/ignore-updates

Revert "server-side apply" patch and apply new fix for never-ending sync loop issue

Author: adrianchiris

controllers/nicclusterpolicy_controller.go

@@ -280,7 +280,7 @@ func (r *NicClusterPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		ctl = ctl.Watches(ws[i], &handler.EnqueueRequestForOwner{
 			IsController: true,
 			OwnerType:    &mellanoxv1alpha1.NicClusterPolicy{},
-		})
+		}, builder.WithPredicates(IgnoreSameContentPredicate{}))
 	}
 
 	return ctl.Complete(r)

controllers/predicate.go

@@ -17,6 +17,11 @@ limitations under the License.
 package controllers
 
 import (
+	"reflect"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
@@ -36,3 +41,35 @@ func (p MlnxLabelChangedPredicate) hasMlnxLabel(labels map[string]string) bool {
 func (p MlnxLabelChangedPredicate) Update(e event.UpdateEvent) bool {
 	return p.hasMlnxLabel(e.ObjectOld.GetLabels()) != p.hasMlnxLabel(e.ObjectNew.GetLabels())
 }
+
+// IgnoreSameContentPredicate filters updates if old and new object are the same,
+// ignores ResourceVersion and ManagedFields while comparing
+type IgnoreSameContentPredicate struct {
+	predicate.Funcs
+}
+
+func (p IgnoreSameContentPredicate) Update(e event.UpdateEvent) bool {
+	if e.ObjectOld == nil {
+		return false
+	}
+	if e.ObjectNew == nil {
+		return false
+	}
+	oldObj := e.ObjectOld.DeepCopyObject().(client.Object)
+	newObj := e.ObjectNew.DeepCopyObject().(client.Object)
+	// ignore resource version
+	oldObj.SetResourceVersion("")
+	newObj.SetResourceVersion("")
+	// ignore managed fields
+	oldObj.SetManagedFields([]metav1.ManagedFieldsEntry{})
+	newObj.SetManagedFields([]metav1.ManagedFieldsEntry{})
+	oldUnstr, err := runtime.DefaultUnstructuredConverter.ToUnstructured(oldObj)
+	if err != nil {
+		return false
+	}
+	newUnstr, err := runtime.DefaultUnstructuredConverter.ToUnstructured(newObj)
+	if err != nil {
+		return false
+	}
+	return !reflect.DeepEqual(oldUnstr, newUnstr)
+}

main.go

@@ -38,7 +38,6 @@ import (
 
 	mellanoxcomv1alpha1 "github.com/Mellanox/network-operator/api/v1alpha1"
 	"github.com/Mellanox/network-operator/controllers"
-	"github.com/Mellanox/network-operator/pkg/consts"
 	"github.com/Mellanox/network-operator/pkg/migrate"
 	// +kubebuilder:scaffold:imports
 )
@@ -110,11 +109,10 @@ func main() {
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
-	clientConf := ctrl.GetConfigOrDie()
-	clientConf.UserAgent = consts.KubernetesClientUserAgent
-
 	stopCtx := ctrl.SetupSignalHandler()
 
+	clientConf := ctrl.GetConfigOrDie()
+
 	mgr, err := ctrl.NewManager(clientConf, ctrl.Options{
 		Scheme:                 scheme,
 		MetricsBindAddress:     metricsAddr,

pkg/consts/consts.go

@@ -32,5 +32,4 @@ const (
 	NicClusterPolicyResourceName = "nic-cluster-policy"
 	OfedDriverLabel              = "nvidia.com/ofed-driver"
 	StateLabel                   = "nvidia.network-operator.state"
-	KubernetesClientUserAgent    = "nvidia.network-operator"
 )

pkg/state/state_skel.go

@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/yaml"
 
 	"github.com/Mellanox/network-operator/pkg/consts"
 	"github.com/Mellanox/network-operator/pkg/nodeinfo"
@@ -165,16 +164,10 @@ func (s *stateSkel) checkDeleteSupported(obj *unstructured.Unstructured) {
 
 func (s *stateSkel) updateObj(obj *unstructured.Unstructured) error {
 	log.V(consts.LogLevelInfo).Info("Updating Object", "Namespace:", obj.GetNamespace(), "Name:", obj.GetName())
-	applyPatchData, err := yaml.Marshal(obj)
-	if err != nil {
-		log.V(consts.LogLevelError).Error(err, "failed to encode apply patch data")
-		return err
-	}
-	patchUseForce := true
-	// use server-side apply
-	if err := s.client.Patch(context.TODO(), obj, client.RawPatch(types.ApplyPatchType, applyPatchData),
-		client.FieldOwner(consts.KubernetesClientUserAgent),
-		&client.PatchOptions{Force: &patchUseForce}); err != nil {
+	// Note: Some objects may require update of the resource version
+	// TODO: using Patch preserves runtime attributes. In the future consider using patch if relevant
+	desired := obj.DeepCopy()
+	if err := s.client.Update(context.TODO(), desired); err != nil {
 		return errors.Wrap(err, "failed to update resource")
 	}
 	log.V(consts.LogLevelInfo).Info("Object updated successfully")
@@ -204,6 +197,16 @@ func (s *stateSkel) createOrUpdateObjs(
 			return err
 		}
 
+		currentObj := desiredObj.DeepCopy()
+		if err := s.getObj(currentObj); err != nil {
+			// Some error occurred
+			return err
+		}
+
+		if err := s.mergeObjects(desiredObj, currentObj); err != nil {
+			return err
+		}
+
 		// Object found, Update it
 		if err := s.updateObj(desiredObj); err != nil {
 			return err
@@ -266,6 +269,44 @@ func (s *stateSkel) deleteStateRelatedObjects() (bool, error) {
 	return found, nil
 }
 
+func (s *stateSkel) mergeObjects(updated, current *unstructured.Unstructured) error {
+	// Set resource version
+	// ResourceVersion must be passed unmodified back to the server.
+	// ResourceVersion helps the kubernetes API server to implement optimistic concurrency for PUT operations
+	// when two PUT requests are specifying the resourceVersion, one of the PUTs will fail.
+	updated.SetResourceVersion(current.GetResourceVersion())
+
+	gvk := updated.GroupVersionKind()
+	if gvk.Group == "" && gvk.Kind == "ServiceAccount" {
+		return s.mergeServiceAccount(updated, current)
+	}
+	return nil
+}
+
+// For Service Account, keep secrets if exists
+func (s *stateSkel) mergeServiceAccount(updated, current *unstructured.Unstructured) error {
+	curSecrets, ok, err := unstructured.NestedSlice(current.Object, "secrets")
+	if err != nil {
+		return err
+	}
+	if ok {
+		if err := unstructured.SetNestedField(updated.Object, curSecrets, "secrets"); err != nil {
+			return err
+		}
+	}
+
+	curImagePullSecrets, ok, err := unstructured.NestedSlice(current.Object, "imagePullSecrets")
+	if err != nil {
+		return err
+	}
+	if ok {
+		if err := unstructured.SetNestedField(updated.Object, curImagePullSecrets, "imagePullSecrets"); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Iterate over objects and check for their readiness
 func (s *stateSkel) getSyncState(objs []*unstructured.Unstructured) (SyncState, error) {
 	log.V(consts.LogLevelInfo).Info("Checking related object states")