Automated backport of #2399: Fix race when out-of-order RemoteEndpoint events are seen

pkg/event/registry.go

@@ -23,15 +23,17 @@ import (
 	"github.com/submariner-io/admiral/pkg/log"
 	submV1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
 	k8sV1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8serrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 type Registry struct {
-	name          string
-	networkPlugin string
-	eventHandlers []Handler
+	name                    string
+	networkPlugin           string
+	eventHandlers           []Handler
+	remoteEndpointTimeStamp map[string]v1.Time
 }
 
 var logger = log.Logger{Logger: logf.Log.WithName("EventRegistry")}
@@ -40,9 +42,10 @@ var logger = log.Logger{Logger: logf.Log.WithName("EventRegistry")}
 // Handlers that match the given networkPlugin name.
 func NewRegistry(name, networkPlugin string) *Registry {
 	return &Registry{
-		name:          name,
-		networkPlugin: networkPlugin,
-		eventHandlers: []Handler{},
+		name:                    name,
+		networkPlugin:           networkPlugin,
+		eventHandlers:           []Handler{},
+		remoteEndpointTimeStamp: map[string]v1.Time{},
 	}
 }
 
@@ -122,9 +125,23 @@ func (er *Registry) LocalEndpointRemoved(endpoint *submV1.Endpoint) error {
 }
 
 func (er *Registry) RemoteEndpointCreated(endpoint *submV1.Endpoint) error {
-	return er.invokeHandlers("RemoteEndpointCreated", func(h Handler) error {
+	lastProcessedTime, ok := er.remoteEndpointTimeStamp[endpoint.Spec.ClusterID]
+
+	if ok && lastProcessedTime.After(endpoint.CreationTimestamp.Time) {
+		logger.Infof("Ignoring new remote %#v since a later endpoint was already"+
+			"processed", endpoint)
+		return nil
+	}
+
+	err := er.invokeHandlers("RemoteEndpointCreated", func(h Handler) error {
 		return h.RemoteEndpointCreated(endpoint) //nolint:wrapcheck  // Let the caller wrap it
 	})
+
+	if err == nil {
+		er.remoteEndpointTimeStamp[endpoint.Spec.ClusterID] = endpoint.CreationTimestamp
+	}
+
+	return err
 }
 
 func (er *Registry) RemoteEndpointUpdated(endpoint *submV1.Endpoint) error {
@@ -134,6 +151,16 @@ func (er *Registry) RemoteEndpointUpdated(endpoint *submV1.Endpoint) error {
 }
 
 func (er *Registry) RemoteEndpointRemoved(endpoint *submV1.Endpoint) error {
+	lastProcessedTime, ok := er.remoteEndpointTimeStamp[endpoint.Spec.ClusterID]
+
+	if ok && lastProcessedTime.After(endpoint.CreationTimestamp.Time) {
+		logger.Infof("Ignoring deleted remote %#v since a later endpoint was already"+
+			"processed", endpoint)
+		return nil
+	}
+
+	delete(er.remoteEndpointTimeStamp, endpoint.Spec.ClusterID)
+
 	return er.invokeHandlers("RemoteEndpointRemoved", func(h Handler) error {
 		return h.RemoteEndpointRemoved(endpoint) //nolint:wrapcheck  // Let the caller wrap it
 	})

pkg/event/registry_test.go

@@ -20,6 +20,7 @@ package event_test
 
 import (
 	"errors"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -101,6 +102,41 @@ var _ = Describe("Event Registry", func() {
 				}
 			})
 		})
+
+		When("the RemoteEndpointCreated notification is fired out of order", func() {
+			It("should skip processing the stale event", func() {
+				now := time.Now()
+				aFewSecondsLater := now.Add(2 * time.Second)
+				staleEndpoint := &submV1.Endpoint{
+					ObjectMeta: v1meta.ObjectMeta{Name: "endpoint1", CreationTimestamp: v1meta.NewTime(now)},
+					Spec:       submV1.EndpointSpec{ClusterID: "eastCluster"},
+				}
+				latestEndpoint := &submV1.Endpoint{
+					ObjectMeta: v1meta.ObjectMeta{Name: "endpoint1", CreationTimestamp: v1meta.NewTime(aFewSecondsLater)},
+					Spec:       submV1.EndpointSpec{ClusterID: "eastCluster"},
+				}
+
+				event := testing.TestEvent{
+					Name:      testing.EvRemoteEndpointCreated,
+					Parameter: latestEndpoint,
+				}
+
+				err := registry.RemoteEndpointCreated(latestEndpoint)
+				Expect(err).To(Succeed())
+
+				for _, h := range matchingHandlers {
+					event.Handler = h.Name
+					Expect(allTestEvents).To(Receive(Equal(event)))
+				}
+
+				err = registry.RemoteEndpointCreated(staleEndpoint)
+				Expect(err).To(Succeed())
+
+				for range matchingHandlers {
+					Expect(allTestEvents).NotTo(Receive(Equal(event)))
+				}
+			})
+		})
 	})
 })
 

pkg/routeagent_driver/handlers/kubeproxy/endpoint_handler.go

