Merge pull request #15 from nginxinc/updated-tests

Improve test coverage

Author: chrisakker

cmd/nginx-k8s-edge-controller/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/util/workqueue"
 )
 
 func main() {
@@ -42,20 +43,24 @@ func run() error {
 		return fmt.Errorf(`error occurred initializing settings: %w`, err)
 	}
 
-	synchronizer, err := synchronization.NewSynchronizer(settings)
+	synchronizerWorkqueue, err := buildWorkQueue(settings.Synchronizer.WorkQueueSettings)
 	if err != nil {
-		return fmt.Errorf(`error initializing synchronizer: %w`, err)
+		return fmt.Errorf(`error occurred building a workqueue: %w`, err)
 	}
 
-	err = synchronizer.Initialize()
+	synchronizer, err := synchronization.NewSynchronizer(settings, synchronizerWorkqueue)
 	if err != nil {
 		return fmt.Errorf(`error initializing synchronizer: %w`, err)
 	}
 
-	handler := observation.NewHandler(synchronizer)
-	handler.Initialize()
+	handlerWorkqueue, err := buildWorkQueue(settings.Synchronizer.WorkQueueSettings)
+	if err != nil {
+		return fmt.Errorf(`error occurred building a workqueue: %w`, err)
+	}
+
+	handler := observation.NewHandler(settings, synchronizer, handlerWorkqueue)
 
-	watcher, err := observation.NewWatcher(ctx, handler, k8sClient)
+	watcher, err := observation.NewWatcher(settings, handler)
 	if err != nil {
 		return fmt.Errorf(`error occurred creating a watcher: %w`, err)
 	}
@@ -97,3 +102,10 @@ func buildKubernetesClient() (*kubernetes.Clientset, error) {
 
 	return client, nil
 }
+
+func buildWorkQueue(settings configuration.WorkQueueSettings) (workqueue.RateLimitingInterface, error) {
+	logrus.Debug("Watcher::buildSynchronizerWorkQueue")
+
+	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(settings.RateLimiterBase, settings.RateLimiterMax)
+	return workqueue.NewNamedRateLimitingQueue(rateLimiter, settings.Name), nil
+}

internal/configuration/settings.go

@@ -14,28 +14,82 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"strings"
+	"time"
 )
 
 const (
 	ConfigMapsNamespace = "nkl"
 	ResyncPeriod        = 0
 )
 
