perf: skip infra ir and status subscription for followers (envoyproxy#7145)

Signed-off-by: Rudrakh Panigrahi <rudrakh97@gmail.com>

Author: rudrakhp

internal/cmd/server.go

@@ -237,15 +237,9 @@ func startRunners(ctx context.Context, cfg *config.Server) (err error) {
 	// Wait until done
 	<-ctx.Done()
 
-	// Close messages
-	closeChannels := []interface{ Close() }{
-		channels.pResources,
-		channels.xdsIR,
-		channels.infraIR,
-	}
-	for _, ch := range closeChannels {
-		ch.Close()
-	}
+	// Close xdsIR channel
+	// No need to close infraIR and pResources channels since they are already closed
+	channels.xdsIR.Close()
 
 	cfg.Logger.Info("runners are shutting down")
 	for _, r := range runners {

internal/infrastructure/runner/runner.go

@@ -57,8 +57,10 @@ func (r *Runner) Start(ctx context.Context) (err error) {
 		return err
 	}
 
-	sub := r.InfraIR.Subscribe(ctx)
-	initInfra := func() {
+	// This is a blocking function that subscribes to the infraIR and initializes the infrastructure.
+	subscribeInitInfraAndCloseInfraIRMessage := func() {
+		// Subscribe and Close in same goroutine to avoid race condition.
+		sub := r.InfraIR.Subscribe(ctx)
 		go r.subscribeToProxyInfraIR(ctx, sub)
 
 		// Enable global ratelimit if it has been configured.
@@ -73,6 +75,9 @@ func (r *Runner) Start(ctx context.Context) (err error) {
 			}()
 		}
 		r.Logger.Info("started")
+		<-ctx.Done()
+		r.InfraIR.Close()
+		r.Logger.Info("shutting down")
 	}
 
 	// When leader election is active, infrastructure initialization occurs only upon acquiring leadership
@@ -82,14 +87,18 @@ func (r *Runner) Start(ctx context.Context) (err error) {
 		go func() {
 			select {
 			case <-ctx.Done():
+				// As a follower EG instance close infraIR when the context is done.
+				r.InfraIR.Close()
 				return
 			case <-r.Elected:
-				initInfra()
+				// As a leader EG instance subscribe to infraIR to initialize the infrastructure and Close when the context is done.
+				subscribeInitInfraAndCloseInfraIRMessage()
 			}
 		}()
-		return
+	} else {
+		// Since leader election is disabled subscribe to infraIR to initialize the infrastructure and Close when the context is done.
+		go subscribeInitInfraAndCloseInfraIRMessage()
 	}
-	initInfra()
 	return
 }
 

internal/message/types.go

@@ -70,6 +70,7 @@ func (p *ProviderResources) Close() {
 	p.GatewayAPIResources.Close()
 	p.GatewayAPIStatuses.Close()
 	p.PolicyStatuses.Close()
+	p.ExtensionStatuses.Close()
 }
 
 // GatewayAPIStatuses contains gateway API resources statuses
@@ -122,6 +123,10 @@ func (p *PolicyStatuses) Close() {
 	p.ExtensionPolicyStatuses.Close()
 }
 
+func (e *ExtensionStatuses) Close() {
+	e.BackendStatuses.Close()
+}
+
 // XdsIR message
 type XdsIR struct {
 	watchable.Map[string, *ir.Xds]

internal/provider/kubernetes/controller.go

@@ -199,23 +199,36 @@ func newGatewayAPIController(ctx context.Context, mgr manager.Manager, cfg *conf
 		return fmt.Errorf("error watching resources: %w", err)
 	}
 
-	// Do not call .Subscribe() inside Goroutine since it is supposed to be called from the same
-	// Goroutine where Close() is called.
-	r.subscribeToResources(ctx)
+	// This is a blocking function that subscribes to the resource updates and updates the status.
+	subscribeUpdateStatusAndCloseResources := func() {
+		// Subscribe to resource updates
+		r.subscribeToResources(ctx)
+		// Update status
+		go r.subscribeAndUpdateStatus(ctx, cfg.EnvoyGateway.ExtensionManager != nil)
+		r.log.Info("started")
+		// Close resources if the context is done.
+		<-ctx.Done()
+		r.resources.Close()
+		r.log.Info("shutting down")
+	}
 
 	// When leader election is enabled, only subscribe to status updates upon acquiring leadership.
 	if cfg.EnvoyGateway.Provider.Type == egv1a1.ProviderTypeKubernetes &&
 		!ptr.Deref(cfg.EnvoyGateway.Provider.Kubernetes.LeaderElection.Disable, false) {
 		go func() {
 			select {
 			case <-ctx.Done():
+				// As a follower EG instance close resources when the context is done.
+				r.resources.Close()
 				return
 			case <-cfg.Elected:
-				r.subscribeAndUpdateStatus(ctx, cfg.EnvoyGateway.ExtensionManager != nil)
+				// As a leader EG instance subscribe to resource updates and Close resources when the context is done.
+				subscribeUpdateStatusAndCloseResources()
 			}
 		}()
 	} else {
-		r.subscribeAndUpdateStatus(ctx, cfg.EnvoyGateway.ExtensionManager != nil)
+		// Since leader election is disabled subscribe to resource updates and Close resources when the context is done.
+		go subscribeUpdateStatusAndCloseResources()
 	}
 	return nil
 }