Add agent registration

.github/workflows/ci.yml

@@ -5,12 +5,14 @@ on:
     branches:
       - main
       - release-*
+      - feature/cp-dp-separation
     tags:
       - 'v[0-9]+.[0-9]+.[0-9]+*'
   pull_request:
     branches:
       - main
       - release-*
+      - feature/cp-dp-separation
     types:
       - opened
       - reopened

build/nginx-with-agent/clusterip.yaml

@@ -0,0 +1,14 @@
+# This service is for testing purposes for the nginx agent
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx-gateway
+  namespace: nginx-gateway
+spec:
+  ports:
+  - port: 54789
+    targetPort: 54789
+    protocol: TCP
+    name: grpc
+  selector:
+    app: nginx-gateway

build/nginx-with-agent/nginx-agent.conf

@@ -10,41 +10,31 @@
 log:
   # set log level (panic, fatal, error, info, debug, trace; default "info")
   level: info
-  # set log path. if empty, don't log to file.
-  path: /var/log/nginx-agent/
-
-nginx:
-  # path of NGINX logs to exclude
-  exclude_logs: ""
-  socket: ""
 
 dataplane:
   status:
     # poll interval for data plane status - the frequency the agent will query the dataplane for changes
     poll_interval: 30s
-    # report interval for data plane status - the maximum duration to wait before syncing dataplane information if no updates have being observed
+    # report interval for data plane status - the maximum duration to wait before syncing dataplane information if
+    # no updates have being observed
     report_interval: 24h
 
 metrics:
   # specify the size of a buffer to build before sending metrics
   bulk_size: 20
-  # setting the report interval to 1 year because our control plane doesn't implement the metrics server
+  # setting the report and collection interval to 1 year because our control plane doesn't implement the metrics server
   # so the agent spams the logs with error messages and retries. Ideally, we'll be able to disable the metrics
   # client altogether in the future.
   report_interval: 8760h
-  collection_interval: 15s
+  collection_interval: 8760h
   mode: aggregated
 
 # OSS NGINX default config path
 # path to aux file dirs can also be added
 config_dirs: "/etc/nginx"
 
-api:
-  # default port for Agent API, this is for the server configuration of the REST API
-  port: 8081
-
 server:
-  host: 127.0.0.1
+  host: 127.0.0.1 # change to nginx-gateway.nginx-gateway if testing agent in separate deployment
   grpcPort: 54789
 
 # TLS is temporarily disabled. Once we fully separate the data plane from the control plane TLS will be enabled.

build/nginx-with-agent/nginx-with-agent.yaml

@@ -14,7 +14,6 @@ spec:
         app: nginx-with-agent
     spec:
       serviceAccountName: default
-      automountServiceAccountToken: false
       securityContext:
         sysctls:
         - name: "net.ipv4.ip_unprivileged_port_start"

deploy/manifests/nginx-gateway.yaml

@@ -86,6 +86,10 @@ spec:
     spec:
       shareProcessNamespace: true
       serviceAccountName: nginx-gateway
+      securityContext:
+        sysctls:
+        - name: "net.ipv4.ip_unprivileged_port_start"
+          value: "0"
       volumes:
       - name: nginx-config
         emptyDir: { }
@@ -102,9 +106,12 @@ spec:
         - name: nginx-config
           mountPath: /etc/nginx
       containers:
-      - image: ghcr.io/nginxinc/nginx-kubernetes-gateway:edge
-        imagePullPolicy: Always
+      - image: docker.io/nginx-kubernetes-gateway:edge # FIXME(kate-osborn): change back to ghcr before merging to main
+        imagePullPolicy: IfNotPresent # FIXME(kate-osborn): change back to Always before merging to main
         name: nginx-gateway
+        ports:
+        - name: grpc
+          containerPort: 54789
         volumeMounts:
         - name: nginx-config
           mountPath: /etc/nginx
@@ -142,6 +149,6 @@ spec:
             - ALL
         ports:
         - name: http
-          containerPort: 8080
+          containerPort: 80
         - name: https
-          containerPort: 8443
+          containerPort: 443

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.9.0
 	github.com/onsi/gomega v1.27.2
 	github.com/spf13/pflag v1.0.5
+	golang.org/x/sync v0.1.0
 	google.golang.org/grpc v1.52.0
 	k8s.io/api v0.26.2
 	k8s.io/apimachinery v0.26.2

go.sum

@@ -429,6 +429,7 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