+type WorkQueueSettings struct {
+	Name            string
+	RateLimiterBase time.Duration
+	RateLimiterMax  time.Duration
+}
+
+type HandlerSettings struct {
+	RetryCount        int
+	Threads           int
+	WorkQueueSettings WorkQueueSettings
+}
+
+type WatcherSettings struct {
+	NginxIngressNamespace string
+	ResyncPeriod          time.Duration
+}
+
+type SynchronizerSettings struct {
+	MaxMillisecondsJitter int
+	MinMillisecondsJitter int
+	RetryCount            int
+	Threads               int
+	WorkQueueSettings     WorkQueueSettings
+}
+
 type Settings struct {
-	ctx                      context.Context
+	Context                  context.Context
 	NginxPlusHosts           []string
-	k8sClient                *kubernetes.Clientset
+	K8sClient                *kubernetes.Clientset
 	informer                 cache.SharedInformer
 	eventHandlerRegistration cache.ResourceEventHandlerRegistration
+
+	Handler      HandlerSettings
+	Synchronizer SynchronizerSettings
+	Watcher      WatcherSettings
 }
 
 func NewSettings(ctx context.Context, k8sClient *kubernetes.Clientset) (*Settings, error) {
-	config := new(Settings)
-
-	config.k8sClient = k8sClient
-	config.ctx = ctx
+	settings := &Settings{
+		Context:   ctx,
+		K8sClient: k8sClient,
+		Handler: HandlerSettings{
+			RetryCount: 5,
+			Threads:    1,
+			WorkQueueSettings: WorkQueueSettings{
+				RateLimiterBase: time.Second * 2,
+				RateLimiterMax:  time.Second * 60,
+				Name:            "nkl-handler",
+			},
+		},
+		Synchronizer: SynchronizerSettings{
+			MaxMillisecondsJitter: 750,
+			MinMillisecondsJitter: 250,
+			RetryCount:            5,
+			Threads:               1,
+			WorkQueueSettings: WorkQueueSettings{
+				RateLimiterBase: time.Second * 2,
+				RateLimiterMax:  time.Second * 60,
+				Name:            "nkl-synchronizer",
+			},
+		},
+		Watcher: WatcherSettings{
+			NginxIngressNamespace: "nginx-ingress",
+			ResyncPeriod:          0,
+		},
+	}
 
-	return config, nil
+	return settings, nil
 }
 
 func (s *Settings) Initialize() error {
@@ -63,14 +117,14 @@ func (s *Settings) Run() {
 
 	defer utilruntime.HandleCrash()
 
-	go s.informer.Run(s.ctx.Done())
+	go s.informer.Run(s.Context.Done())
 
-	<-s.ctx.Done()
+	<-s.Context.Done()
 }
 
 func (s *Settings) buildInformer() (cache.SharedInformer, error) {
 	options := informers.WithNamespace(ConfigMapsNamespace)
-	factory := informers.NewSharedInformerFactoryWithOptions(s.k8sClient, ResyncPeriod, options)
+	factory := informers.NewSharedInformerFactoryWithOptions(s.K8sClient, ResyncPeriod, options)
 	informer := factory.Core().V1().ConfigMaps().Informer()
 
 	return informer, nil

internal/core/events_test.go

@@ -0,0 +1,60 @@
+package core
+
+import (
+	nginxClient "github.com/nginxinc/nginx-plus-go-client/client"
+	"testing"
+)
+
+func TestServerUpdateEventWithIdAndHost(t *testing.T) {
+	event := NewServerUpdateEvent(Created, "upstream", []nginxClient.StreamUpstreamServer{})
+
+	if event.Id != "" {
+		t.Errorf("expected empty Id, got %s", event.Id)
+	}
+
+	if event.NginxHost != "" {
+		t.Errorf("expected empty NginxHost, got %s", event.NginxHost)
+	}
+
+	eventWithIdAndHost := ServerUpdateEventWithIdAndHost(event, "id", "host")
+
+	if eventWithIdAndHost.Id != "id" {
+		t.Errorf("expected Id to be 'id', got %s", eventWithIdAndHost.Id)
+	}
+
+	if eventWithIdAndHost.NginxHost != "host" {
+		t.Errorf("expected NginxHost to be 'host', got %s", eventWithIdAndHost.NginxHost)
+	}
+}
+
+func TestTypeNameCreated(t *testing.T) {
+	event := NewServerUpdateEvent(Created, "upstream", []nginxClient.StreamUpstreamServer{})
+
+	if event.TypeName() != "Created" {
+		t.Errorf("expected 'Created', got %s", event.TypeName())
+	}
+}
+
+func TestTypeNameUpdated(t *testing.T) {
+	event := NewServerUpdateEvent(Updated, "upstream", []nginxClient.StreamUpstreamServer{})
+
+	if event.TypeName() != "Updated" {
+		t.Errorf("expected 'Updated', got %s", event.TypeName())
+	}
+}
+
+func TestTypeNameDeleted(t *testing.T) {
+	event := NewServerUpdateEvent(Deleted, "upstream", []nginxClient.StreamUpstreamServer{})
+
+	if event.TypeName() != "Deleted" {
+		t.Errorf("expected 'Deleted', got %s", event.TypeName())
+	}
+}
+
+func TestTypeNameUnknown(t *testing.T) {
+	event := NewServerUpdateEvent(EventType(100), "upstream", []nginxClient.StreamUpstreamServer{})
+
+	if event.TypeName() != "Unknown" {
+		t.Errorf("expected 'Unknown', got %s", event.TypeName())
+	}
+}

internal/observation/handler.go

@@ -6,28 +6,31 @@ package observation
 
 import (
 	"fmt"
+	"github.com/nginxinc/kubernetes-nginx-ingress/internal/configuration"
 	"github.com/nginxinc/kubernetes-nginx-ingress/internal/core"
 	"github.com/nginxinc/kubernetes-nginx-ingress/internal/synchronization"
 	"github.com/nginxinc/kubernetes-nginx-ingress/internal/translation"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/workqueue"
-	"time"
 )
 
-const RateLimiterBase = time.Second * 2
-const RateLimiterMax = time.Second * 60
-const RetryCount = 5
-const Threads = 1
-const WatcherQueueName = `nkl-handler`
+type HandlerInterface interface {
+	AddRateLimitedEvent(event *core.Event)
+	Run(stopCh <-chan struct{})
+	ShutDown()
+}
 
 type Handler struct {
 	eventQueue   workqueue.RateLimitingInterface
-	synchronizer *synchronization.Synchronizer
+	settings     *configuration.Settings
+	synchronizer synchronization.Interface
 }
 
-func NewHandler(synchronizer *synchronization.Synchronizer) *Handler {
+func NewHandler(settings *configuration.Settings, synchronizer synchronization.Interface, eventQueue workqueue.RateLimitingInterface) *Handler {
 	return &Handler{
+		eventQueue:   eventQueue,
+		settings:     settings,
 		synchronizer: synchronizer,
 	}
 }
@@ -37,15 +40,10 @@ func (h *Handler) AddRateLimitedEvent(event *core.Event) {
 	h.eventQueue.AddRateLimited(event)
 }
 
-func (h *Handler) Initialize() {
-	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(RateLimiterBase, RateLimiterMax)
-	h.eventQueue = workqueue.NewNamedRateLimitingQueue(rateLimiter, WatcherQueueName)
-}
-
 func (h *Handler) Run(stopCh <-chan struct{}) {
 	logrus.Debug("Handler::Run")
 
-	for i := 0; i < Threads; i++ {
+	for i := 0; i < h.settings.Handler.Threads; i++ {
 		go wait.Until(h.worker, 0, stopCh)
 	}
 
@@ -97,7 +95,7 @@ func (h *Handler) withRetry(err error, event *core.Event) {
 	logrus.Debug("Handler::withRetry")
 	if err != nil {
 		// TODO: Add Telemetry
-		if h.eventQueue.NumRequeues(event) < RetryCount { // TODO: Make this configurable
+		if h.eventQueue.NumRequeues(event) < h.settings.Handler.RetryCount {
 			h.eventQueue.AddRateLimited(event)
 			logrus.Infof(`Handler::withRetry: requeued event: %#v; error: %v`, event, err)
 		} else {

internal/observation/handler_test.go

@@ -0,0 +1,54 @@
+package observation
+
+import (
+	"context"
+	"fmt"
+	"github.com/nginxinc/kubernetes-nginx-ingress/internal/configuration"
+	"github.com/nginxinc/kubernetes-nginx-ingress/internal/core"
+	"github.com/nginxinc/kubernetes-nginx-ingress/test/mocks"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/util/workqueue"
+	"testing"
+)
+
+func TestHandler_AddsEventToSynchronizer(t *testing.T) {
+	_, _, synchronizer, handler, err := buildHandler()
+	if err != nil {
+		t.Errorf(`should have been no error, %v`, err)
+	}
+
+	event := &core.Event{
+		Type: core.Created,
+		Service: &v1.Service{
+			Spec: v1.ServiceSpec{
+				Ports: []v1.ServicePort{
+					{
+						Name: "nkl-back",
+					},
+				},
+			},
+		},
+	}
+
+	handler.AddRateLimitedEvent(event)
+
+	handler.handleNextEvent()
+
+	if len(synchronizer.Events) != 1 {
+		t.Errorf(`handler.AddRateLimitedEvent did not add the event to the queue`)
+	}
+}
+
+func buildHandler() (*configuration.Settings, workqueue.RateLimitingInterface, *mocks.MockSynchronizer, *Handler, error) {
+	settings, err := configuration.NewSettings(context.Background(), nil)
+	if err != nil {
+		return nil, nil, nil, nil, fmt.Errorf(`should have been no error, %v`, err)
+	}
+
+	eventQueue := &mocks.MockRateLimiter{}
+	synchronizer := &mocks.MockSynchronizer{}
+
+	handler := NewHandler(settings, synchronizer, eventQueue)
+
+	return settings, eventQueue, synchronizer, handler, nil
+}