Add multiproject manager which will be responsible for starting and tracking controllers per ClusterSlice

pkg/multiproject/manager/manager.go

@@ -0,0 +1,121 @@
+package manager
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/types"
+	informers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	providerconfig "k8s.io/ingress-gce/pkg/apis/providerconfig/v1"
+	"k8s.io/ingress-gce/pkg/multiproject/finalizer"
+	"k8s.io/ingress-gce/pkg/multiproject/neg"
+	"k8s.io/ingress-gce/pkg/neg/syncers/labels"
+	providerconfigclient "k8s.io/ingress-gce/pkg/providerconfig/client/clientset/versioned"
+	svcnegclient "k8s.io/ingress-gce/pkg/svcneg/client/clientset/versioned"
+	"k8s.io/ingress-gce/pkg/utils/namer"
+	"k8s.io/klog/v2"
+)
+
+type ProviderConfigControllersManager struct {
+	controllers          map[string]*ControllerSet
+	mu                   sync.Mutex
+	logger               klog.Logger
+	providerConfigClient providerconfigclient.Interface
+	informersFactory     informers.SharedInformerFactory
+	kubeClient           kubernetes.Interface
+	svcNegClient         svcnegclient.Interface
+	eventRecorderClient  kubernetes.Interface
+	kubeSystemUID        types.UID
+	clusterNamer         *namer.Namer
+	l4Namer              *namer.L4Namer
+	lpConfig             labels.PodLabelPropagationConfig
+	defaultCloudConfig   string
+	globalStopCh         <-chan struct{}
+}
+
+type ControllerSet struct {
+	stopCh chan struct{}
+}
+
+func NewProviderConfigControllerManager(
+	kubeClient kubernetes.Interface,
+	informersFactory informers.SharedInformerFactory,
+	providerConfigClient providerconfigclient.Interface,
+	svcNegClient svcnegclient.Interface,
+	eventRecorderClient kubernetes.Interface,
+	kubeSystemUID types.UID,
+	clusterNamer *namer.Namer,
+	l4Namer *namer.L4Namer,
+	lpConfig labels.PodLabelPropagationConfig,
+	defaultCloudConfig string,
+	globalStopCh <-chan struct{},
+	logger klog.Logger,
+) *ProviderConfigControllersManager {
+	return &ProviderConfigControllersManager{
+		controllers:          make(map[string]*ControllerSet),
+		logger:               logger,
+		providerConfigClient: providerConfigClient,
+		informersFactory:     informersFactory,
+		kubeClient:           kubeClient,
+		svcNegClient:         svcNegClient,
+		eventRecorderClient:  eventRecorderClient,
+		kubeSystemUID:        kubeSystemUID,
+		clusterNamer:         clusterNamer,
+		l4Namer:              l4Namer,
+		lpConfig:             lpConfig,
+		defaultCloudConfig:   defaultCloudConfig,
+		globalStopCh:         globalStopCh,
+	}
+}
+
+func providerConfigKey(cs *providerconfig.ProviderConfig) string {
+	return cs.Namespace + "/" + cs.Name
+}
+
+func (cscm *ProviderConfigControllersManager) StartControllersForProviderConfig(cs *providerconfig.ProviderConfig) error {
+	cscm.mu.Lock()
+	defer cscm.mu.Unlock()
+
+	csKey := providerConfigKey(cs)
+	if _, exists := cscm.controllers[csKey]; exists {
+		cscm.logger.Info("Controllers for provider config already exist, skipping start", "providerConfigId", csKey)
+		return nil
+	}
+
+	err := finalizer.EnsureProviderConfigNEGCleanupFinalizer(cs, cscm.providerConfigClient, cscm.logger)
+	if err != nil {
+		return fmt.Errorf("failed to ensure NEG cleanup finalizer for project %s: %v", csKey, err)
+	}
+	negControllerStopCh, err := neg.StartNEGController(cscm.informersFactory, cscm.kubeClient, cscm.svcNegClient, cscm.eventRecorderClient, cscm.kubeSystemUID, cscm.clusterNamer, cscm.l4Namer, cscm.lpConfig, cscm.defaultCloudConfig, cscm.globalStopCh, cscm.logger, cs)
+	if err != nil {
+		return fmt.Errorf("failed to start NEG controller for project %s: %v", csKey, err)
+	}
+	cscm.controllers[csKey] = &ControllerSet{
+		stopCh: negControllerStopCh,
+	}
+
+	cscm.logger.Info("Started controllers for provider config", "providerConfigId", csKey)
+	return nil
+}
+
+func (cscm *ProviderConfigControllersManager) StopControllersForProviderConfig(cs *providerconfig.ProviderConfig) {
+	cscm.mu.Lock()
+	defer cscm.mu.Unlock()
+
+	csKey := providerConfigKey(cs)
+	_, exists := cscm.controllers[csKey]
+	if !exists {
+		cscm.logger.Info("Controllers for provider config do not exist", "providerConfigId", csKey)
+		return
+	}
+
+	close(cscm.controllers[csKey].stopCh)
+
+	delete(cscm.controllers, csKey)
+	err := finalizer.DeleteProviderConfigNEGCleanupFinalizer(cs, cscm.providerConfigClient, cscm.logger)
+	if err != nil {
+		cscm.logger.Error(err, "failed to delete NEG cleanup finalizer for project", "providerConfigId", csKey)
+	}
+	cscm.logger.Info("Stopped controllers for provider config", "providerConfigId", csKey)
+}

