Revert "CP/DP split: Add leader election (#3092)"

This reverts commit a5c989e.

Author: salonichf5

docs/proposals/control-data-plane-split/graph-conns.png

@@ -34,33 +34,29 @@ func (r *LeaderOrNonLeader) NeedLeaderElection() bool {
 	return false
 }
 
-// CallFunctionsAfterBecameLeader is a Runnable that will call the given functions when the current instance becomes
+// EnableAfterBecameLeader is a Runnable that will call the enable function when the current instance becomes
 // the leader.
-type CallFunctionsAfterBecameLeader struct {
-	enableFunctions []func(context.Context)
+type EnableAfterBecameLeader struct {
+	enable func(context.Context)
 }
 
 var (
-	_ manager.LeaderElectionRunnable = &CallFunctionsAfterBecameLeader{}
-	_ manager.Runnable               = &CallFunctionsAfterBecameLeader{}
+	_ manager.LeaderElectionRunnable = &EnableAfterBecameLeader{}
+	_ manager.Runnable               = &EnableAfterBecameLeader{}
 )
 
-// NewCallFunctionsAfterBecameLeader creates a new CallFunctionsAfterBecameLeader Runnable.
-func NewCallFunctionsAfterBecameLeader(
-	enableFunctions []func(context.Context),
-) *CallFunctionsAfterBecameLeader {
-	return &CallFunctionsAfterBecameLeader{
-		enableFunctions: enableFunctions,
+// NewEnableAfterBecameLeader creates a new EnableAfterBecameLeader Runnable.
+func NewEnableAfterBecameLeader(enable func(context.Context)) *EnableAfterBecameLeader {
+	return &EnableAfterBecameLeader{
+		enable: enable,
 	}
 }
 
-func (j *CallFunctionsAfterBecameLeader) Start(ctx context.Context) error {
-	for _, f := range j.enableFunctions {
-		f(ctx)
-	}
+func (j *EnableAfterBecameLeader) Start(ctx context.Context) error {
+	j.enable(ctx)
 	return nil
 }
 
-func (j *CallFunctionsAfterBecameLeader) NeedLeaderElection() bool {
+func (j *EnableAfterBecameLeader) NeedLeaderElection() bool {
 	return true
 }

internal/framework/runnables/runnables.go

@@ -23,25 +23,19 @@ func TestLeaderOrNonLeader(t *testing.T) {
 	g.Expect(leaderOrNonLeader.NeedLeaderElection()).To(BeFalse())
 }
 
-func TestCallFunctionsAfterBecameLeader(t *testing.T) {
+func TestEnableAfterBecameLeader(t *testing.T) {
 	t.Parallel()
-	statusUpdaterEnabled := false
-	healthCheckEnableLeader := false
-	eventHandlerEnabled := false
-
-	callFunctionsAfterBecameLeader := NewCallFunctionsAfterBecameLeader([]func(ctx context.Context){
-		func(_ context.Context) { statusUpdaterEnabled = true },
-		func(_ context.Context) { healthCheckEnableLeader = true },
-		func(_ context.Context) { eventHandlerEnabled = true },
+	enabled := false
+	enableAfterBecameLeader := NewEnableAfterBecameLeader(func(_ context.Context) {
+		enabled = true
 	})
 
 	g := NewWithT(t)
-	g.Expect(callFunctionsAfterBecameLeader.NeedLeaderElection()).To(BeTrue())
+	g.Expect(enableAfterBecameLeader.NeedLeaderElection()).To(BeTrue())
+	g.Expect(enabled).To(BeFalse())
 
-	err := callFunctionsAfterBecameLeader.Start(context.Background())
+	err := enableAfterBecameLeader.Start(context.Background())
 	g.Expect(err).ToNot(HaveOccurred())
 
-	g.Expect(statusUpdaterEnabled).To(BeTrue())
-	g.Expect(healthCheckEnableLeader).To(BeTrue())
-	g.Expect(eventHandlerEnabled).To(BeTrue())
+	g.Expect(enabled).To(BeTrue())
 }

internal/framework/runnables/runnables_test.go

@@ -161,27 +161,14 @@ func (h *eventHandlerImpl) HandleEventBatch(ctx context.Context, logger logr.Log
 
 	changeType, gr := h.cfg.processor.Process()
 
-	// Once we've processed resources on startup and built our first graph, mark the Pod as having built the graph.
-	if !h.cfg.graphBuiltHealthChecker.graphBuilt {
-		h.cfg.graphBuiltHealthChecker.setGraphBuilt()
-	}
-
-	// if this Pod is not the leader or does not have the leader lease yet,
-	// the nginx conf should not be updated.
-	if !h.cfg.graphBuiltHealthChecker.leader {
-		return
+	// Once we've processed resources on startup and built our first graph, mark the Pod as ready.
+	if !h.cfg.graphBuiltHealthChecker.ready {
+		h.cfg.graphBuiltHealthChecker.setAsReady()
 	}
 
 	h.sendNginxConfig(ctx, logger, gr, changeType)
 }
 
-func (h *eventHandlerImpl) eventHandlerEnable(ctx context.Context) {
-	// Latest graph is guaranteed to not be nil since the leader election process takes longer than
-	// the initial call to HandleEventBatch when NGF starts up. And GatewayClass will typically always exist which
-	// triggers an event.
-	h.sendNginxConfig(ctx, h.cfg.logger, h.cfg.processor.GetLatestGraph(), state.ClusterStateChange)
-}
-
 func (h *eventHandlerImpl) sendNginxConfig(
 	ctx context.Context,
 	logger logr.Logger,

internal/mode/static/handler.go

@@ -141,9 +141,7 @@ var _ = Describe("eventHandler", func() {
 			metricsCollector:         collectors.NewControllerNoopCollector(),
 			updateGatewayClassStatus: true,
 		})
-		Expect(handler.cfg.graphBuiltHealthChecker.graphBuilt).To(BeFalse())
-
-		handler.cfg.graphBuiltHealthChecker.leader = true
+		Expect(handler.cfg.graphBuiltHealthChecker.ready).To(BeFalse())
 	})
 
 	AfterEach(func() {
@@ -177,7 +175,7 @@ var _ = Describe("eventHandler", func() {
 		})
 
 		AfterEach(func() {
-			Expect(handler.cfg.graphBuiltHealthChecker.graphBuilt).To(BeTrue())
+			Expect(handler.cfg.graphBuiltHealthChecker.ready).To(BeTrue())
 		})
 
 		When("a batch has one event", func() {
@@ -458,35 +456,21 @@ 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{}}
 		batch := []interface{}{e}
 		readyChannel := handler.cfg.graphBuiltHealthChecker.getReadyCh()
 
 		fakeProcessor.ProcessReturns(state.ClusterStateChange, &graph.Graph{Gateway: &graph.Gateway{Valid: true}})
 
-		handler.HandleEventBatch(context.Background(), logr.Discard(), batch)
-
-		// graph is built, but since the graphBuiltHealthChecker.leader is false, configuration isn't created and
-		// the readyCheck fails
-		Expect(handler.cfg.graphBuiltHealthChecker.graphBuilt).To(BeTrue())
-		Expect(handler.GetLatestConfiguration()).To(BeNil())
 		Expect(handler.cfg.graphBuiltHealthChecker.readyCheck(nil)).ToNot(Succeed())
-		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(ctx)
-		handler.eventHandlerEnable(ctx)
+		handler.HandleEventBatch(context.Background(), logr.Discard(), batch)
 
-		// nginx conf has been set
 		dcfg := dataplane.GetDefaultConfiguration(&graph.Graph{}, 1)
 		Expect(helpers.Diff(handler.GetLatestConfiguration(), &dcfg)).To(BeEmpty())
 
-		// ready check is also set
+		Expect(readyChannel).To(BeClosed())
+
 		Expect(handler.cfg.graphBuiltHealthChecker.readyCheck(nil)).To(Succeed())
-		Expect(handler.cfg.graphBuiltHealthChecker.getReadyCh()).To(BeClosed())
 	})
 
 	It("should panic for an unknown event type", func() {

internal/mode/static/handler_test.go

@@ -1,17 +1,9 @@
 package static
 
 import (
-	"context"
 	"errors"
-	"fmt"
-	"net"
 	"net/http"
 	"sync"
-	"time"
-
-	"sigs.k8s.io/controller-runtime/pkg/manager"
-
-	"github.com/nginx/nginx-gateway-fabric/internal/mode/static/config"
 )
 
 // newGraphBuiltHealthChecker creates a new graphBuiltHealthChecker.
@@ -21,94 +13,37 @@ func newGraphBuiltHealthChecker() *graphBuiltHealthChecker {
 	}
 }
 
-// graphBuiltHealthChecker is used to check if the NGF Pod is ready. The NGF Pod is ready if the initial graph has
-// been built and if it is leader.
+// graphBuiltHealthChecker is used to check if the initial graph is built and the NGF Pod is ready.
 type graphBuiltHealthChecker struct {
 	// readyCh is a channel that is initialized in newGraphBuiltHealthChecker and represents if the NGF Pod is ready.
-	readyCh    chan struct{}
-	lock       sync.RWMutex
-	graphBuilt bool
-	leader     bool
-}
-
-// createHealthProbe creates a Server runnable to serve as our health and readiness checker.
-func createHealthProbe(cfg config.Config, healthChecker *graphBuiltHealthChecker) (manager.Server, error) {
-	// we chose to create our own health probe server instead of using the controller-runtime one because
-	// of repetitive log which would flood our logs on non-ready non-leader NGF Pods. This health probe is
-	// similar to the controller-runtime's health probe.
-
-	mux := http.NewServeMux()
-
-	// copy of controller-runtime sane defaults for new http.Server
-	s := &http.Server{
-		Handler:           mux,
-		MaxHeaderBytes:    1 << 20,
-		IdleTimeout:       90 * time.Second, // matches http.DefaultTransport keep-alive timeout
-		ReadHeaderTimeout: 32 * time.Second,
-	}
-
-	mux.HandleFunc(readinessEndpointName, healthChecker.readyHandler)
-
-	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", cfg.HealthConfig.Port))
-	if err != nil {
-		return manager.Server{},
-			fmt.Errorf("error listening on %s: %w", fmt.Sprintf(":%d", cfg.HealthConfig.Port), err)
-	}
-
-	return manager.Server{
-		Name:     "health probe",
-		Server:   s,
-		Listener: ln,
-	}, nil
-}
-
-func (h *graphBuiltHealthChecker) readyHandler(resp http.ResponseWriter, req *http.Request) {
-	if err := h.readyCheck(req); err != nil {
-		resp.WriteHeader(http.StatusServiceUnavailable)
-	} else {
-		resp.WriteHeader(http.StatusOK)
-	}
+	readyCh chan struct{}
+	lock    sync.RWMutex
+	ready   bool
 }
 
 // readyCheck returns the ready-state of the Pod. It satisfies the controller-runtime Checker type.
-// We are considered ready after the first graph is built and if the NGF Pod is leader.
+// We are considered ready after the first graph is built.
 func (h *graphBuiltHealthChecker) readyCheck(_ *http.Request) error {
 	h.lock.RLock()
 	defer h.lock.RUnlock()
 
-	if !h.leader {
-		return errors.New("this Pod is not currently leader")
-	}
-
-	if !h.graphBuilt {
-		return errors.New("control plane initial graph has not been built")
+	if !h.ready {
+		return errors.New("control plane is not yet ready")
 	}
 
 	return nil
 }
 
-// setGraphBuilt marks the health check as having the initial graph built.
-func (h *graphBuiltHealthChecker) setGraphBuilt() {
+// setAsReady marks the health check as ready.
+func (h *graphBuiltHealthChecker) setAsReady() {
 	h.lock.Lock()
 	defer h.lock.Unlock()
 
-	h.graphBuilt = true
+	h.ready = true
+	close(h.readyCh)
 }
 
 // getReadyCh returns a read-only channel, which determines if the NGF Pod is ready.
 func (h *graphBuiltHealthChecker) getReadyCh() <-chan struct{} {
 	return h.readyCh
 }
-
-// setAsLeader marks the health check as leader.
-func (h *graphBuiltHealthChecker) setAsLeader(_ context.Context) {
-	h.lock.Lock()
-	defer h.lock.Unlock()
-
-	h.leader = true
-
-	// setGraphBuilt should already have been called when processing the resources on startup because the leader
-	// election process takes longer than the initial call to HandleEventBatch. Thus, the NGF Pod should be marked as
-	// ready and have this channel be closed.
-	close(h.readyCh)
-}

internal/mode/static/health.go

@@ -1,99 +1,17 @@
 package static
 
 import (
-	"context"
-	"errors"
-	"net"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	. "github.com/onsi/gomega"
-
-	"github.com/nginx/nginx-gateway-fabric/internal/mode/static/config"
 )
 
 func TestReadyCheck(t *testing.T) {
 	t.Parallel()
 	g := NewWithT(t)
 	healthChecker := newGraphBuiltHealthChecker()
+	g.Expect(healthChecker.readyCheck(nil)).ToNot(Succeed())
 
-	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 Pod is not currently leader")))
-
-	healthChecker.graphBuilt = false
-	healthChecker.leader = true
-	g.Expect(healthChecker.readyCheck(nil)).
-		To(MatchError(errors.New("control plane initial graph has not been built")))
-
-	healthChecker.graphBuilt = true
+	healthChecker.ready = true
 	g.Expect(healthChecker.readyCheck(nil)).To(Succeed())
 }
-
-func TestSetAsLeader(t *testing.T) {
-	t.Parallel()
-	g := NewWithT(t)
-	healthChecker := newGraphBuiltHealthChecker()
-
-	g.Expect(healthChecker.leader).To(BeFalse())
-	g.Expect(healthChecker.readyCh).ShouldNot(BeClosed())
-
-	healthChecker.setAsLeader(context.Background())
-
-	g.Expect(healthChecker.leader).To(BeTrue())
-	g.Expect(healthChecker.readyCh).To(BeClosed())
-}
-
-func TestSetGraphBuilt(t *testing.T) {
-	t.Parallel()
-	g := NewWithT(t)
-	healthChecker := newGraphBuiltHealthChecker()
-
-	g.Expect(healthChecker.graphBuilt).To(BeFalse())
-
-	healthChecker.setGraphBuilt()
-
-	g.Expect(healthChecker.graphBuilt).To(BeTrue())
-}
-
-func TestReadyHandler(t *testing.T) {
-	t.Parallel()
-	g := NewWithT(t)
-	healthChecker := newGraphBuiltHealthChecker()
-
-	r := httptest.NewRequest(http.MethodGet, "/readyz", nil)
-	w := httptest.NewRecorder()
-
-	healthChecker.readyHandler(w, r)
-	g.Expect(w.Result().StatusCode).To(Equal(http.StatusServiceUnavailable))
-
-	healthChecker.graphBuilt = true
-	healthChecker.leader = true
-
-	w = httptest.NewRecorder()
-	healthChecker.readyHandler(w, r)
-	g.Expect(w.Result().StatusCode).To(Equal(http.StatusOK))
-}
-
-func TestCreateHealthProbe(t *testing.T) {
-	t.Parallel()
-	g := NewWithT(t)
-
-	healthChecker := newGraphBuiltHealthChecker()
-
-	cfg := config.Config{HealthConfig: config.HealthConfig{Port: 100000}}
-	_, err := createHealthProbe(cfg, healthChecker)
-	g.Expect(err).To(MatchError("error listening on :100000: listen tcp: address 100000: invalid port"))
-
-	cfg = config.Config{HealthConfig: config.HealthConfig{Port: 8081}}
-	hp, err := createHealthProbe(cfg, healthChecker)
-	g.Expect(err).ToNot(HaveOccurred())
-
-	addr, ok := (hp.Listener.Addr()).(*net.TCPAddr)
-	g.Expect(ok).To(BeTrue())
-
-	g.Expect(addr.Port).To(Equal(cfg.HealthConfig.Port))
-	g.Expect(hp.Server).ToNot(BeNil())
-}