@@ -104,14 +104,6 @@ func (kp *SyncHandler) RemoteEndpointCreated(endpoint *submV1.Endpoint) error {
 	kp.syncHandlerMutex.Lock()
 	defer kp.syncHandlerMutex.Unlock()
 
-	lastProcessedTime, ok := kp.remoteEndpointTimeStamp[endpoint.Spec.ClusterID]
-
-	if ok && lastProcessedTime.After(endpoint.CreationTimestamp.Time) {
-		logger.Infof("Ignoring new remote %#v since a later endpoint was already"+
-			"processed", endpoint)
-		return nil
-	}
-
 	for _, inputCidrBlock := range endpoint.Spec.Subnets {
 		if !kp.remoteSubnets.Has(inputCidrBlock) {
 			kp.remoteSubnets.Insert(inputCidrBlock)
@@ -131,8 +123,6 @@ func (kp *SyncHandler) RemoteEndpointCreated(endpoint *submV1.Endpoint) error {
 	kp.updateRoutingRulesForHostNetworkSupport(endpoint.Spec.Subnets, Add)
 	kp.updateIptableRulesForInterClusterTraffic(endpoint.Spec.Subnets, Add)
 
-	kp.remoteEndpointTimeStamp[endpoint.Spec.ClusterID] = endpoint.CreationTimestamp
-
 	return nil
 }
 
@@ -144,16 +134,6 @@ func (kp *SyncHandler) RemoteEndpointRemoved(endpoint *submV1.Endpoint) error {
 	kp.syncHandlerMutex.Lock()
 	defer kp.syncHandlerMutex.Unlock()
 
-	lastProcessedTime, ok := kp.remoteEndpointTimeStamp[endpoint.Spec.ClusterID]
-
-	if ok && lastProcessedTime.After(endpoint.CreationTimestamp.Time) {
-		logger.Infof("Ignoring deleted remote %#v since a later endpoint was already"+
-			"processed", endpoint)
-		return nil
-	}
-
-	delete(kp.remoteEndpointTimeStamp, endpoint.Spec.ClusterID)
-
 	for _, inputCidrBlock := range endpoint.Spec.Subnets {
 		kp.remoteSubnets.Delete(inputCidrBlock)
 		delete(kp.remoteSubnetGw, inputCidrBlock)

pkg/routeagent_driver/handlers/kubeproxy/kp_iptables.go

@@ -30,7 +30,6 @@ import (
 	"github.com/submariner-io/submariner/pkg/event"
 	"github.com/submariner-io/submariner/pkg/netlink"
 	cniapi "github.com/submariner-io/submariner/pkg/routeagent_driver/cni"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
@@ -41,11 +40,10 @@ type SyncHandler struct {
 	localClusterCidr []string
 	localServiceCidr []string
 
-	remoteSubnets           sets.Set[string]
-	remoteSubnetGw          map[string]net.IP
-	remoteVTEPs             sets.Set[string]
-	routeCacheGWNode        sets.Set[string]
-	remoteEndpointTimeStamp map[string]v1.Time
+	remoteSubnets    sets.Set[string]
+	remoteSubnetGw   map[string]net.IP
+	remoteVTEPs      sets.Set[string]
+	routeCacheGWNode sets.Set[string]
 
 	syncHandlerMutex     sync.Mutex
 	isGatewayNode        bool
@@ -63,17 +61,16 @@ var logger = log.Logger{Logger: logf.Log.WithName("KubeProxy")}
 
 func NewSyncHandler(localClusterCidr, localServiceCidr []string) *SyncHandler {
 	return &SyncHandler{
-		localClusterCidr:        cidr.ExtractIPv4Subnets(localClusterCidr),
-		localServiceCidr:        cidr.ExtractIPv4Subnets(localServiceCidr),
-		localCableDriver:        "",
-		remoteSubnets:           sets.New[string](),
-		remoteSubnetGw:          map[string]net.IP{},
-		remoteEndpointTimeStamp: map[string]v1.Time{},
-		remoteVTEPs:             sets.New[string](),
-		routeCacheGWNode:        sets.New[string](),
-		isGatewayNode:           false,
-		wasGatewayPreviously:    false,
-		netLink:                 netlink.New(),
+		localClusterCidr:     cidr.ExtractIPv4Subnets(localClusterCidr),
+		localServiceCidr:     cidr.ExtractIPv4Subnets(localServiceCidr),
+		localCableDriver:     "",
+		remoteSubnets:        sets.New[string](),
+		remoteSubnetGw:       map[string]net.IP{},
+		remoteVTEPs:          sets.New[string](),
+		routeCacheGWNode:     sets.New[string](),
+		isGatewayNode:        false,
+		wasGatewayPreviously: false,
+		netLink:              netlink.New(),
 	}
 }
 

pkg/routeagent_driver/handlers/ovn/handler.go

@@ -139,14 +139,6 @@ func (ovn *Handler) RemoteEndpointCreated(endpoint *submV1.Endpoint) error {
 	ovn.mutex.Lock()
 	defer ovn.mutex.Unlock()
 
-	existingEndPoint, ok := ovn.remoteEndpoints[endpoint.Spec.ClusterID]
-
-	if ok && existingEndPoint.CreationTimestamp.After(endpoint.CreationTimestamp.Time) {
-		logger.Infof("Ignoring new remote %#v since a later endpoint was already"+
-			"processed", endpoint)
-		return nil
-	}
-
 	ovn.remoteEndpoints[endpoint.Name] = endpoint
 
 	err := ovn.updateHostNetworkDataplane()
@@ -195,14 +187,6 @@ func (ovn *Handler) RemoteEndpointRemoved(endpoint *submV1.Endpoint) error {
 	ovn.mutex.Lock()
 	defer ovn.mutex.Unlock()
 
-	existingEndPoint, ok := ovn.remoteEndpoints[endpoint.Spec.ClusterID]
-
-	if ok && existingEndPoint.CreationTimestamp.After(endpoint.CreationTimestamp.Time) {
-		logger.Infof("Ignoring deleted remote %#v since a later endpoint was already"+
-			"processed", endpoint)
-		return nil
-	}
-
 	delete(ovn.remoteEndpoints, endpoint.Name)
 
 	err := ovn.updateHostNetworkDataplane()