CP/DP split: Add leader election

internal/framework/runnables/runnables.go

@@ -37,9 +37,7 @@ func (r *LeaderOrNonLeader) NeedLeaderElection() bool {
 // CallFunctionsAfterBecameLeader is a Runnable that will call the given functions when the current instance becomes
 // the leader.
 type CallFunctionsAfterBecameLeader struct {
-	statusUpdaterEnable     func(context.Context)
-	healthCheckEnableLeader func()
-	eventHandlerEnable      func(context.Context)
+	enableFunctions []func(context.Context)
 }
 
 var (
@@ -49,21 +47,17 @@ var (
 
 // NewCallFunctionsAfterBecameLeader creates a new CallFunctionsAfterBecameLeader Runnable.
 func NewCallFunctionsAfterBecameLeader(
-	statusUpdaterEnable func(context.Context),
-	healthCheckEnableLeader func(),
-	eventHandlerEnable func(context.Context),
+	enableFunctions []func(context.Context),
 ) *CallFunctionsAfterBecameLeader {
 	return &CallFunctionsAfterBecameLeader{
-		statusUpdaterEnable:     statusUpdaterEnable,
-		healthCheckEnableLeader: healthCheckEnableLeader,
-		eventHandlerEnable:      eventHandlerEnable,
+		enableFunctions: enableFunctions,
 	}
 }
 
 func (j *CallFunctionsAfterBecameLeader) Start(ctx context.Context) error {
-	j.statusUpdaterEnable(ctx)
-	j.healthCheckEnableLeader()
-	j.eventHandlerEnable(ctx)
+	for _, function := range j.enableFunctions {
+		function(ctx)
+	}
 	return nil
 }
 

internal/framework/runnables/runnables_test.go

@@ -29,11 +29,11 @@ func TestCallFunctionsAfterBecameLeader(t *testing.T) {
 	healthCheckEnableLeader := false
 	eventHandlerEnabled := false
 
-	callFunctionsAfterBecameLeader := NewCallFunctionsAfterBecameLeader(
+	callFunctionsAfterBecameLeader := NewCallFunctionsAfterBecameLeader([]func(ctx context.Context){
 		func(_ context.Context) { statusUpdaterEnabled = true },
-		func() { healthCheckEnableLeader = true },
+		func(_ context.Context) { healthCheckEnableLeader = true },
 		func(_ context.Context) { eventHandlerEnabled = true },
-	)
+	})
 
 	g := NewWithT(t)
 	g.Expect(callFunctionsAfterBecameLeader.NeedLeaderElection()).To(BeTrue())

internal/mode/static/handler_test.go

@@ -487,6 +487,7 @@ var _ = Describe("eventHandler", func() {
 	})
 
 	It("should update nginx conf only when leader", func() {
+		ctx := context.Background()
 		handler.cfg.graphBuiltHealthChecker.leader = false
 
 		e := &events.UpsertEvent{Resource: &gatewayv1.HTTPRoute{}}
@@ -505,8 +506,8 @@ var _ = Describe("eventHandler", func() {
 		Expect(readyChannel).ShouldNot(BeClosed())
 
 		// Once the pod becomes leader, these two functions will be called through the runnables we set in the manager
-		handler.cfg.graphBuiltHealthChecker.setAsLeader()
-		handler.eventHandlerEnable(context.Background())
+		handler.cfg.graphBuiltHealthChecker.setAsLeader(ctx)
+		handler.eventHandlerEnable(ctx)
 
 		// nginx conf has been set
 		dcfg := dataplane.GetDefaultConfiguration(&graph.Graph{}, 1)

internal/mode/static/health.go

@@ -1,6 +1,7 @@
 package static
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net"
@@ -100,7 +101,7 @@ func (h *graphBuiltHealthChecker) getReadyCh() <-chan struct{} {
 }
 
 // setAsLeader marks the health check as leader.
-func (h *graphBuiltHealthChecker) setAsLeader() {
+func (h *graphBuiltHealthChecker) setAsLeader(_ context.Context) {
 	h.lock.Lock()
 	defer h.lock.Unlock()
 

internal/mode/static/health_test.go

@@ -1,6 +1,7 @@
 package static
 
 import (
+	"context"
 	"errors"
 	"net"
 	"net/http"
@@ -17,10 +18,10 @@ func TestReadyCheck(t *testing.T) {
 	g := NewWithT(t)
 	healthChecker := newGraphBuiltHealthChecker()
 
-	g.Expect(healthChecker.readyCheck(nil)).To(MatchError(errors.New("this NGF Pod is not currently leader")))
+	g.Expect(healthChecker.readyCheck(nil)).To(MatchError(errors.New("this Pod is not currently leader")))
 
 	healthChecker.graphBuilt = true
-	g.Expect(healthChecker.readyCheck(nil)).To(MatchError(errors.New("this NGF Pod is not currently leader")))
+	g.Expect(healthChecker.readyCheck(nil)).To(MatchError(errors.New("this Pod is not currently leader")))
 
 	healthChecker.graphBuilt = false
 	healthChecker.leader = true
@@ -39,7 +40,7 @@ func TestSetAsLeader(t *testing.T) {
 	g.Expect(healthChecker.leader).To(BeFalse())
 	g.Expect(healthChecker.readyCh).ShouldNot(BeClosed())
 
-	healthChecker.setAsLeader()
+	healthChecker.setAsLeader(context.Background())
 
 	g.Expect(healthChecker.leader).To(BeTrue())
 	g.Expect(healthChecker.readyCh).To(BeClosed())

internal/mode/static/manager.go

@@ -247,12 +247,12 @@
 		return fmt.Errorf("cannot register event loop: %w", err)
 	}
 
-	if err = mgr.Add(runnables.NewCallFunctionsAfterBecameLeader(
+	if err = mgr.Add(runnables.NewCallFunctionsAfterBecameLeader([]func(context.Context){
 		groupStatusUpdater.Enable,
 		healthChecker.setAsLeader,
 		eventHandler.eventHandlerEnable,
-	)); err != nil {
+	})); err != nil {
 		return fmt.Errorf("cannot register functions that get called after Pod becomes leader: %w", err)
 	}
 
 	if cfg.ProductTelemetryConfig.Enabled {
@@ -280,7 +280,7 @@
 	}
 
 	cfg.Logger.Info("Starting manager")
 	cfg.Logger.Info("NGINX Gateway Fabric Pod will be marked as unready until it has the leader lease")
 	go func() {
 		<-ctx.Done()
 		cfg.Logger.Info("Shutting down")
@@ -342,13 +342,13 @@
 	}
 
 	if cfg.HealthConfig.Enabled {
 		healthProbeServer, err := createHealthProbe(cfg, healthChecker)
 		if err != nil {
 			return nil, fmt.Errorf("error creating health probe: %w", err)
 		}
 
 		if err := mgr.Add(&healthProbeServer); err != nil {
 			return nil, fmt.Errorf("error adding health probe: %w", err)
 		}
 	}