Fix race in GN pod for out-of-order RemoteEndpoint events

pkg/globalnet/controllers/gateway_monitor.go

@@ -37,6 +37,7 @@ import (
 	"github.com/submariner-io/submariner/pkg/netlink"
 	routeAgent "github.com/submariner-io/submariner/pkg/routeagent_driver/constants"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/dynamic"
@@ -47,11 +48,12 @@ import (
 func NewGatewayMonitor(spec Specification, localCIDRs []string, config *watcher.Config) (Interface, error) {
 	// We'll panic if config is nil, this is intentional
 	gatewayMonitor := &gatewayMonitor{
-		baseController: newBaseController(),
-		spec:           spec,
-		isGatewayNode:  atomic.Bool{},
-		localSubnets:   sets.New(localCIDRs...).UnsortedList(),
-		remoteSubnets:  sets.New[string](),
+		baseController:          newBaseController(),
+		spec:                    spec,
+		isGatewayNode:           atomic.Bool{},
+		localSubnets:            sets.New(localCIDRs...).UnsortedList(),
+		remoteSubnets:           sets.New[string](),
+		remoteEndpointTimeStamp: map[string]metav1.Time{},
 	}
 
 	var err error
@@ -143,6 +145,14 @@ func (g *gatewayMonitor) handleCreatedOrUpdatedEndpoint(obj runtime.Object, _ in
 	logger.V(log.DEBUG).Infof("In processNextEndpoint, endpoint info: %+v", endpoint)
 
 	if endpoint.Spec.ClusterID != g.spec.ClusterID {
+		lastProcessedTime, ok := g.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 false
+		}
+
 		logger.V(log.DEBUG).Infof("Endpoint %q, host: %q belongs to a remote cluster",
 			endpoint.Spec.ClusterID, endpoint.Spec.Hostname)
 
@@ -168,6 +178,8 @@ func (g *gatewayMonitor) handleCreatedOrUpdatedEndpoint(obj runtime.Object, _ in
 			}
 		}
 
+		g.remoteEndpointTimeStamp[endpoint.Spec.ClusterID] = endpoint.CreationTimestamp
+
 		return false
 	}
 
@@ -206,6 +218,16 @@ func (g *gatewayMonitor) handleCreatedOrUpdatedEndpoint(obj runtime.Object, _ in
 func (g *gatewayMonitor) handleRemovedEndpoint(obj runtime.Object, _ int) bool {
 	endpoint := obj.(*v1.Endpoint)
 
+	lastProcessedTime, ok := g.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 false
+	}
+
+	delete(g.remoteEndpointTimeStamp, endpoint.Spec.ClusterID)
+
 	logger.V(log.DEBUG).Infof("Informed of removed endpoint for gateway monitor: %v", endpoint)
 
 	hostname, err := os.Hostname()

pkg/globalnet/controllers/types.go

@@ -91,16 +91,17 @@ type baseController struct {
 
 type gatewayMonitor struct {
 	*baseController
-	syncerConfig     *syncer.ResourceSyncerConfig
-	endpointWatcher  watcher.Interface
-	spec             Specification
-	ipt              iptables.Interface
-	isGatewayNode    atomic.Bool
-	nodeName         string
-	localSubnets     []string
-	remoteSubnets    sets.Set[string]
-	controllersMutex sync.Mutex // Protects controllers
-	controllers      []Interface
+	syncerConfig            *syncer.ResourceSyncerConfig
+	endpointWatcher         watcher.Interface
+	remoteEndpointTimeStamp map[string]metav1.Time
+	spec                    Specification
+	ipt                     iptables.Interface
+	isGatewayNode           atomic.Bool
+	nodeName                string
+	localSubnets            []string
+	remoteSubnets           sets.Set[string]
+	controllersMutex        sync.Mutex // Protects controllers
+	controllers             []Interface
 }
 
 type baseSyncerController struct {