Prevent LoadBalancer updates on follower.

Start watching Service updates only after becoming leader, when starting
loadBalancerStatusWriter. This avoids piling up unprocessed Service updates on
follower instance, consuming memory and eventually causing an out-of-memory
condition and killing Contour process.

Signed-off-by: Tero Saarni <[email protected]>

Author: tsaarni

changelogs/unreleased/6872-tsaarni-small.md

@@ -0,0 +1 @@
+Fixed a memory leak in Contour follower instance due to unprocessed LoadBalancer status updates.

cmd/contour/ingressstatus.go

@@ -22,6 +22,7 @@ import (
 	core_v1 "k8s.io/api/core/v1"
 	networking_v1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/types"
+	client_go_cache "k8s.io/client-go/tools/cache"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	gatewayapi_v1 "sigs.k8s.io/gateway-api/apis/v1"
@@ -55,13 +56,41 @@ type loadBalancerStatusWriter struct {
 	statusUpdater     k8s.StatusUpdater
 	ingressClassNames []string
 	gatewayRef        *types.NamespacedName
+	statusAddress     string
+	serviceName       string
+	serviceNamespace  string
 }
 
 func (isw *loadBalancerStatusWriter) NeedLeaderElection() bool {
 	return true
 }
 
 func (isw *loadBalancerStatusWriter) Start(ctx context.Context) error {
+	// Register an informer to watch envoy's service if we haven't been given static details.
+	if lbAddress := isw.statusAddress; len(lbAddress) > 0 {
+		isw.log.WithField("loadbalancer-address", lbAddress).Info("Using supplied information for Ingress status")
+		isw.lbStatus <- parseStatusFlag(lbAddress)
+	} else {
+		// Register Service informer to watch for status updates.
+		var serviceHandler client_go_cache.ResourceEventHandler = &k8s.ServiceStatusLoadBalancerWatcher{
+			ServiceName: isw.serviceName,
+			LBStatus:    isw.lbStatus,
+			Log:         isw.log.WithField("context", "serviceStatusLoadBalancerWatcher"),
+		}
+		var handler client_go_cache.ResourceEventHandler = serviceHandler
+		if isw.serviceNamespace != "" {
+			serviceHandler = k8s.NewNamespaceFilter([]string{isw.serviceNamespace}, handler)
+		}
+		inf, err := isw.cache.GetInformer(ctx, &core_v1.Service{})
+		if err != nil {
+			isw.log.WithError(err).Fatal("failed to get Service informer")
+		}
+		_, err = inf.AddEventHandler(serviceHandler)
+		if err != nil {
+			isw.log.WithError(err).Fatal("failed to add Service event handler")
+		}
+	}
+
 	u := &k8s.StatusAddressUpdater{
 		Logger: func() logrus.FieldLogger {
 			// Configure the StatusAddressUpdater logger.

cmd/contour/serve.go

@@ -700,36 +700,14 @@ func (s *Server) doServe() error {
 		ingressClassNames: ingressClassNames,
 		gatewayRef:        gatewayRef,
 		statusUpdater:     sh.Writer(),
+		statusAddress:     contourConfiguration.Ingress.StatusAddress,
+		serviceName:       contourConfiguration.Envoy.Service.Name,
+		serviceNamespace:  contourConfiguration.Envoy.Service.Namespace,
 	}
 	if err := s.mgr.Add(lbsw); err != nil {
 		return err
 	}
 
-	// Register an informer to watch envoy's service if we haven't been given static details.
-	if lbAddress := contourConfiguration.Ingress.StatusAddress; len(lbAddress) > 0 {
-		s.log.WithField("loadbalancer-address", lbAddress).Info("Using supplied information for Ingress status")
-		lbsw.lbStatus <- parseStatusFlag(lbAddress)
-	} else {
-		serviceHandler := &k8s.ServiceStatusLoadBalancerWatcher{
-			ServiceName: contourConfiguration.Envoy.Service.Name,
-			LBStatus:    lbsw.lbStatus,
-			Log:         s.log.WithField("context", "serviceStatusLoadBalancerWatcher"),
-		}
-
-		var handler cache.ResourceEventHandler = serviceHandler
-		if contourConfiguration.Envoy.Service.Namespace != "" {
-			handler = k8s.NewNamespaceFilter([]string{contourConfiguration.Envoy.Service.Namespace}, handler)
-		}
-
-		if err := s.informOnResource(&core_v1.Service{}, handler); err != nil {
-			s.log.WithError(err).WithField("resource", "services").Fatal("failed to create informer")
-		}
-
-		s.log.WithField("envoy-service-name", contourConfiguration.Envoy.Service.Name).
-			WithField("envoy-service-namespace", contourConfiguration.Envoy.Service.Namespace).
-			Info("Watching Service for Ingress status")
-	}
-
 	xdsServer := &xdsServer{
 		log:             s.log,
 		registry:        s.registry,

internal/k8s/statusaddress_test.go

@@ -41,6 +41,7 @@ func TestServiceStatusLoadBalancerWatcherOnAdd(t *testing.T) {
 		LBStatus:    lbstatus,
 		Log:         fixture.NewTestLogger(t),
 	}
+	sw.OnElectedLeader()
 
 	recv := func() (core_v1.LoadBalancerStatus, bool) {
 		select {
@@ -89,6 +90,7 @@ func TestServiceStatusLoadBalancerWatcherOnUpdate(t *testing.T) {
 		LBStatus:    lbstatus,
 		Log:         fixture.NewTestLogger(t),
 	}
+	sw.OnElectedLeader()
 
 	recv := func() (core_v1.LoadBalancerStatus, bool) {
 		select {
@@ -139,6 +141,7 @@ func TestServiceStatusLoadBalancerWatcherOnDelete(t *testing.T) {
 		LBStatus:    lbstatus,
 		Log:         fixture.NewTestLogger(t),
 	}
+	sw.OnElectedLeader()
 
 	recv := func() (core_v1.LoadBalancerStatus, bool) {
 		select {
@@ -179,6 +182,31 @@ func TestServiceStatusLoadBalancerWatcherOnDelete(t *testing.T) {
 	assert.Equal(t, want, got)
 }
 
+func TestServiceStatusLoadBalancerWatcherNoNotificationsOnFollower(t *testing.T) {
+	lbstatus := make(chan core_v1.LoadBalancerStatus, 1)
+
+	sw := ServiceStatusLoadBalancerWatcher{
+		ServiceName: "envoy",
+		LBStatus:    lbstatus,
+		Log:         fixture.NewTestLogger(t),
+	}
+
+	recv := func() bool {
+		select {
+		case <-sw.LBStatus:
+			return true
+		default:
+			return false
+		}
+	}
+
+	// assert that when not elected leader, no notifications are sent.
+	var svc core_v1.Service
+	svc.Name = "envoy"
+	sw.OnAdd(&svc, false)
+	assert.False(t, recv())
+}
+
 func TestStatusAddressUpdater(t *testing.T) {
 	const objName = "someobjfoo"