Use state file for updating N+ upstreams (#2897)

Problem: When splitting the data and control plane, we want to be able to maintain the ability to dynamically update upstreams without a reload if using NGINX Plus. With the current process, we write the upstreams into the nginx conf but don't reload, then call the API to actually do the update. Writing into the conf will trigger a reload from the agent once we split, however.

There were also two bugs:
- if metrics were disabled, the nginx plus client was not initialized, preventing API calls from occuring and instead a reload occurred
- stream upstreams were not updated with the API

Solution: Don't write the upstream servers into the nginx conf anymore when using NGINX Plus. Instead, utilize the `state` file option that the NGINX Plus API will populate with the upstream servers. This way we can just call the API and don't unintentionally reload by writing servers into the conf.

Also added support for updating stream upstreams, and fixed the initialization bug.

Author: sjberman

internal/mode/static/handler.go

@@ -2,6 +2,7 @@ package static
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -182,11 +183,11 @@ func (h *eventHandlerImpl) HandleEventBatch(ctx context.Context, logger logr.Log
 
 		h.setLatestConfiguration(&cfg)
 
-		err = h.updateUpstreamServers(
-			ctx,
-			logger,
-			cfg,
-		)
+		if h.cfg.plus {
+			err = h.updateUpstreamServers(cfg)
+		} else {
+			err = h.updateNginxConf(ctx, cfg)
+		}
 	case state.ClusterStateChange:
 		h.version++
 		cfg := dataplane.BuildConfiguration(ctx, gr, h.cfg.serviceResolver, h.version)
@@ -198,10 +199,7 @@ func (h *eventHandlerImpl) HandleEventBatch(ctx context.Context, logger logr.Log
 
 		h.setLatestConfiguration(&cfg)
 
-		err = h.updateNginxConf(
-			ctx,
-			cfg,
-		)
+		err = h.updateNginxConf(ctx, cfg)
 	}
 
 	var nginxReloadRes status.NginxReloadResult
@@ -306,7 +304,10 @@ func (h *eventHandlerImpl) parseAndCaptureEvent(ctx context.Context, logger logr
 }
 
 // updateNginxConf updates nginx conf files and reloads nginx.
-func (h *eventHandlerImpl) updateNginxConf(ctx context.Context, conf dataplane.Configuration) error {
+func (h *eventHandlerImpl) updateNginxConf(
+	ctx context.Context,
+	conf dataplane.Configuration,
+) error {
 	files := h.cfg.generator.Generate(conf)
 	if err := h.cfg.nginxFileMgr.ReplaceFiles(files); err != nil {
 		return fmt.Errorf("failed to replace NGINX configuration files: %w", err)
@@ -316,89 +317,114 @@ func (h *eventHandlerImpl) updateNginxConf(ctx context.Context, conf dataplane.C
 		return fmt.Errorf("failed to reload NGINX: %w", err)
 	}
 
+	// If using NGINX Plus, update upstream servers using the API.
+	if err := h.updateUpstreamServers(conf); err != nil {
+		return fmt.Errorf("failed to update upstream servers: %w", err)
+	}
+
 	return nil
 }
 
-// updateUpstreamServers is called only when endpoints have changed. It updates nginx conf files and then:
-// - if using NGINX Plus, determines which servers have changed and uses the N+ API to update them;
-// - otherwise if not using NGINX Plus, or an error was returned from the API, reloads nginx.
-func (h *eventHandlerImpl) updateUpstreamServers(
-	ctx context.Context,
-	logger logr.Logger,
-	conf dataplane.Configuration,
-) error {
-	isPlus := h.cfg.nginxRuntimeMgr.IsPlus()
-
-	files := h.cfg.generator.Generate(conf)
-	if err := h.cfg.nginxFileMgr.ReplaceFiles(files); err != nil {
-		return fmt.Errorf("failed to replace NGINX configuration files: %w", err)
+// updateUpstreamServers determines which servers have changed and uses the NGINX Plus API to update them.
+// Only applicable when using NGINX Plus.
+func (h *eventHandlerImpl) updateUpstreamServers(conf dataplane.Configuration) error {
+	if !h.cfg.plus {
+		return nil
 	}
 
-	reload := func() error {
-		if err := h.cfg.nginxRuntimeMgr.Reload(ctx, conf.Version); err != nil {
-			return fmt.Errorf("failed to reload NGINX: %w", err)
-		}
+	prevUpstreams, prevStreamUpstreams, err := h.cfg.nginxRuntimeMgr.GetUpstreams()
+	if err != nil {
+		return fmt.Errorf("failed to get upstreams from API: %w", err)
+	}
 
-		return nil
+	type upstream struct {
+		name    string
+		servers []ngxclient.UpstreamServer
 	}
+	var upstreams []upstream
 
-	if isPlus {
-		type upstream struct {
-			name    string
-			servers []ngxclient.UpstreamServer
+	for _, u := range conf.Upstreams {
+		confUpstream := upstream{
+			name:    u.Name,
+			servers: ngxConfig.ConvertEndpoints(u.Endpoints),
 		}
-		var upstreams []upstream
 
-		prevUpstreams, err := h.cfg.nginxRuntimeMgr.GetUpstreams()
-		if err != nil {
-			logger.Error(err, "failed to get upstreams from API, reloading configuration instead")
-			return reload()
+		if u, ok := prevUpstreams[confUpstream.name]; ok {
+			if !serversEqual(confUpstream.servers, u.Peers) {
+				upstreams = append(upstreams, confUpstream)
+			}
 		}
+	}
 
-		for _, u := range conf.Upstreams {
-			confUpstream := upstream{
-				name:    u.Name,
-				servers: ngxConfig.ConvertEndpoints(u.Endpoints),
-			}
+	type streamUpstream struct {
+		name    string
+		servers []ngxclient.StreamUpstreamServer
+	}
+	var streamUpstreams []streamUpstream
 
-			if u, ok := prevUpstreams[confUpstream.name]; ok {
-				if !serversEqual(confUpstream.servers, u.Peers) {
-					upstreams = append(upstreams, confUpstream)
-				}
-			}
+	for _, u := range conf.StreamUpstreams {
+		confUpstream := streamUpstream{
+			name:    u.Name,
+			servers: ngxConfig.ConvertStreamEndpoints(u.Endpoints),
 		}
 
-		var reloadPlus bool
-		for _, upstream := range upstreams {
-			if err := h.cfg.nginxRuntimeMgr.UpdateHTTPServers(upstream.name, upstream.servers); err != nil {
-				logger.Error(
-					err, "couldn't update upstream via the API, reloading configuration instead",
-					"upstreamName", upstream.name,
-				)
-				reloadPlus = true
+		if u, ok := prevStreamUpstreams[confUpstream.name]; ok {
+			if !serversEqual(confUpstream.servers, u.Peers) {
+				streamUpstreams = append(streamUpstreams, confUpstream)
 			}
 		}
+	}
 
-		if !reloadPlus {
-			return nil
+	var updateErr error
+	for _, upstream := range upstreams {
+		if err := h.cfg.nginxRuntimeMgr.UpdateHTTPServers(upstream.name, upstream.servers); err != nil {
+			updateErr = errors.Join(updateErr, fmt.Errorf(
+				"couldn't update upstream %q via the API: %w", upstream.name, err))
 		}
 	}
 
-	return reload()
+	for _, upstream := range streamUpstreams {
+		if err := h.cfg.nginxRuntimeMgr.UpdateStreamServers(upstream.name, upstream.servers); err != nil {
+			updateErr = errors.Join(updateErr, fmt.Errorf(
+				"couldn't update stream upstream %q via the API: %w", upstream.name, err))
+		}
+	}
+
+	return updateErr
 }
 
-func serversEqual(newServers []ngxclient.UpstreamServer, oldServers []ngxclient.Peer) bool {
+// serversEqual accepts lists of either UpstreamServer/Peer or StreamUpstreamServer/StreamPeer and determines
+// if the server names within these lists are equal.
+func serversEqual[
+	upstreamServer ngxclient.UpstreamServer | ngxclient.StreamUpstreamServer,
+	peer ngxclient.Peer | ngxclient.StreamPeer,
+](newServers []upstreamServer, oldServers []peer) bool {
 	if len(newServers) != len(oldServers) {
 		return false
 	}
 
+	getServerVal := func(T any) string {
+		var server string
+		switch t := T.(type) {
+		case ngxclient.UpstreamServer:
+			server = t.Server
+		case ngxclient.StreamUpstreamServer:
+			server = t.Server
+		case ngxclient.Peer:
+			server = t.Server
+		case ngxclient.StreamPeer:
+			server = t.Server
+		}
+		return server
+	}
+
 	diff := make(map[string]struct{}, len(newServers))
 	for _, s := range newServers {
-		diff[s.Server] = struct{}{}
+		diff[getServerVal(s)] = struct{}{}
 	}
 
 	for _, s := range oldServers {
-		if _, ok := diff[s.Server]; !ok {
+		if _, ok := diff[getServerVal(s)]; !ok {
 			return false
 		}
 	}

internal/mode/static/handler_test.go

@@ -423,20 +423,29 @@ var _ = Describe("eventHandler", func() {
 					},
 				},
 			}
-			fakeNginxRuntimeMgr.GetUpstreamsReturns(upstreams, nil)
+
+			streamUpstreams := ngxclient.StreamUpstreams{
+				"two": ngxclient.StreamUpstream{
+					Peers: []ngxclient.StreamPeer{
+						{Server: "server2"},
+					},
+				},
+			}
+
+			fakeNginxRuntimeMgr.GetUpstreamsReturns(upstreams, streamUpstreams, nil)
 		})
 
 		When("running NGINX Plus", func() {
 			It("should call the NGINX Plus API", func() {
-				fakeNginxRuntimeMgr.IsPlusReturns(true)
+				handler.cfg.plus = true
 
 				handler.HandleEventBatch(context.Background(), ctlrZap.New(), batch)
 
 				dcfg := dataplane.GetDefaultConfiguration(&graph.Graph{}, 1)
 				Expect(helpers.Diff(handler.GetLatestConfiguration(), &dcfg)).To(BeEmpty())
 
-				Expect(fakeGenerator.GenerateCallCount()).To(Equal(1))
-				Expect(fakeNginxFileMgr.ReplaceFilesCallCount()).To(Equal(1))
+				Expect(fakeGenerator.GenerateCallCount()).To(Equal(0))
+				Expect(fakeNginxFileMgr.ReplaceFilesCallCount()).To(Equal(0))
 				Expect(fakeNginxRuntimeMgr.GetUpstreamsCallCount()).To(Equal(1))
 			})
 		})
@@ -463,19 +472,11 @@ var _ = Describe("eventHandler", func() {
 					Name: "one",
 				},
 			},
-		}
-
-		type callCounts struct {
-			generate int
-			update   int
-			reload   int
-		}
-
-		assertCallCounts := func(cc callCounts) {
-			Expect(fakeGenerator.GenerateCallCount()).To(Equal(cc.generate))
-			Expect(fakeNginxFileMgr.ReplaceFilesCallCount()).To(Equal(cc.generate))
-			Expect(fakeNginxRuntimeMgr.UpdateHTTPServersCallCount()).To(Equal(cc.update))
-			Expect(fakeNginxRuntimeMgr.ReloadCallCount()).To(Equal(cc.reload))
+			StreamUpstreams: []dataplane.Upstream{
+				{
+					Name: "two",
+				},
+			},
 		}
 
 		BeforeEach(func() {
@@ -486,47 +487,49 @@ var _ = Describe("eventHandler", func() {
 					},
 				},
 			}
-			fakeNginxRuntimeMgr.GetUpstreamsReturns(upstreams, nil)
+
+			streamUpstreams := ngxclient.StreamUpstreams{
+				"two": ngxclient.StreamUpstream{
+					Peers: []ngxclient.StreamPeer{
+						{Server: "server2"},
+					},
+				},
+			}
+
+			fakeNginxRuntimeMgr.GetUpstreamsReturns(upstreams, streamUpstreams, nil)
 		})
 
 		When("running NGINX Plus", func() {
 			BeforeEach(func() {
-				fakeNginxRuntimeMgr.IsPlusReturns(true)
+				handler.cfg.plus = true
 			})
 
 			It("should update servers using the NGINX Plus API", func() {
-				Expect(handler.updateUpstreamServers(context.Background(), ctlrZap.New(), conf)).To(Succeed())
-
-				assertCallCounts(callCounts{generate: 1, update: 1, reload: 0})
+				Expect(handler.updateUpstreamServers(conf)).To(Succeed())
+				Expect(fakeNginxRuntimeMgr.UpdateHTTPServersCallCount()).To(Equal(1))
 			})
 
-			It("should reload when GET API returns an error", func() {
-				fakeNginxRuntimeMgr.GetUpstreamsReturns(nil, errors.New("error"))
-				Expect(handler.updateUpstreamServers(context.Background(), ctlrZap.New(), conf)).To(Succeed())
-
-				assertCallCounts(callCounts{generate: 1, update: 0, reload: 1})
+			It("should return error when GET API returns an error", func() {
+				fakeNginxRuntimeMgr.GetUpstreamsReturns(nil, nil, errors.New("error"))
+				Expect(handler.updateUpstreamServers(conf)).ToNot(Succeed())
 			})
 
-			It("should reload when POST API returns an error", func() {
+			It("should return error when UpdateHTTPServers API returns an error", func() {
 				fakeNginxRuntimeMgr.UpdateHTTPServersReturns(errors.New("error"))
-				Expect(handler.updateUpstreamServers(context.Background(), ctlrZap.New(), conf)).To(Succeed())
+				Expect(handler.updateUpstreamServers(conf)).ToNot(Succeed())
+			})
 
-				assertCallCounts(callCounts{generate: 1, update: 1, reload: 1})
+			It("should return error when UpdateStreamServers API returns an error", func() {
+				fakeNginxRuntimeMgr.UpdateStreamServersReturns(errors.New("error"))
+				Expect(handler.updateUpstreamServers(conf)).ToNot(Succeed())
 			})
 		})
 
 		When("not running NGINX Plus", func() {
-			It("should update servers by reloading", func() {
-				Expect(handler.updateUpstreamServers(context.Background(), ctlrZap.New(), conf)).To(Succeed())
-
-				assertCallCounts(callCounts{generate: 1, update: 0, reload: 1})
-			})
+			It("should not do anything", func() {
+				Expect(handler.updateUpstreamServers(conf)).To(Succeed())
 
-			It("should return an error when reloading fails", func() {
-				fakeNginxRuntimeMgr.ReloadReturns(errors.New("error"))
-				Expect(handler.updateUpstreamServers(context.Background(), ctlrZap.New(), conf)).ToNot(Succeed())
-
-				assertCallCounts(callCounts{generate: 1, update: 0, reload: 1})
+				Expect(fakeNginxRuntimeMgr.UpdateHTTPServersCallCount()).To(Equal(0))
 			})
 		})
 	})
@@ -612,7 +615,7 @@ var _ = Describe("eventHandler", func() {
 })
 
 var _ = Describe("serversEqual", func() {
-	DescribeTable("determines if server lists are equal",
+	DescribeTable("determines if HTTP server lists are equal",
 		func(newServers []ngxclient.UpstreamServer, oldServers []ngxclient.Peer, equal bool) {
 			Expect(serversEqual(newServers, oldServers)).To(Equal(equal))
 		},
@@ -649,6 +652,43 @@ var _ = Describe("serversEqual", func() {
 			true,
 		),
 	)
+	DescribeTable("determines if stream server lists are equal",
+		func(newServers []ngxclient.StreamUpstreamServer, oldServers []ngxclient.StreamPeer, equal bool) {
+			Expect(serversEqual(newServers, oldServers)).To(Equal(equal))
+		},
+		Entry("different length",
+			[]ngxclient.StreamUpstreamServer{
+				{Server: "server1"},
+			},
+			[]ngxclient.StreamPeer{
+				{Server: "server1"},
+				{Server: "server2"},
+			},
+			false,
+		),
+		Entry("differing elements",
+			[]ngxclient.StreamUpstreamServer{
+				{Server: "server1"},
+				{Server: "server2"},
+			},
+			[]ngxclient.StreamPeer{
+				{Server: "server1"},
+				{Server: "server3"},
+			},
+			false,
+		),
+		Entry("same elements",
+			[]ngxclient.StreamUpstreamServer{
+				{Server: "server1"},
+				{Server: "server2"},
+			},
+			[]ngxclient.StreamPeer{
+				{Server: "server1"},
+				{Server: "server2"},
+			},
+			true,
+		),
+	)
 })
 
 var _ = Describe("getGatewayAddresses", func() {