Ensure event ordering in the resource syncer

If a resource is deleted and then quickly re-created, depending
on timing, the delete event might not get processed so consumers
only observe the re-creation event which is not desirable in some
cases. To ensure events are not missed and are processed in order,
a per-resource queue was added to mantain create and delete operations
in place of the separate created and deleted maps.

Signed-off-by: Tom Pantelis <tompantelis@gmail.com>

pkg/syncer/operation_queue_map.go

@@ -0,0 +1,74 @@
+/*
+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 syncer
+
+import (
+	"sync"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+type createOperation *unstructured.Unstructured
+
+type deleteOperation *unstructured.Unstructured
+
+type operationQueueMap struct {
+	mutex  sync.Mutex
+	queues map[string][]any
+}
+
+func (m *operationQueueMap) peek(key string) any {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	q := m.queues[key]
+	if len(q) == 0 {
+		return nil
+	}
+
+	return q[0]
+}
+
+func (m *operationQueueMap) remove(key string, op any) bool {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	q := m.queues[key]
+	if len(q) == 0 {
+		return false
+	}
+
+	if op != nil && op == q[0] {
+		m.queues[key] = q[1:]
+	}
+
+	newLen := len(m.queues[key])
+	if newLen == 0 {
+		delete(m.queues, key)
+	}
+
+	return newLen > 0
+}
+
+func (m *operationQueueMap) add(key string, v any) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	m.queues[key] = append(m.queues[key], v)
+}

pkg/syncer/resource_syncer.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"reflect"
-	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -199,8 +198,7 @@ type resourceSyncer struct {
 	informer          cache.Controller
 	store             cache.Store
 	config            ResourceSyncerConfig
-	deleted           sync.Map
-	created           sync.Map
+	operationQueues   *operationQueueMap
 	stopped           chan struct{}
 	syncCounter       *prometheus.GaugeVec
 	stopCh            <-chan struct{}
@@ -271,7 +269,10 @@ func NewResourceSyncerWithSharedInformer(config *ResourceSyncerConfig, informer
 
 func newResourceSyncer(config *ResourceSyncerConfig) (*resourceSyncer, error) {
 	syncer := &resourceSyncer{
-		config:            *config,
+		config: *config,
+		operationQueues: &operationQueueMap{
+			queues: map[string][]any{},
+		},
 		stopped:           make(chan struct{}),
 		log:               log.Logger{Logger: logf.Log.WithName("ResourceSyncer")},
 		missingNamespaces: map[string]set.Set[string]{},
@@ -496,7 +497,7 @@ func (r *resourceSyncer) doReconcile(resourceLister func() []runtime.Object) {
 		}
 
 		obj := resourceUtil.MustToUnstructuredUsingScheme(resource, r.config.Scheme)
-		r.deleted.Store(key, obj)
+		r.operationQueues.add(key, deleteOperation(obj))
 		r.workQueue.Enqueue(obj)
 	}
 }
@@ -518,27 +519,56 @@ func (r *resourceSyncer) processNextWorkItem(key, name, ns string) (bool, error)
 		return false, nil
 	}
 
-	obj, exists, err := r.store.GetByKey(key)
-	if err != nil {
-		return true, errors.Wrapf(err, "error retrieving resource %q", key)
+	var (
+		requeue bool
+		err     error
+	)
+
+	resourceOp := r.operationQueues.peek(key)
+
+	switch t := resourceOp.(type) {
+	case deleteOperation:
+		requeue, err = r.handleDeleted(key, t)
+	case createOperation:
+		requeue, err = r.handleCreatedOrUpdated(key, t)
+	default:
+		requeue, err = r.handleCreatedOrUpdated(key, nil)
 	}
 
-	if !exists {
-		return r.handleDeleted(key)
+	// If not re-queueing the current operation then remove it from the operation queue. If there's another operation queued
+	// then add the key back to the work queue, so it's processed later. Note that we don't simply return true to re-queue
+	// b/c we're not retrying the current operation, and we don't want the re-queue limit to be reached.
+	if !requeue && r.operationQueues.remove(key, resourceOp) {
+		r.workQueue.Enqueue(cache.ExplicitKey(key))
 	}
 
-	resource := r.assertUnstructured(obj)
+	return requeue, err
+}
+
+func (r *resourceSyncer) handleCreatedOrUpdated(key string, created *unstructured.Unstructured) (bool, error) {
+	resource := created
 
 	op := Update
-	_, found := r.created.Load(key)
-	if found {
+	if created != nil {
 		op = Create
 	}
 
+	obj, exists, err := r.store.GetByKey(key)
+	if err != nil {
+		return true, errors.Wrapf(err, "error retrieving resource %q", key)
+	}
+
+	// Use the latest resource from the cache regardless of the operation. If it doesn't exist, for a create operation, this means
+	// a deletion occurred afterward, in which case we'll process the 'created' resource.
+	if exists {
+		resource = r.assertUnstructured(obj)
+	} else if op == Update {
+		return false, nil
+	}
+
 	r.log.V(log.LIBTRACE).Infof("Syncer %q retrieved %sd resource %q: %#v", r.config.Name, op, resource.GetName(), resource)
 
 	if !r.shouldSync(resource) {
-		r.created.Delete(key)
 		return false, nil
 	}
 
@@ -575,25 +605,12 @@ func (r *resourceSyncer) processNextWorkItem(key, name, ns string) (bool, error)
 		r.log.V(log.LIBDEBUG).Infof("Syncer %q successfully synced %q", r.config.Name, resource.GetName())
 	}
 
-	if !requeue {
-		r.created.Delete(key)
-	}
-
 	return requeue, nil
 }
 
-func (r *resourceSyncer) handleDeleted(key string) (bool, error) {
+func (r *resourceSyncer) handleDeleted(key string, deletedResource *unstructured.Unstructured) (bool, error) {
 	r.log.V(log.LIBDEBUG).Infof("Syncer %q informed of deleted resource %q", r.config.Name, key)
 
-	obj, found := r.deleted.Load(key)
-	if !found {
-		r.log.V(log.LIBDEBUG).Infof("Syncer %q: resource %q not found in deleted object cache", r.config.Name, key)
-		return false, nil
-	}
-
-	r.deleted.Delete(key)
-
-	deletedResource := r.assertUnstructured(obj)
 	if !r.shouldSync(deletedResource) {
 		return false, nil
 	}
@@ -613,7 +630,6 @@ func (r *resourceSyncer) handleDeleted(key string) (bool, error) {
 		}
 
 		if err != nil || r.onSuccessfulSync(resource, transformed, Delete) {
-			r.deleted.Store(key, deletedResource)
 			return true, errors.Wrapf(err, "error deleting resource %q", key)
 		}
 
@@ -630,10 +646,6 @@ func (r *resourceSyncer) handleDeleted(key string) (bool, error) {
 		}
 	}
 
-	if requeue {
-		r.deleted.Store(key, deletedResource)
-	}
-
 	return requeue, nil
 }
 
@@ -687,14 +699,16 @@ func (r *resourceSyncer) onSuccessfulSync(resource, converted runtime.Object, op
 	return r.config.OnSuccessfulSync(converted, op)
 }
 
-func (r *resourceSyncer) onCreate(resource interface{}) {
-	if !r.shouldProcess(resource.(*unstructured.Unstructured), Create) {
+func (r *resourceSyncer) onCreate(obj interface{}) {
+	resource := r.assertUnstructured(obj)
+
+	if !r.shouldProcess(resource, Create) {
 		return
 	}
 
 	key, _ := cache.MetaNamespaceKeyFunc(resource)
-	v := true
-	r.created.Store(key, &v)
+
+	r.operationQueues.add(key, createOperation(resource))
 	r.workQueue.Enqueue(resource)
 }
 
@@ -729,8 +743,8 @@ func (r *resourceSyncer) onDelete(obj interface{}) {
 	}
 
 	key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	r.deleted.Store(key, resource)
 
+	r.operationQueues.add(key, deleteOperation(resource))
 	r.workQueue.Enqueue(obj)
 }
 

pkg/syncer/resource_syncer_test.go

@@ -77,6 +77,7 @@ var _ = Describe("Resource Syncer", func() {
 	Describe("Trim Resource Fields", testTrimResourceFields)
 	Describe("With SharedInformer", testWithSharedInformer)
 	Describe("With missing namespace", testWithMissingNamespace)
+	Describe("Event Ordering", testEventOrdering)
 })
 
 func testReconcileLocalToRemote() {
@@ -536,7 +537,7 @@ func testTransformFunction() {
 				transformFuncRet.Store(transformed)
 				expOperation <- op
 
-				return ret, true
+				return ret, ret == nil
 			}
 		})
 
@@ -1296,6 +1297,69 @@ func testWithMissingNamespace() {
 	})
 }
 
+func testEventOrdering() {
+	d := newTestDriver(test.LocalNamespace, "", syncer.LocalToRemote)
+
+	var opChan chan syncer.Operation
+
+	BeforeEach(func() {
+		opChan = make(chan syncer.Operation, 20)
+
+		d.config.Transform = func(from runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
+			opChan <- op
+			return from, false
+		}
+	})
+
+	When("a create occurs immediately following a delete", func() {
+		It("should process both events in order", func() {
+			first := test.CreateResource(d.sourceClient, d.resource)
+			d.federator.VerifyDistribute(first)
+			Eventually(opChan).Should(Receive(Equal(syncer.Create)))
+
+			d.resource = test.NewPodWithImage(d.config.SourceNamespace, "apache")
+			Expect(d.sourceClient.Delete(context.TODO(), first.GetName(), metav1.DeleteOptions{})).To(Succeed())
+			second := test.CreateResource(d.sourceClient, d.resource)
+
+			Eventually(opChan).Should(Receive(Equal(syncer.Delete)))
+			Eventually(opChan).Should(Receive(Equal(syncer.Create)))
+			Consistently(opChan).ShouldNot(Receive())
+
+			d.federator.VerifyDelete(first)
+			d.federator.VerifyDistribute(second)
+		})
+	})
+
+	When("a delete occurs immediately following a create", func() {
+		It("should process both events in order", func() {
+			r := test.CreateResource(d.sourceClient, d.resource)
+			Expect(d.sourceClient.Delete(context.TODO(), r.GetName(), metav1.DeleteOptions{})).To(Succeed())
+
+			Eventually(opChan).Should(Receive(Equal(syncer.Create)))
+			Eventually(opChan).Should(Receive(Equal(syncer.Delete)))
+			Consistently(opChan).ShouldNot(Receive())
+
+			d.federator.VerifyDelete(r)
+			d.federator.VerifyDistribute(r)
+		})
+	})
+
+	When("a delete occurs immediately following an update", func() {
+		It("should process both events in order", func() {
+			d.federator.VerifyDistribute(test.CreateResource(d.sourceClient, d.resource))
+			Eventually(opChan).Should(Receive(Equal(syncer.Create)))
+
+			d.federator.VerifyDistribute(test.UpdateResource(d.sourceClient,
+				test.NewPodWithImage(d.config.SourceNamespace, "apache")))
+			Expect(d.sourceClient.Delete(context.TODO(), d.resource.GetName(), metav1.DeleteOptions{})).To(Succeed())
+
+			Eventually(opChan).Should(Receive(Equal(syncer.Update)))
+			Eventually(opChan).Should(Receive(Equal(syncer.Delete)))
+			Consistently(opChan).ShouldNot(Receive())
+		})
+	})
+}
+
 func assertResourceList(actual []runtime.Object, expected ...*corev1.Pod) {
 	expSpecs := map[string]*corev1.PodSpec{}
 	for i := range expected {