pkg/multiproject/neg/neg.go

@@ -0,0 +1,221 @@
+package neg
+
+import (
+	"fmt"
+
+	networkparams "github.com/GoogleCloudPlatform/gke-networking-api/apis/network/v1"
+	nodetopology "github.com/GoogleCloudPlatform/gke-networking-api/apis/nodetopology/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/cloud-provider-gcp/providers/gce"
+	providerconfig "k8s.io/ingress-gce/pkg/apis/providerconfig/v1"
+	svcneg "k8s.io/ingress-gce/pkg/apis/svcneg/v1beta1"
+	"k8s.io/ingress-gce/pkg/flags"
+	"k8s.io/ingress-gce/pkg/multiproject/filteredinformer"
+	multiprojectgce "k8s.io/ingress-gce/pkg/multiproject/gce"
+	"k8s.io/ingress-gce/pkg/neg"
+	"k8s.io/ingress-gce/pkg/neg/syncers/labels"
+	negtypes "k8s.io/ingress-gce/pkg/neg/types"
+	"k8s.io/ingress-gce/pkg/network"
+	svcnegclient "k8s.io/ingress-gce/pkg/svcneg/client/clientset/versioned"
+	"k8s.io/ingress-gce/pkg/utils"
+	"k8s.io/ingress-gce/pkg/utils/namer"
+	"k8s.io/ingress-gce/pkg/utils/zonegetter"
+	"k8s.io/klog/v2"
+)
+
+var svcNegResource = schema.GroupVersionResource{Group: svcneg.SchemeGroupVersion.Group, Version: svcneg.SchemeGroupVersion.Version, Resource: "servicenetworkendpointgroups"}
+var gkeNetworkParamsResource = schema.GroupVersionResource{Group: networkparams.GroupVersion.Group, Version: networkparams.GroupVersion.Version, Resource: "gkenetworkparamsets"}
+var nodeTopologyResource = schema.GroupVersionResource{Group: nodetopology.GroupVersion.Group, Version: nodetopology.GroupVersion.Version, Resource: "nodetopologies"}
+
+func StartNEGController(
+	informersFactory informers.SharedInformerFactory,
+	kubeClient kubernetes.Interface,
+	svcNegClient svcnegclient.Interface,
+	eventRecorderClient kubernetes.Interface,
+	kubeSystemUID types.UID,
+	clusterNamer *namer.Namer,
+	l4Namer *namer.L4Namer,
+	lpConfig labels.PodLabelPropagationConfig,
+	defaultCloudConfig string,
+	globalStopCh <-chan struct{},
+	logger klog.Logger,
+	providerConfig *providerconfig.ProviderConfig,
+) (chan struct{}, error) {
+	cloud, err := multiprojectgce.NewGCEForProviderConfig(defaultCloudConfig, providerConfig, logger)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GCE client for provider config %+v: %v", providerConfig, err)
+	}
+
+	providerConfigName := providerConfig.Name
+
+	// Using informer factory, create required namespaced informers for the NEG controller.
+	ingressInformer := filteredinformer.NewProviderConfigFilteredInformer(informersFactory.Networking().V1().Ingresses().Informer(), providerConfigName)
+	serviceInformer := filteredinformer.NewProviderConfigFilteredInformer(informersFactory.Core().V1().Services().Informer(), providerConfigName)
+	podInformer := filteredinformer.NewProviderConfigFilteredInformer(informersFactory.Core().V1().Pods().Informer(), providerConfigName)
+	nodeInformer := filteredinformer.NewProviderConfigFilteredInformer(informersFactory.Core().V1().Nodes().Informer(), providerConfigName)
+	endpointSliceInformer := filteredinformer.NewProviderConfigFilteredInformer(informersFactory.Discovery().V1().EndpointSlices().Informer(), providerConfigName)
+
+	svcNegInformer, err := informersFactory.ForResource(svcNegResource)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get svcNeg informer: %v", err)
+	}
+	providerConfigFilteredSvcNegInformer := filteredinformer.NewProviderConfigFilteredInformer(svcNegInformer.Informer(), providerConfigName)
+
+	networkInformer, err := informersFactory.ForResource(gkeNetworkParamsResource)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get network informer: %v", err)
+	}
+	providerConfigFilteredNetworkInformer := filteredinformer.NewProviderConfigFilteredInformer(networkInformer.Informer(), providerConfigName)
+
+	gkeNetworkParamsInformer, err := informersFactory.ForResource(gkeNetworkParamsResource)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get gkeNetworkParams informer: %v", err)
+	}
+	providerConfigFilteredGkeNetworkParamsInformer := filteredinformer.NewProviderConfigFilteredInformer(gkeNetworkParamsInformer.Informer(), providerConfigName)
+
+	nodeTopologyInformer, err := informersFactory.ForResource(nodeTopologyResource)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get nodeTopology informer: %v", err)
+	}
+	providerConfigFilteredNodeTopologyInformer := filteredinformer.NewProviderConfigFilteredInformer(nodeTopologyInformer.Informer(), providerConfigName)
+
+	// Create a function to check if all the informers have synced.
+	hasSynced := func() bool {
+		return ingressInformer.HasSynced() &&
+			serviceInformer.HasSynced() &&
+			podInformer.HasSynced() &&
+			nodeInformer.HasSynced() &&
+			endpointSliceInformer.HasSynced() &&
+			providerConfigFilteredSvcNegInformer.HasSynced() &&
+			providerConfigFilteredNetworkInformer.HasSynced() &&
+			providerConfigFilteredGkeNetworkParamsInformer.HasSynced() &&
+			providerConfigFilteredNodeTopologyInformer.HasSynced()
+	}
+
+	zoneGetter := zonegetter.NewZoneGetter(nodeInformer, providerConfigFilteredNodeTopologyInformer, cloud.SubnetworkURL())
+
+	// Create a channel to stop the controller for this specific provider config.
+	providerConfigStopCh := make(chan struct{})
+	defer close(providerConfigStopCh)
+
+	// joinedStopCh is a channel that will be closed when the global stop channel or the provider config stop channel is closed.
+	joinedStopCh := make(chan struct{})
+	go func() {
+		defer close(joinedStopCh)
+		select {
+		case <-globalStopCh:
+		case <-providerConfigStopCh:
+		}
+	}()
+
+	negController := createNEGController(
+		kubeClient,
+		svcNegClient,
+		eventRecorderClient,
+		kubeSystemUID,
+		ingressInformer,
+		serviceInformer,
+		podInformer,
+		nodeInformer,
+		endpointSliceInformer,
+		providerConfigFilteredSvcNegInformer,
+		providerConfigFilteredNetworkInformer,
+		providerConfigFilteredGkeNetworkParamsInformer,
+		hasSynced,
+		cloud,
+		zoneGetter,
+		clusterNamer,
+		l4Namer,
+		lpConfig,
+		joinedStopCh,
+		logger,
+	)
+
+	go negController.Run()
+
+	return providerConfigStopCh, nil
+}
+
+func createNEGController(
+	kubeClient kubernetes.Interface,
+	svcNegClient svcnegclient.Interface,
+	eventRecorderClient kubernetes.Interface,
+	kubeSystemUID types.UID,
+	ingressInformer cache.SharedIndexInformer,
+	serviceInformer cache.SharedIndexInformer,
+	podInformer cache.SharedIndexInformer,
+	nodeInformer cache.SharedIndexInformer,
+	endpointSliceInformer cache.SharedIndexInformer,
+	svcNegInformer cache.SharedIndexInformer,
+	networkInformer cache.SharedIndexInformer,
+	gkeNetworkParamsInformer cache.SharedIndexInformer,
+	hasSynced func() bool,
+	cloud *gce.Cloud,
+	zoneGetter *zonegetter.ZoneGetter,
+	clusterNamer *namer.Namer,
+	l4Namer *namer.L4Namer,
+	lpConfig labels.PodLabelPropagationConfig,
+	stopCh <-chan struct{},
+	logger klog.Logger,
+) *neg.Controller {
+
+	// The following adapter will use Network Selflink as Network Url instead of the NetworkUrl itself.
+	// Network Selflink is always composed by the network name even if the cluster was initialized with Network Id.
+	// All the components created from it will be consistent and always use the Url with network name and not the url with netowork Id
+	adapter, err := network.NewAdapterNetworkSelfLink(cloud)
+	if err != nil {
+		logger.Error(err, "Failed to create network adapter with SelfLink")
+		// if it was not possible to retrieve network information use standard context as cloud network provider
+		adapter = cloud
+	}
+
+	noDefaultBackendServicePort := utils.ServicePort{} // we don't need default backend service port for standalone NEGs.
+
+	var noNodeTopologyInformer cache.SharedIndexInformer = nil
+
+	asmServiceNEGSkipNamespaces := []string{}
+	enableASM := false
+
+	negController := neg.NewController(
+		kubeClient,
+		svcNegClient,
+		eventRecorderClient,
+		kubeSystemUID,
+		ingressInformer,
+		serviceInformer,
+		podInformer,
+		nodeInformer,
+		endpointSliceInformer,
+		svcNegInformer,
+		networkInformer,
+		gkeNetworkParamsInformer,
+		noNodeTopologyInformer,
+		hasSynced,
+		l4Namer,
+		noDefaultBackendServicePort,
+		negtypes.NewAdapterWithRateLimitSpecs(cloud, flags.F.GCERateLimit.Values(), adapter),
+		zoneGetter,
+		clusterNamer,
+		flags.F.ResyncPeriod,
+		flags.F.NegGCPeriod,
+		flags.F.NumNegGCWorkers,
+		flags.F.EnableReadinessReflector,
+		flags.F.EnableL4NEG,
+		flags.F.EnableNonGCPMode,
+		flags.F.EnableDualStackNEG,
+		enableASM,
+		asmServiceNEGSkipNamespaces,
+		lpConfig,
+		flags.F.EnableMultiNetworking,
+		flags.F.EnableIngressRegionalExternal,
+		flags.F.EnableL4NetLBNEG,
+		stopCh,
+		logger,
+	)
+
+	return negController
+}