fix: use lock when accessing mergeGateways Set (#7124)

* fix: use lock when accessing mergeGateways Set

* accessed in multiple goroutines to map proxy fleet to resource via
  labels

Fixes this panic

```
fatal error: concurrent map read and map write

   4 internal/runtime/maps.fatal({0x3b83570?, 0x8b3de58?})
   5     /opt/hostedtoolcache/go/1.24.7/x64/src/runtime/panic.go:1058 +0x18
   6 k8s.io/apimachinery/pkg/util/sets.Set[...[].Has(...)
   7     /home/runner/go/pkg/mod/k8s.io/apimachinery@v0.33.3/pkg/util/sets/set.go:78
   8 github.com/envoyproxy/gateway/internal/provider/kubernetes.(*gatewayAPIReconciler).envoyObjectForGateway.func1({0x8bc2dc8, 0xc0054163f0})
   9     /home/runner/work/gateway/gateway/internal/provider/kubernetes/predicates.go:666 +0x7b
  10 github.com/envoyproxy/gateway/internal/provider/kubernetes.(*gatewayAPIReconciler).envoyObjectForGateway(0xc0018858e8?, {0x8ba6648?, 0xc0009c8b90?}, 0x11?)
  11     /home/runner/work/gateway/gateway/internal/provider/kubernetes/predicates.go:682 +0x5e
  12 github.com/envoyproxy/gateway/internal/provider/kubernetes.(*gatewayAPIReconciler).updateStatusForGateway(0xc000516600, {0x8ba6648, 0xc0009c8b90}, 0xc00541a380)
  13     /home/runner/work/gateway/gateway/internal/provider/kubernetes/status.go:579 +0x5a
  14 github.com/envoyproxy/gateway/internal/provider/kubernetes.(*gatewayAPIReconciler).subscribeAndUpdateStatus.func2.1({{{0xc0096a7e46, 0x7}, {0xc001944cc0, 0x11}}, 0x0, 0xc0052fa690}, 0xc00a582a1     0)
```

Relates to #7115 (comment)

Signed-off-by: Arko Dasgupta <arko@tetrate.io>

* lint

Signed-off-by: Arko Dasgupta <arko@tetrate.io>

---------

Signed-off-by: Arko Dasgupta <arko@tetrate.io>

Author: arkodg

internal/provider/kubernetes/controller.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/telepresenceio/watchable"
@@ -68,6 +69,7 @@ type gatewayAPIReconciler struct {
 	namespaceLabel       *metav1.LabelSelector
 	envoyGateway         *egv1a1.EnvoyGateway
 	mergeGateways        sets.Set[string]
+	mergeGatewaysMu      sync.RWMutex
 	resources            *message.ProviderResources
 	subscriptions        *subscriptions
 	extGVKs              []schema.GroupVersionKind
@@ -93,6 +95,24 @@ type gatewayAPIReconciler struct {
 	clusterTrustBundleExits bool
 }
 
+// isGatewayClassMerged reports whether the supplied GatewayClass has mergeGateways enabled.
+func (r *gatewayAPIReconciler) isGatewayClassMerged(name string) bool {
+	r.mergeGatewaysMu.RLock()
+	defer r.mergeGatewaysMu.RUnlock()
+	return r.mergeGateways.Has(name)
+}
+
+// setGatewayClassMerge toggles mergeGateways tracking for the supplied GatewayClass.
+func (r *gatewayAPIReconciler) setGatewayClassMerge(name string, merged bool) {
+	r.mergeGatewaysMu.Lock()
+	defer r.mergeGatewaysMu.Unlock()
+	if merged {
+		r.mergeGateways.Insert(name)
+		return
+	}
+	r.mergeGateways.Delete(name)
+}
+
 type subscriptions struct {
 	gatewayClassStatuses         <-chan watchable.Snapshot[types.NamespacedName, *gwapiv1.GatewayClassStatus]
 	gatewayStatuses              <-chan watchable.Snapshot[types.NamespacedName, *gwapiv1.GatewayStatus]
@@ -492,11 +512,7 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques
 		}
 
 		if gwcResource.EnvoyProxyForGatewayClass != nil && gwcResource.EnvoyProxyForGatewayClass.Spec.MergeGateways != nil {
-			if *gwcResource.EnvoyProxyForGatewayClass.Spec.MergeGateways {
-				r.mergeGateways.Insert(managedGC.Name)
-			} else {
-				r.mergeGateways.Delete(managedGC.Name)
-			}
+			r.setGatewayClassMerge(managedGC.Name, *gwcResource.EnvoyProxyForGatewayClass.Spec.MergeGateways)
 		}
 
 		if len(gwcResource.Gateways) == 0 {
@@ -1402,7 +1418,7 @@ func (r *gatewayAPIReconciler) processGateways(ctx context.Context, managedGC *g
 	}
 
 	mergedGateways := false
-	if r.mergeGateways.Has(managedGC.Name) {
+	if r.isGatewayClassMerged(managedGC.Name) {
 		mergedGateways = true
 		// processGatewayClassParamsRef has been called for this gatewayclass, its EnvoyProxy should exist in resourceTree
 		r.processServiceClusterForGatewayClass(resourceTree.EnvoyProxyForGatewayClass, managedGC, resourceMap)

internal/provider/kubernetes/controller_offline.go

@@ -28,7 +28,7 @@ import (
 // OfflineGatewayAPIReconciler can be used for non-kuberetes provider.
 // It can let other providers to have the same reconcile logic without rely on apiserver.
 type OfflineGatewayAPIReconciler struct {
-	gatewayAPIReconciler
+	*gatewayAPIReconciler
 
 	Client client.Client
 }
@@ -63,7 +63,7 @@ func NewOfflineGatewayAPIController(
 	}
 
 	cli := newOfflineGatewayAPIClient()
-	r := gatewayAPIReconciler{
+	r := &gatewayAPIReconciler{
 		client:            cli,
 		log:               cfg.Logger,
 		classController:   gwapiv1.GatewayController(cfg.EnvoyGateway.Gateway.ControllerName),

internal/provider/kubernetes/predicates.go

@@ -403,7 +403,7 @@ func (r *gatewayAPIReconciler) validateServiceForReconcile(obj client.Object) bo
 
 	// Merged gateways will have only this label, update status of all Gateways under found GatewayClass.
 	gcName, ok := labels[gatewayapi.OwningGatewayClassLabel]
-	if ok && r.mergeGateways.Has(gcName) {
+	if ok && r.isGatewayClassMerged(gcName) {
 		if err := r.updateStatusForGatewaysUnderGatewayClass(ctx, gcName); err != nil {
 			r.log.Info("no Gateways found under GatewayClass", "name", gcName)
 			return false
@@ -641,7 +641,7 @@ func (r *gatewayAPIReconciler) validateObjectForReconcile(obj client.Object) boo
 
 	// Merged gateways will have only this label, update status of all Gateways under found GatewayClass.
 	gcName, ok := labels[gatewayapi.OwningGatewayClassLabel]
-	if ok && r.mergeGateways.Has(gcName) {
+	if ok && r.isGatewayClassMerged(gcName) {
 		if err := r.updateStatusForGatewaysUnderGatewayClass(ctx, gcName); err != nil {
 			r.log.Info("no Gateways found under GatewayClass", "name", gcName)
 			return false
@@ -663,9 +663,11 @@ func envoyObjectNamespace(r *gatewayAPIReconciler, gateway *gwapiv1.Gateway) str
 // envoyObjectForGateway returns the Envoy Deployment or DaemonSet, returning nil if neither exists.
 func (r *gatewayAPIReconciler) envoyObjectForGateway(ctx context.Context, gateway *gwapiv1.Gateway) (client.Object, error) {
 	// Helper func to list and return the first object from results
+	merged := r.isGatewayClassMerged(string(gateway.Spec.GatewayClassName))
+
 	listResource := func(list client.ObjectList) (client.Object, error) {
 		if err := r.client.List(ctx, list, &client.ListOptions{
-			LabelSelector: labels.SelectorFromSet(gatewayapi.OwnerLabels(gateway, r.mergeGateways.Has(string(gateway.Spec.GatewayClassName)))),
+			LabelSelector: labels.SelectorFromSet(gatewayapi.OwnerLabels(gateway, merged)),
 			Namespace:     envoyObjectNamespace(r, gateway),
 		}); err != nil {
 			if !kerrors.IsNotFound(err) {
@@ -697,7 +699,8 @@ func (r *gatewayAPIReconciler) envoyObjectForGateway(ctx context.Context, gatewa
 // envoyServiceForGateway returns the Envoy service, returning nil if the service doesn't exist.
 func (r *gatewayAPIReconciler) envoyServiceForGateway(ctx context.Context, gateway *gwapiv1.Gateway) (*corev1.Service, error) {
 	var services corev1.ServiceList
-	labelSelector := labels.SelectorFromSet(labels.Set(gatewayapi.OwnerLabels(gateway, r.mergeGateways.Has(string(gateway.Spec.GatewayClassName)))))
+	merged := r.isGatewayClassMerged(string(gateway.Spec.GatewayClassName))
+	labelSelector := labels.SelectorFromSet(labels.Set(gatewayapi.OwnerLabels(gateway, merged)))
 	if err := r.client.List(ctx, &services, &client.ListOptions{
 		LabelSelector: labelSelector,
 		Namespace:     envoyObjectNamespace(r, gateway),
@@ -955,7 +958,7 @@ func (r *gatewayAPIReconciler) isProxyServiceCluster(labels map[string]string) b
 	}
 
 	gcName, ok := labels[gatewayapi.OwningGatewayClassLabel]
-	if ok && r.mergeGateways.Has(gcName) {
+	if ok && r.isGatewayClassMerged(gcName) {
 		return true
 	}