internal/agent/agent_suite_test.go

@@ -0,0 +1,13 @@
+package agent_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestAgent(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Agent Suite")
+}

internal/agent/doc.go

@@ -0,0 +1,7 @@
+/*
+Package agent contains objects and methods for interacting with agents.
+
+The package includes:
+- Pool: A concurrent-safe connection pool for managing commander.Agent.
+*/
+package agent

internal/agent/pool.go

@@ -0,0 +1,60 @@
+package agent
+
+import (
+	"sync"
+
+	"github.com/go-logr/logr"
+
+	"github.com/nginxinc/nginx-kubernetes-gateway/internal/grpc/commander"
+)
+
+// Pool is a concurrent safe pool of commander.Agents.
+type Pool struct {
+	agents map[string]commander.Agent
+	logger logr.Logger
+
+	lock sync.Mutex
+}
+
+// NewPool returns a new instance of Pool.
+func NewPool(logger logr.Logger) *Pool {
+	return &Pool{
+		agents: make(map[string]commander.Agent),
+		logger: logger,
+	}
+}
+
+// AddAgent adds an agent to the Pool.
+func (ap *Pool) AddAgent(agent commander.Agent) {
+	ap.lock.Lock()
+	defer ap.lock.Unlock()
+
+	ap.agents[agent.ID()] = agent
+
+	ap.logger.Info("Added new agent", "id", agent.ID(), "total number of agents", len(ap.agents))
+}
+
+// RemoveAgent removes an agent from the Pool with the given ID.
+func (ap *Pool) RemoveAgent(id string) {
+	ap.lock.Lock()
+	defer ap.lock.Unlock()
+
+	delete(ap.agents, id)
+	ap.logger.Info("Removed agent", "id", id, "total number of agents", len(ap.agents))
+}
+
+// GetAgent returns the agent with the given ID from the Pool.
+func (ap *Pool) GetAgent(id string) commander.Agent {
+	ap.lock.Lock()
+	defer ap.lock.Unlock()
+
+	return ap.agents[id]
+}
+
+// Size is used for testing purposes.
+func (ap *Pool) Size() int {
+	ap.lock.Lock()
+	defer ap.lock.Unlock()
+
+	return len(ap.agents)
+}

internal/agent/pool_test.go

@@ -0,0 +1,110 @@
+package agent_test
+
+import (
+	"fmt"
+	"sync"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	"github.com/nginxinc/nginx-kubernetes-gateway/internal/agent"
+	"github.com/nginxinc/nginx-kubernetes-gateway/internal/grpc/commander"
+	"github.com/nginxinc/nginx-kubernetes-gateway/internal/grpc/commander/commanderfakes"
+)
+
+func newFakeAgent(id string) commander.Agent {
+	return &commanderfakes.FakeAgent{
+		IDStub: func() string {
+			return id
+		},
+	}
+}
+
+var _ = Describe("Agent Pool", func() {
+	var (
+		pool                   *agent.Pool
+		agent1, agent2, agent3 commander.Agent
+	)
+
+	BeforeEach(func() {
+		pool = agent.NewPool(zap.New())
+		agent1, agent2, agent3 = newFakeAgent("1"), newFakeAgent("2"), newFakeAgent("3")
+	})
+
+	It("can add and get agents", func() {
+		pool.AddAgent(agent1)
+		Expect(pool.Size()).To(Equal(1))
+		Expect(pool.GetAgent("1")).To(Equal(agent1))
+
+		pool.AddAgent(agent2)
+		Expect(pool.Size()).To(Equal(2))
+		Expect(pool.GetAgent("2")).To(Equal(agent2))
+
+		pool.AddAgent(agent3)
+		Expect(pool.Size()).To(Equal(3))
+		Expect(pool.GetAgent("3")).To(Equal(agent3))
+	})
+	It("can remove agents", func() {
+		pool.AddAgent(agent1)
+		pool.AddAgent(agent2)
+		pool.AddAgent(agent3)
+		Expect(pool.Size()).To(Equal(3))
+
+		pool.RemoveAgent("2")
+		Expect(pool.Size()).To(Equal(2))
+		Expect(pool.GetAgent("1")).To(Equal(agent1))
+		Expect(pool.GetAgent("3")).To(Equal(agent3))
+
+		pool.RemoveAgent("1")
+		Expect(pool.Size()).To(Equal(1))
+		Expect(pool.GetAgent("3")).To(Equal(agent3))
+
+		pool.RemoveAgent("3")
+		Expect(pool.Size()).To(Equal(0))
+	})
+	When("an agent does not exist in pool", func() {
+		It("remove agent does nothing", func() {
+			Expect(pool.Size()).To(Equal(0))
+			pool.RemoveAgent("dne")
+			Expect(pool.Size()).To(Equal(0))
+		})
+	})
+	It("can handle concurrent CRUD", func() {
+		// populate pool with 5 agents which will be removed.
+		for i := 1; i <= 5; i++ {
+			pool.AddAgent(newFakeAgent(fmt.Sprintf("%d", i)))
+		}
+
+		addAndGetAgent := func(id string, wg *sync.WaitGroup) {
+			defer wg.Done()
+			pool.AddAgent(newFakeAgent(id))
+			Expect(pool.GetAgent(id).ID()).To(Equal(id))
+		}
+
+		removeAndGetAgent := func(id string, wg *sync.WaitGroup) {
+			defer wg.Done()
+			Expect(pool.GetAgent(id).ID()).To(Equal(id))
+			pool.RemoveAgent(id)
+			Expect(pool.GetAgent(id)).To(BeNil())
+		}
+
+		wg := &sync.WaitGroup{}
+		for i := 0; i < 15; i++ {
+			id := fmt.Sprintf("%d", i+1)
+
+			wg.Add(1)
+			// remove first five
+			if i < 5 {
+				go removeAndGetAgent(id, wg)
+			} else {
+				// add 10 new
+				go addAndGetAgent(id, wg)
+			}
+		}
+
+		wg.Wait()
+
+		Expect(pool.Size()).To(Equal(10))
+	})
+})

