N+ support; fix channel blocking

Author: sjberman

charts/nginx-gateway-fabric/templates/tmp-nginx-agent-conf.yaml

@@ -18,5 +18,8 @@ data:
     - configuration
     - certificates
     - metrics
+    {{- if .Values.nginx.plus }}
+    - api-action
+    {{- end }}
     log:
       level: debug

deploy/experimental-nginx-plus/deploy.yaml

@@ -163,6 +163,7 @@ data:
     - configuration
     - certificates
     - metrics
+    - api-action
     log:
       level: debug
 kind: ConfigMap

deploy/nginx-plus/deploy.yaml

@@ -158,6 +158,7 @@ data:
     - configuration
     - certificates
     - metrics
+    - api-action
     log:
       level: debug
 kind: ConfigMap

deploy/snippets-filters-nginx-plus/deploy.yaml

@@ -160,6 +160,7 @@ data:
     - configuration
     - certificates
     - metrics
+    - api-action
     log:
       level: debug
 kind: ConfigMap

internal/mode/static/handler.go

@@ -185,7 +185,12 @@ func (h *eventHandlerImpl) HandleEventBatch(ctx context.Context, logger logr.Log
 		h.setLatestConfiguration(&cfg)
 
 		if h.cfg.plus {
-			h.cfg.nginxUpdater.UpdateUpstreamServers()
+			// TODO(sberman): hardcode this deployment name until we support provisioning data planes
+			deployment := types.NamespacedName{
+				Name:      "tmp-nginx-deployment",
+				Namespace: h.cfg.gatewayPodConfig.Namespace,
+			}
+			err = h.cfg.nginxUpdater.UpdateUpstreamServers(ctx, deployment, cfg)
 		} else {
 			err = h.updateNginxConf(ctx, cfg)
 		}
@@ -314,7 +319,9 @@ func (h *eventHandlerImpl) updateNginxConf(ctx context.Context, conf dataplane.C
 
 	// If using NGINX Plus, update upstream servers using the API.
 	if h.cfg.plus {
-		h.cfg.nginxUpdater.UpdateUpstreamServers()
+		if err := h.cfg.nginxUpdater.UpdateUpstreamServers(ctx, deployment, conf); err != nil {
+			return err
+		}
 	}
 
 	return nil

internal/mode/static/nginx/agent/agent.go

@@ -2,13 +2,19 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/go-logr/logr"
+	pb "github.com/nginx/agent/v3/api/grpc/mpi/v1"
+	"google.golang.org/protobuf/types/known/structpb"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/nginxinc/nginx-gateway-fabric/internal/mode/static/nginx/agent/broadcast"
 	agentgrpc "github.com/nginxinc/nginx-gateway-fabric/internal/mode/static/nginx/agent/grpc"
+	"github.com/nginxinc/nginx-gateway-fabric/internal/mode/static/state/dataplane"
+	"github.com/nginxinc/nginx-gateway-fabric/internal/mode/static/state/resolver"
 )
 
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
@@ -17,8 +23,16 @@ import (
 
 // NginxUpdater is an interface for updating NGINX using the NGINX agent.
 type NginxUpdater interface {
-	UpdateConfig(context.Context, types.NamespacedName, []File) error
-	UpdateUpstreamServers()
+	UpdateConfig(
+		ctx context.Context,
+		deploymentNsName types.NamespacedName,
+		files []File,
+	) error
+	UpdateUpstreamServers(
+		ctx context.Context,
+		deploymentNsName types.NamespacedName,
+		conf dataplane.Configuration,
+	) error
 }
 
 // NginxUpdaterImpl implements the NginxUpdater interface.
@@ -76,10 +90,76 @@ func (n *NginxUpdaterImpl) UpdateConfig(ctx context.Context, nsName types.Namesp
 
 // UpdateUpstreamServers sends an APIRequest to the agent to update upstream servers using the NGINX Plus API.
 // Only applicable when using NGINX Plus.
-func (n *NginxUpdaterImpl) UpdateUpstreamServers() {
+func (n *NginxUpdaterImpl) UpdateUpstreamServers(
+	ctx context.Context,
+	nsName types.NamespacedName,
+	conf dataplane.Configuration,
+) error {
 	if !n.plus {
-		return
+		return nil
 	}
 
 	n.logger.Info("Updating upstream servers using NGINX Plus API")
+
+	deployment := n.nginxDeployments.GetOrStore(ctx, nsName)
+	if deployment == nil {
+		return fmt.Errorf("failed to register nginx deployment %q", nsName.Name)
+	}
+	broadcaster := deployment.GetBroadcaster()
+
+	var updateErr error
+	for _, upstream := range conf.Upstreams {
+		msg := broadcast.NginxAgentMessage{
+			Type: broadcast.APIRequest,
+			NGINXPlusAction: &pb.NGINXPlusAction{
+				Action: &pb.NGINXPlusAction_UpdateHttpUpstreamServers{
+					UpdateHttpUpstreamServers: buildUpstreamServers(upstream),
+				},
+			},
+		}
+
+		if err := broadcaster.Send(msg); err != nil {
+			updateErr = errors.Join(updateErr, fmt.Errorf(
+				"couldn't update upstream %q via the API: %w", upstream.Name, err))
+		}
+	}
+
+	return updateErr
+}
+
+func buildUpstreamServers(upstream dataplane.Upstream) *pb.UpdateHTTPUpstreamServers {
+	servers := make([]*structpb.Struct, 0, len(upstream.Endpoints))
+
+	for _, endpoint := range upstream.Endpoints {
+		port, format := getPortAndIPFormat(endpoint)
+		value := fmt.Sprintf(format, endpoint.Address, port)
+
+		server := &structpb.Struct{
+			Fields: map[string]*structpb.Value{
+				"server": structpb.NewStringValue(value),
+			},
+		}
+
+		servers = append(servers, server)
+	}
+
+	return &pb.UpdateHTTPUpstreamServers{
+		HttpUpstreamName: upstream.Name,
+		Servers:          servers,
+	}
+}
+
+func getPortAndIPFormat(ep resolver.Endpoint) (string, string) {
+	var port string
+
+	if ep.Port != 0 {
+		port = fmt.Sprintf(":%d", ep.Port)
+	}
+
+	format := "%s%s"
+	if ep.IPv6 {
+		format = "[%s]%s"
+	}
+
+	return port, format
 }