Ensure Gateway resource is deleted on shutdown

main.go

@@ -88,6 +88,7 @@ type componentsType struct {
 	gatewayPod         *pod.GatewayPod
 	submSpec           types.SubmarinerSpecification
 	stopCh             <-chan struct{}
+	waitGroup          sync.WaitGroup
 }
 
 const (
@@ -182,7 +183,7 @@ func main() {
 	components.gatewayPod, err = pod.NewGatewayPod(k8sClient)
 	logger.FatalOnError(err, "Error creating a handler to update the gateway pod")
 
-	components.cableEngineSyncer.Run(components.stopCh)
+	components.runAsync(components.cableEngineSyncer.Run)
 
 	if !airGapped {
 		components.initPublicIPWatcher(k8sClient, submarinerClient, localEndpoint)
@@ -214,6 +215,8 @@ func main() {
 	if err := httpServer.Shutdown(context.TODO()); err != nil {
 		logger.Errorf(err, "Error shutting down metrics HTTP server")
 	}
+
+	components.waitGroup.Wait()
 }
 
 func isAirGappedDeployment() bool {
@@ -317,6 +320,15 @@ func (c *componentsType) initPublicIPWatcher(k8sClient kubernetes.Interface, sub
 	c.publicIPWatcher = endpoint.NewPublicIPWatcher(publicIPConfig)
 }
 
+func (c *componentsType) runAsync(run func(<-chan struct{})) {
+	c.waitGroup.Add(1)
+
+	go func() {
+		defer c.waitGroup.Done()
+		run(c.stopCh)
+	}()
+}
+
 func submarinerClusterFrom(submSpec *types.SubmarinerSpecification) *types.SubmarinerCluster {
 	return &types.SubmarinerCluster{
 		ID: submSpec.ClusterID,

pkg/cableengine/syncer/syncer.go

@@ -83,10 +83,8 @@ func NewGatewaySyncer(engine cableengine.Engine, client v1typed.GatewayInterface
 }
 
 func (gs *GatewaySyncer) Run(stopCh <-chan struct{}) {
-	go func() {
-		wait.Until(gs.syncGatewayStatus, GatewayUpdateInterval, stopCh)
-		gs.CleanupGatewayEntry()
-	}()
+	wait.Until(gs.syncGatewayStatus, GatewayUpdateInterval, stopCh)
+	gs.CleanupGatewayEntry()
 
 	logger.Info("CableEngine syncer started")
 }

pkg/cableengine/syncer/syncer_test.go

@@ -560,7 +560,7 @@ func (t *testDriver) run() {
 	go informer.Run(t.stopInformer)
 	Expect(cache.WaitForCacheSync(t.stopInformer, informer.HasSynced)).To(BeTrue())
 
-	t.syncer.Run(t.stopSyncer)
+	go t.syncer.Run(t.stopSyncer)
 
 	Expect(t.healthChecker.Start(t.stopSyncer)).To(Succeed())
 }

test/e2e/redundancy/gateway_failover.go

@@ -104,7 +104,7 @@ func testGatewayPodRestartScenario(f *subFramework.Framework) {
 	By(fmt.Sprintf("Found new submariner gateway pod %q", newGatewayPod.Name))
 
 	By(fmt.Sprintf("Waiting for the gateway to be up and connected %q", newGatewayPod.Name))
-	f.AwaitGatewayFullyConnected(framework.ClusterIndex(primaryCluster), resource.EnsureValidName(activeGateway.Name))
+	AwaitNewSubmarinerGatewayFullyConnected(f, framework.ClusterIndex(primaryCluster), activeGateway.Name, activeGateway.UID)
 
 	By(fmt.Sprintf("Verifying TCP connectivity from gateway node on %q to gateway node on %q", secondaryClusterName, primaryClusterName))
 	subFramework.VerifyDatapathConnectivity(tcp.ConnectivityTestParams{
@@ -147,6 +147,21 @@ func AwaitNewSubmarinerGatewayPod(f *subFramework.Framework, cluster framework.C
 	}).(*v1.Pod)
 }
 
+func AwaitNewSubmarinerGatewayFullyConnected(f *subFramework.Framework, cluster framework.ClusterIndex, name string,
+	prevPodUID types.UID,
+) *subv1.Gateway {
+	return framework.AwaitUntil("await new submariner gateway", func() (interface{}, error) {
+		return f.AwaitGatewayFullyConnected(cluster, resource.EnsureValidName(name)), nil
+	}, func(result interface{}) (bool, string, error) {
+		gw := result.(*subv1.Gateway)
+		if gw.ObjectMeta.UID != prevPodUID {
+			return true, "", nil
+		}
+
+		return false, fmt.Sprintf("Expecting new gateway (UID %q matches previous instance)", prevPodUID), nil
+	}).(*subv1.Gateway)
+}
+
 func defaultEndpointType() tcp.EndpointType {
 	if framework.TestContext.GlobalnetEnabled {
 		return tcp.GlobalIP