Automated backport of #895: Add support for resource syncer to use a shared informer #913: Use informer to retry failed resources due to missing

pkg/fake/basic_reactors_test.go

@@ -74,6 +74,32 @@ var _ = Describe("Create", func() {
 			Expect(err).To(HaveOccurred())
 		})
 	})
+
+	Context("with namespace verification", func() {
+		BeforeEach(func() {
+			fake.AddVerifyNamespaceReactor(&t.client.Fake, "pods")
+		})
+
+		When("the namespace does not exist", func() {
+			It("should return an error", func() {
+				_, err := t.doCreate(t.pod)
+				missing, ns := resource.IsMissingNamespaceErr(err)
+				Expect(missing).To(BeTrue())
+				Expect(ns).To(Equal(testNamespace))
+			})
+		})
+
+		When("the namespace does exist", func() {
+			It("should succeed", func() {
+				_, err := t.client.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: testNamespace},
+				}, metav1.CreateOptions{})
+				Expect(err).To(Succeed())
+
+				t.assertCreateSuccess(t.pod)
+			})
+		})
+	})
 })
 
 var _ = Describe("Update", func() {

pkg/fake/verify_namespace_reactor.go

@@ -0,0 +1,50 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fake
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/testing"
+)
+
+func AddVerifyNamespaceReactor(f *testing.Fake, resources ...string) {
+	f.Lock()
+	defer f.Unlock()
+
+	reactors := f.ReactionChain[0:]
+
+	react := func(a testing.Action) (bool, runtime.Object, error) {
+		action := a.(testing.CreateAction)
+
+		namespace := action.GetNamespace()
+
+		_, err := invokeReactors(testing.NewGetAction(corev1.SchemeGroupVersion.WithResource("namespaces"), "", namespace),
+			reactors)
+		if err != nil {
+			return true, nil, err
+		}
+
+		return false, nil, nil
+	}
+
+	for _, res := range resources {
+		f.PrependReactor("create", res, react)
+	}
+}

pkg/federate/fake/federator.go

@@ -21,6 +21,8 @@ package fake
 
 import (
 	"context"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	. "github.com/onsi/gomega"
@@ -30,26 +32,46 @@ import (
 )
 
 type Federator struct {
+	lock             sync.Mutex
 	distribute       chan *unstructured.Unstructured
 	delete           chan *unstructured.Unstructured
-	FailOnDistribute error
-	FailOnDelete     error
-	ResetOnFailure   bool
+	failOnDistribute error
+	failOnDelete     error
+	ResetOnFailure   atomic.Bool
 }
 
 func New() *Federator {
-	return &Federator{
-		distribute:     make(chan *unstructured.Unstructured, 100),
-		delete:         make(chan *unstructured.Unstructured, 100),
-		ResetOnFailure: true,
+	f := &Federator{
+		distribute: make(chan *unstructured.Unstructured, 100),
+		delete:     make(chan *unstructured.Unstructured, 100),
 	}
+	f.ResetOnFailure.Store(true)
+
+	return f
+}
+
+func (f *Federator) FailOnDistribute(err error) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.failOnDistribute = err
+}
+
+func (f *Federator) FailOnDelete(err error) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.failOnDelete = err
 }
 
 func (f *Federator) Distribute(_ context.Context, obj runtime.Object) error {
-	err := f.FailOnDistribute
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	err := f.failOnDistribute
 	if err != nil {
-		if f.ResetOnFailure {
-			f.FailOnDistribute = nil
+		if f.ResetOnFailure.Load() {
+			f.failOnDistribute = nil
 		}
 
 		return err
@@ -61,10 +83,13 @@ func (f *Federator) Distribute(_ context.Context, obj runtime.Object) error {
 }
 
 func (f *Federator) Delete(_ context.Context, obj runtime.Object) error {
-	err := f.FailOnDelete
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	err := f.failOnDelete
 	if err != nil {
-		if f.ResetOnFailure {
-			f.FailOnDelete = nil
+		if f.ResetOnFailure.Load() {
+			f.failOnDelete = nil
 		}
 
 		return err

pkg/resource/resource_test.go

@@ -22,6 +22,10 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/submariner-io/admiral/pkg/resource"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/utils/ptr"
 )
 
 var _ = Describe("EnsureValidName", func() {
@@ -44,3 +48,60 @@ var _ = Describe("EnsureValidName", func() {
 		})
 	})
 })
+
+var _ = Describe("TrimManagedFields", func() {
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "pod",
+			Namespace:   "ns",
+			Labels:      map[string]string{"app": "test"},
+			Annotations: map[string]string{"foo": "bar"},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Image: "image",
+					Name:  "httpd",
+				},
+			},
+		},
+	}
+
+	When("the object has metadata", func() {
+		It("should succeed and trim the ManagedFields", func() {
+			podWithMF := pod.DeepCopy()
+			podWithMF.ManagedFields = []metav1.ManagedFieldsEntry{
+				{
+					Manager:    "kubectl",
+					Operation:  metav1.ManagedFieldsOperationApply,
+					APIVersion: "v1",
+					Time:       ptr.To(metav1.Now()),
+					FieldsType: "FieldsV1",
+					FieldsV1:   ptr.To(metav1.FieldsV1{}),
+				},
+			}
+
+			retObj, err := resource.TrimManagedFields(podWithMF)
+			Expect(err).To(Succeed())
+			Expect(podWithMF).To(Equal(pod))
+			Expect(retObj).To(BeIdenticalTo(podWithMF))
+
+			_, err = resource.TrimManagedFields(podWithMF)
+			Expect(err).To(Succeed())
+			Expect(podWithMF).To(Equal(pod))
+		})
+	})
+
+	When("the object does not have metadata", func() {
+		It("should succeed and return the object", func() {
+			obj := &cache.DeletedFinalStateUnknown{
+				Key: "key",
+				Obj: pod,
+			}
+
+			retObj, err := resource.TrimManagedFields(obj)
+			Expect(err).To(Succeed())
+			Expect(retObj).To(Equal(obj))
+		})
+	})
+})

pkg/resource/util.go

@@ -125,3 +125,14 @@ func ToJSON(o any) string {
 	out, _ := json.MarshalIndent(o, "", "  ")
 	return string(out)
 }
+
+// TrimManagedFields is a cache.TransformFunc that removes the ManagedFields metadata field if present. Note that if 'obj' does not
+// implement metav1.Object then it is ignored and no error is returned.
+func TrimManagedFields(obj interface{}) (interface{}, error) {
+	objMeta, err := meta.Accessor(obj)
+	if err == nil && len(objMeta.GetManagedFields()) > 0 {
+		objMeta.SetManagedFields(nil)
+	}
+
+	return obj, nil
+}

pkg/syncer/broker/syncer.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/ptr"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
@@ -134,6 +135,9 @@ type SyncerConfig struct {
 
 	// Scheme used to convert resource objects. By default the global k8s Scheme is used.
 	Scheme *runtime.Scheme
+
+	// NamespaceInformer if specified, used to retry local resources that initially failed due to missing namespace.
+	NamespaceInformer cache.SharedInformer
 }
 
 type Syncer struct {
@@ -254,6 +258,7 @@ func NewSyncer(config SyncerConfig) (*Syncer, error) { //nolint:gocritic // Mini
 			Scheme:              config.Scheme,
 			ResyncPeriod:        rc.BrokerResyncPeriod,
 			SyncCounter:         syncCounter,
+			NamespaceInformer:   config.NamespaceInformer,
 		})
 		if err != nil {
 			return nil, errors.Wrap(err, "error creating remote resource syncer")