internal/events/handler_test.go

@@ -103,7 +103,7 @@ var _ = Describe("EventHandler", func() {
 
 				batch := []interface{}{e}
 
-				handler.HandleEventBatch(context.TODO(), batch)
+				handler.HandleEventBatch(context.Background(), batch)
 
 				// Check that the events were captured
 				switch typedEvent := e.(type) {
@@ -195,7 +195,7 @@ var _ = Describe("EventHandler", func() {
 				},
 			}
 
-			handler.HandleEventBatch(context.TODO(), batch)
+			handler.HandleEventBatch(context.Background(), batch)
 
 			Expect(fakeSecretStore.UpsertCallCount()).Should(Equal(1))
 			Expect(fakeSecretStore.UpsertArgsForCall(0)).Should(Equal(secret))
@@ -213,7 +213,7 @@ var _ = Describe("EventHandler", func() {
 				},
 			}
 
-			handler.HandleEventBatch(context.TODO(), batch)
+			handler.HandleEventBatch(context.Background(), batch)
 
 			Expect(fakeSecretStore.DeleteCallCount()).Should(Equal(1))
 			Expect(fakeSecretStore.DeleteArgsForCall(0)).Should(Equal(nsname))
@@ -266,7 +266,7 @@ var _ = Describe("EventHandler", func() {
 		fakeCfg := []byte("fake")
 		fakeGenerator.GenerateReturns(fakeCfg)
 
-		handler.HandleEventBatch(context.TODO(), batch)
+		handler.HandleEventBatch(context.Background(), batch)
 
 		// Check that the events for Gateway API resources were captured
 
@@ -302,7 +302,7 @@ var _ = Describe("EventHandler", func() {
 			func(e interface{}) {
 				handle := func() {
 					batch := []interface{}{e}
-					handler.HandleEventBatch(context.TODO(), batch)
+					handler.HandleEventBatch(context.Background(), batch)
 				}
 
 				Expect(handle).Should(Panic())

internal/grpc/commander/agent.go

@@ -0,0 +1,24 @@
+package commander
+
+//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . AgentManager
+//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . Agent
+
+// Agent represents a connected Agent.
+// This interface is used for testing purposes because it allows easy mocking of an agent.
+type Agent interface {
+	// ID returns the unique ID of the Agent.
+	ID() string
+	// State returns the State of the Agent.
+	State() State
+}
+
+// AgentManager manages all the connected agents.
+// The commander uses the AgentManager to track all the connected Agents.
+type AgentManager interface {
+	// AddAgent adds an Agent to the manager.
+	AddAgent(agent Agent)
+	// RemoveAgent removes the Agent with the provided ID from the manager.
+	RemoveAgent(id string)
+	// GetAgent returns the Agent with the provided ID.
+	GetAgent(id string) Agent
+}