Fixed metrics collector aggregation issue for NEG metrics in MT

Author: priyapande

OWNERS

@@ -1,20 +1,18 @@
 approvers:
-- aojea
-- thockin
-- bowei
-- freehan
-- mrhohn
-- prameshj
-- swetharepakula
-- panslava
-- cezarygerard
-- kl52752
-- spencerhance
-- code-elinka
-- mmamczur
-- alexkats
-- gauravkghildiyal
-- tortillazhawaii
+  - aojea
+  - thockin
+  - bowei
+  - mrhohn
+  - swetharepakula
+  - panslava
+  - cezarygerard
+  - kl52752
+  - spencerhance
+  - code-elinka
+  - mmamczur
+  - gauravkghildiyal
+  - tortillazhawaii
+  - felipeyepez
 reviewers:
-- aojea
-- sawsa307
+  - aojea
+  - sawsa307

cmd/glbc/app/init.go

@@ -35,6 +35,11 @@ import (
 // DefaultBackendServicePort returns the ServicePort which will be
 // used as the default backend for load balancers.
 func DefaultBackendServicePort(kubeClient kubernetes.Interface, logger klog.Logger) utils.ServicePort {
+	if !flags.F.RunIngressController && !flags.F.EnableNEGsForIngress {
+		noDefaultBackendServicePort := utils.ServicePort{}
+		return noDefaultBackendServicePort
+	}
+
 	if flags.F.DefaultSvc == "" {
 		klog.Fatalf("Please specify --default-backend-service")
 	}

cmd/glbc/main.go

@@ -59,18 +59,19 @@ import (
 	"k8s.io/ingress-gce/pkg/utils"
 	"k8s.io/klog/v2"
 
-	ingctx "k8s.io/ingress-gce/pkg/context"
-	"k8s.io/ingress-gce/pkg/controller"
-	"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/cmd/glbc/app"
 	"k8s.io/ingress-gce/pkg/backendconfig"
+	ingctx "k8s.io/ingress-gce/pkg/context"
+	"k8s.io/ingress-gce/pkg/controller"
 	"k8s.io/ingress-gce/pkg/crd"
 	"k8s.io/ingress-gce/pkg/firewalls"
 	"k8s.io/ingress-gce/pkg/flags"
 	_ "k8s.io/ingress-gce/pkg/klog"
+	"k8s.io/ingress-gce/pkg/neg"
+	"k8s.io/ingress-gce/pkg/neg/metrics"
+	syncMetrics "k8s.io/ingress-gce/pkg/neg/metrics/metricscollector"
+	"k8s.io/ingress-gce/pkg/neg/syncers/labels"
+	negtypes "k8s.io/ingress-gce/pkg/neg/types"
 	"k8s.io/ingress-gce/pkg/utils/zonegetter"
 	"k8s.io/ingress-gce/pkg/version"
 )
@@ -265,9 +266,12 @@ func main() {
 			if nodeTopologyClient != nil {
 				nodeTopologyFactory = informernodetopology.NewSharedInformerFactory(nodeTopologyClient, flags.F.ResyncPeriod)
 			}
+			ctx := context.Background()
+			syncerMetrics := syncMetrics.NewNegMetricsCollector(flags.F.NegMetricsExportInterval, rootLogger, "")
+
 			if flags.F.LeaderElection.LeaderElect {
 				err := multiprojectstart.StartWithLeaderElection(
-					context.Background(),
+					ctx,
 					leaderElectKubeClient,
 					hostname,
 					rootLogger,
@@ -283,10 +287,12 @@ func main() {
 					gceCreator,
 					namer,
 					stopCh,
+					syncerMetrics,
 				)
 				if err != nil {
 					rootLogger.Error(err, "Failed to start multi-project syncer with leader election")
 				}
+				rOption.closeStopCh()
 			} else {
 				multiprojectstart.Start(
 					rootLogger,
@@ -302,6 +308,7 @@ func main() {
 					gceCreator,
 					namer,
 					stopCh,
+					syncerMetrics,
 				)
 			}
 		}, rOption.wg)
@@ -364,7 +371,7 @@ func main() {
 		id: fmt.Sprintf("%v_%x", hostname, rand.Intn(1e6)),
 	}
 
-	enableOtherControllers := flags.F.RunIngressController || flags.F.RunL4Controller || flags.F.RunL4NetLBController || flags.F.EnableIGController || flags.F.EnablePSC
+	enableL4Controllers := flags.F.RunL4Controller || flags.F.RunL4NetLBController || flags.F.EnableIGController || flags.F.EnablePSC
 	runNEG := func() {
 		logger := rootLogger.WithName("NEGController")
 		logger.Info("Start running the enabled controllers",
@@ -376,15 +383,22 @@ func main() {
 		}
 	}
 	runIngress := func() {
-		logger := rootLogger.WithName("Other controllers")
+		logger := rootLogger.WithName("Ingress controller")
 		logger.Info("Start running the enabled controllers",
 			"Ingress controller", flags.F.RunIngressController,
+		)
+		runIngressControllers(ctx, systemHealth, rOption, leOption, logger)
+	}
+
+	runL4 := func() {
+		logger := rootLogger.WithName("L4 controllers")
+		logger.Info("Start running the enabled controllers",
 			"L4 controller", flags.F.RunL4Controller,
 			"L4 NetLB controller", flags.F.RunL4NetLBController,
 			"InstanceGroup controller", flags.F.EnableIGController,
 			"PSC controller", flags.F.EnablePSC,
 		)
-		runControllers(ctx, systemHealth, rOption, leOption, logger)
+		runL4Controllers(ctx, systemHealth, rOption, leOption, logger)
 	}
 
 	if flags.F.LeaderElection.LeaderElect {
@@ -401,28 +415,45 @@ func main() {
 			logger.Info("NEG Controller exited.")
 		}
 		runIngress = func() {
-			logger := rootLogger.WithName("Other controllers")
+			logger := rootLogger.WithName("Ingress controller")
 			logger.Info("Start running Ingress leader election",
 				"Ingress controller", flags.F.RunIngressController,
+			)
+			ingressRunner, err := makeIngressRunnerWithLeaderElection(ctx, systemHealth, rOption, leOption, logger)
+			if err != nil {
+				klog.Fatalf("makeLeaderElectionConfig()=%v, want nil", err)
+			}
+			leaderelection.RunOrDie(context.Background(), *ingressRunner)
+		}
+		if !flags.F.GateL4ByLock {
+			klog.Fatalf("--gate-l4-by-lock must be true when --leader-elect=true")
+		}
+		runL4 = func() {
+			logger := rootLogger.WithName("L4 controller")
+			logger.Info("Start running L4 leader election",
 				"L4 controller", flags.F.RunL4Controller,
 				"L4 NetLB controller", flags.F.RunL4NetLBController,
 				"InstanceGroup controller", flags.F.EnableIGController,
 				"PSC controller", flags.F.EnablePSC,
 			)
-			ingressRunner, err := makeIngressRunnerWithLeaderElection(ctx, systemHealth, rOption, leOption, logger)
+			l4Runner, err := makeL4RunnerWithLeaderElection(ctx, systemHealth, rOption, leOption, logger)
 			if err != nil {
 				klog.Fatalf("makeLeaderElectionConfig()=%v, want nil", err)
 			}
-			leaderelection.RunOrDie(context.Background(), *ingressRunner)
+			leaderelection.RunOrDie(context.Background(), *l4Runner)
 		}
+
 	}
 
 	if flags.F.EnableNEGController {
 		go runNEG()
 	}
-	if enableOtherControllers {
+	if flags.F.RunIngressController {
 		go runIngress()
 	}
+	if enableL4Controllers {
+		go runL4()
+	}
 
 	<-rOption.stopCh
 	waitWithTimeout(rOption.wg, rootLogger)
@@ -479,14 +510,35 @@ func makeIngressRunnerWithLeaderElection(
 		leOption,
 		flags.F.LeaderElection.LockObjectName,
 		func(context.Context) {
-			runControllers(ctx, systemHealth, runOption, leOption, logger)
+			runIngressControllers(ctx, systemHealth, runOption, leOption, logger)
 		},
 		func() {
 			logger.Info("lost master")
 		},
 	)
 }
 
+func makeL4RunnerWithLeaderElection(
+	ctx *ingctx.ControllerContext,
+	systemHealth *systemhealth.SystemHealth,
+	runOption runOption,
+	leOption leaderElectionOption,
+	logger klog.Logger,
+) (*leaderelection.LeaderElectionConfig, error) {
+	lockLogger := logger.WithValues("lock", l4LockName)
+	return makeRunnerWithLeaderElection(
+		leOption,
+		l4LockName,
+		func(context.Context) {
+			lockLogger.V(0).Info("Acquired L4 Leader election lock")
+			runL4Controllers(ctx, systemHealth, runOption, leOption, logger)
+		},
+		func() {
+			lockLogger.Info("lost master")
+		},
+	)
+}
+
 // makeRunnerWithLeaderElection creates a LeaderElectionConfig with the provided options and callbacks.
 // It will create a new resource lock associated with the configuration.
 func makeRunnerWithLeaderElection(
@@ -520,7 +572,7 @@ func makeRunnerWithLeaderElection(
 	}, nil
 }
 
-func runControllers(ctx *ingctx.ControllerContext, systemHealth *systemhealth.SystemHealth, option runOption, leOption leaderElectionOption, logger klog.Logger) {
+func runIngressControllers(ctx *ingctx.ControllerContext, systemHealth *systemhealth.SystemHealth, option runOption, leOption leaderElectionOption, logger klog.Logger) {
 	if flags.F.RunIngressController {
 		lbc := controller.NewLoadBalancerController(ctx, option.stopCh, logger)
 		systemHealth.AddHealthCheck("ingress", lbc.SystemHealth)
@@ -538,72 +590,55 @@ func runControllers(ctx *ingctx.ControllerContext, systemHealth *systemhealth.Sy
 		logger.V(0).Info("firewall controller started")
 	}
 
-	runL4Controllers(ctx, systemHealth, option, leOption, logger)
-
 	ctx.Start(option.stopCh)
 }
 
 func runL4Controllers(ctx *ingctx.ControllerContext, systemHealth *systemhealth.SystemHealth, option runOption, leOption leaderElectionOption, logger klog.Logger) {
 	if !flags.F.RunL4Controller && !flags.F.EnablePSC && !flags.F.EnableIGController && !flags.F.RunL4NetLBController {
 		return
 	}
-	run := func() {
-		if flags.F.RunL4Controller {
-			l4Controller := l4lb.NewILBController(ctx, option.stopCh, logger)
-			systemHealth.AddHealthCheck(l4lb.L4ILBControllerName, l4Controller.SystemHealth)
-			runWithWg(l4Controller.Run, option.wg)
-			logger.V(0).Info("L4 controller started")
-		}
-
-		if flags.F.EnablePSC {
-			pscController := psc.NewController(ctx, option.stopCh, logger)
-			runWithWg(pscController.Run, option.wg)
-			logger.V(0).Info("PSC Controller started")
-		}
 
-		if flags.F.EnableIGController {
-			igControllerParams := &instancegroups.ControllerConfig{
-				NodeInformer:             ctx.NodeInformer,
-				ZoneGetter:               ctx.ZoneGetter,
-				IGManager:                ctx.InstancePool,
-				HasSynced:                ctx.HasSynced,
-				EnableMultiSubnetCluster: flags.F.EnableIGMultiSubnetCluster,
-				ReadOnlyMode:             flags.F.ReadOnlyMode,
-				StopCh:                   option.stopCh,
-			}
-			igController := instancegroups.NewController(igControllerParams, logger)
-			runWithWg(igController.Run, option.wg)
-		}
+	if flags.F.GateL4ByLock {
+		go collectLockAvailabilityMetrics(l4LockName, flags.F.GKEClusterType, option.stopCh, logger)
+	}
 
-		// The L4NetLbController will be run when RbsMode flag is Set
-		if flags.F.RunL4NetLBController {
-			l4netlbController := l4lb.NewL4NetLBController(ctx, option.stopCh, logger)
-			systemHealth.AddHealthCheck(l4lb.L4NetLBControllerName, l4netlbController.SystemHealth)
+	if flags.F.RunL4Controller {
+		l4Controller := l4lb.NewILBController(ctx, option.stopCh, logger)
+		systemHealth.AddHealthCheck(l4lb.L4ILBControllerName, l4Controller.SystemHealth)
+		runWithWg(l4Controller.Run, option.wg)
+		logger.V(0).Info("L4 controller started")
+	}
 
-			runWithWg(l4netlbController.Run, option.wg)
-			logger.V(0).Info("L4NetLB controller started")
+	if flags.F.EnablePSC {
+		pscController := psc.NewController(ctx, option.stopCh, logger)
+		runWithWg(pscController.Run, option.wg)
+		logger.V(0).Info("PSC Controller started")
+	}
+
+	if flags.F.EnableIGController {
+		igControllerParams := &instancegroups.ControllerConfig{
+			NodeInformer:             ctx.NodeInformer,
+			ZoneGetter:               ctx.ZoneGetter,
+			IGManager:                ctx.InstancePool,
+			HasSynced:                ctx.HasSynced,
+			EnableMultiSubnetCluster: flags.F.EnableIGMultiSubnetCluster,
+			ReadOnlyMode:             flags.F.ReadOnlyMode,
+			StopCh:                   option.stopCh,
 		}
+		igController := instancegroups.NewController(igControllerParams, logger)
+		runWithWg(igController.Run, option.wg)
 	}
-	if !flags.F.LeaderElection.LeaderElect || !flags.F.GateL4ByLock {
-		run()
-		return
-	}
-	lockLogger := logger.WithValues("lock", l4LockName)
-	runner, err := makeRunnerWithLeaderElection(leOption, l4LockName, func(ctx context.Context) {
-		lockLogger.V(0).Info("Acquired L4 Leader election lock")
-		go collectLockAvailabilityMetrics(l4LockName, flags.F.GKEClusterType, option.stopCh, lockLogger)
-		run()
-	}, func() {
-		lockLogger.V(0).Info("Stop running L4 Leader election")
-	})
-	if err != nil {
-		klog.Fatalf("L4 makeLeaderElectionConfig()=%v, want nil", err)
+
+	// The L4NetLbController will be run when RbsMode flag is Set
+	if flags.F.RunL4NetLBController {
+		l4netlbController := l4lb.NewL4NetLBController(ctx, option.stopCh, logger)
+		systemHealth.AddHealthCheck(l4lb.L4NetLBControllerName, l4netlbController.SystemHealth)
+
+		runWithWg(l4netlbController.Run, option.wg)
+		logger.V(0).Info("L4NetLB controller started")
 	}
-	// run in a separate goroutine to not block further operation if lock can't be acquired.
-	go func() {
-		lockLogger.V(0).Info("Attempt to acquire L4 Leader election lock")
-		leaderelection.RunOrDie(context.Background(), *runner)
-	}()
+
+	ctx.Start(option.stopCh)
 }
 
 func runNEGController(ctx *ingctx.ControllerContext, systemHealth *systemhealth.SystemHealth, option runOption, logger klog.Logger) error {
@@ -656,6 +691,13 @@ func createNEGController(ctx *ingctx.ControllerContext, systemHealth *systemheal
 		adapter = ctx.Cloud
 	}
 
+	// register NEG prometheus metrics
+	metrics.RegisterMetrics()
+	syncMetrics.RegisterMetrics()
+
+	negMetrics := metrics.NewNegMetrics("")
+	syncerMetrics := syncMetrics.NewNegMetricsCollector(flags.F.NegMetricsExportInterval, logger, negMetrics.ProviderConfigID)
+
 	// TODO: Refactor NEG to use cloud mocks so ctx.Cloud can be referenced within NewController.
 	negController, err := neg.NewController(
 		ctx.KubeClient,
@@ -674,7 +716,7 @@ func createNEGController(ctx *ingctx.ControllerContext, systemHealth *systemheal
 		ctx.HasSynced,
 		ctx.L4Namer,
 		ctx.DefaultBackendSvcPort,
-		negtypes.NewAdapterWithRateLimitSpecs(ctx.Cloud, flags.F.GCERateLimit.Values(), adapter),
+		negtypes.NewAdapterWithRateLimitSpecs(ctx.Cloud, flags.F.GCERateLimit.Values(), adapter, negMetrics),
 		zoneGetter,
 		ctx.ClusterNamer,
 		flags.F.ResyncPeriod,
@@ -689,8 +731,11 @@ func createNEGController(ctx *ingctx.ControllerContext, systemHealth *systemheal
 		ctx.EnableIngressRegionalExternal,
 		flags.F.EnableL4NetLBNEG,
 		flags.F.ReadOnlyMode,
+		flags.F.EnableNEGsForIngress,
 		stopCh,
 		logger,
+		negMetrics,
+		syncerMetrics,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create NEG controller: %w", err)