REORG/MAJOR: Extract k8s dependency from controller struct

Author: Mo3m3n

main.go

@@ -19,15 +19,20 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
 	//nolint:gosec
 	_ "net/http/pprof"
 
+	"k8s.io/apimachinery/pkg/watch"
+
 	"github.com/jessevdk/go-flags"
 
-	c "github.com/haproxytech/kubernetes-ingress/pkg/controller"
+	"github.com/haproxytech/kubernetes-ingress/pkg/controller"
 	"github.com/haproxytech/kubernetes-ingress/pkg/controller/annotations"
+	"github.com/haproxytech/kubernetes-ingress/pkg/ingress"
+	"github.com/haproxytech/kubernetes-ingress/pkg/k8s"
 	"github.com/haproxytech/kubernetes-ingress/pkg/store"
 	"github.com/haproxytech/kubernetes-ingress/pkg/utils"
 )
@@ -72,21 +77,43 @@ func main() {
 	annotations.SetDefaultValue("ssl-certificate", defaultCertificate)
 
 	// Start Controller
-	s := store.NewK8sStore(osArgs)
-	for _, namespace := range osArgs.NamespaceWhitelist {
-		s.NamespacesAccess.Whitelist[namespace] = struct{}{}
+	var chanSize int64 = int64(watch.DefaultChanSize * 6)
+	if osArgs.ChannelSize > 0 {
+		chanSize = osArgs.ChannelSize
 	}
-	for _, namespace := range osArgs.NamespaceBlacklist {
-		s.NamespacesAccess.Blacklist[namespace] = struct{}{}
+	eventChan := make(chan k8s.SyncDataEvent, chanSize)
+	ingressChan := make(chan ingress.Sync, chanSize)
+	stop := make(chan struct{})
+
+	publishService := getNamespaceValue(osArgs.PublishService)
+
+	s := store.NewK8sStore(osArgs)
+
+	c := controller.NewBuilder().
+		WithEventChan(eventChan).
+		WithIngressChan(ingressChan).
+		WithStore(s).
+		WithPublishService(publishService).
+		WithArgs(osArgs).Build()
+
+	k := k8s.New(
+		osArgs,
+		s.NamespacesAccess.Whitelist,
+		publishService,
+	)
+
+	go k.MonitorChanges(eventChan, ingressChan, stop)
+	go c.Start(haproxyConf)
+	if publishService != nil {
+		go ingress.UpdateStatus(k.GetClientset(), s, osArgs.IngressClass, osArgs.EmptyIngressClass, ingressChan)
 	}
-	controller := c.NewBuilder().WithStore(s).WithArgs(osArgs).Build()
-	controller.Start(haproxyConf)
 
 	// Catch QUIT signals
 	signalC := make(chan os.Signal, 1)
 	signal.Notify(signalC, os.Interrupt, syscall.SIGTERM, syscall.SIGUSR1)
 	<-signalC
-	controller.Stop()
+	c.Stop()
+	close(stop)
 }
 
 func logInfo(logger utils.Logger, osArgs utils.OSArgs) {
@@ -146,3 +173,15 @@ func logInfo(logger utils.Logger, osArgs utils.OSArgs) {
 	logger.Error(err)
 	logger.Printf("Running on %s", hostname)
 }
+
+func getNamespaceValue(name string) *utils.NamespaceValue {
+	parts := strings.Split(name, "/")
+	var result *utils.NamespaceValue
+	if len(parts) == 2 {
+		result = &utils.NamespaceValue{
+			Namespace: parts[0],
+			Name:      parts[1],
+		}
+	}
+	return result
+}

pkg/controller/builder.go

@@ -10,14 +10,19 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	config "github.com/haproxytech/kubernetes-ingress/pkg/configuration"
+	"github.com/haproxytech/kubernetes-ingress/pkg/ingress"
+	"github.com/haproxytech/kubernetes-ingress/pkg/k8s"
 	"github.com/haproxytech/kubernetes-ingress/pkg/store"
 	"github.com/haproxytech/kubernetes-ingress/pkg/utils"
 )
 
 type Builder struct {
-	osArgs utils.OSArgs
-	cfg    config.ControllerCfg
-	store  store.K8s
+	osArgs         utils.OSArgs
+	cfg            config.ControllerCfg
+	store          store.K8s
+	publishService *utils.NamespaceValue
+	eventChan      chan k8s.SyncDataEvent
+	ingressChan    chan ingress.Sync
 }
 
 var defaultCfg = config.ControllerCfg{
@@ -36,6 +41,16 @@ func NewBuilder() *Builder {
 	}
 }
 
+func (builder *Builder) WithEventChan(eventChan chan k8s.SyncDataEvent) *Builder {
+	builder.eventChan = eventChan
+	return builder
+}
+
+func (builder *Builder) WithIngressChan(ingressChan chan ingress.Sync) *Builder {
+	builder.ingressChan = ingressChan
+	return builder
+}
+
 func (builder *Builder) WithStore(store store.K8s) *Builder {
 	builder.store = store
 	return builder
@@ -51,6 +66,11 @@ func (builder *Builder) WithArgs(osArgs utils.OSArgs) *Builder {
 	return builder
 }
 
+func (builder *Builder) WithPublishService(publishService *utils.NamespaceValue) *Builder {
+	builder.publishService = publishService
+	return builder
+}
+
 func (builder *Builder) Build() *HAProxyController {
 	if builder.osArgs.External {
 		builder.cfg = setupExternalMode(builder.osArgs)
@@ -72,13 +92,15 @@ func (builder *Builder) Build() *HAProxyController {
 
 	prefix, errPrefix := utils.GetPodPrefix(os.Getenv("POD_NAME"))
 	logger.Error(errPrefix)
-
 	return &HAProxyController{
-		cfg:          builder.cfg,
-		osArgs:       builder.osArgs,
-		podNamespace: os.Getenv("POD_NAMESPACE"),
-		podPrefix:    prefix,
-		store:        builder.store,
+		cfg:            builder.cfg,
+		osArgs:         builder.osArgs,
+		podNamespace:   os.Getenv("POD_NAMESPACE"),
+		podPrefix:      prefix,
+		store:          builder.store,
+		eventChan:      builder.eventChan,
+		ingressChan:    builder.ingressChan,
+		publishService: builder.publishService,
 	}
 }
 

pkg/controller/controller.go

@@ -16,11 +16,6 @@ package controller
 
 import (
 	"os"
-	"path/filepath"
-	"strings"
-
-	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/google/renameio"
 
@@ -33,7 +28,6 @@ import (
 	"github.com/haproxytech/kubernetes-ingress/pkg/k8s"
 	"github.com/haproxytech/kubernetes-ingress/pkg/store"
 	"github.com/haproxytech/kubernetes-ingress/pkg/utils"
-	"k8s.io/apimachinery/pkg/watch"
 )
 
 var logger = utils.GetLogger()
@@ -48,7 +42,6 @@ type HAProxyController struct {
 	auxCfgModTime  int64
 	eventChan      chan k8s.SyncDataEvent
 	ingressChan    chan ingress.Sync
-	k8s            k8s.K8s
 	ready          bool
 	reload         bool
 	restart        bool
@@ -101,49 +94,7 @@ func (c *HAProxyController) Start(haproxyConf []byte) {
 	c.initHandlers()
 	c.haproxyStartup()
 
-	// Controller PublishService
-	parts := strings.Split(c.osArgs.PublishService, "/")
-	if len(parts) == 2 {
-		c.publishService = &utils.NamespaceValue{
-			Namespace: parts[0],
-			Name:      parts[1],
-		}
-	}
-
-	// Get K8s client
-	var restConfig *rest.Config
-	if c.osArgs.External {
-		kubeconfig := filepath.Join(utils.HomeDir(), ".kube", "config")
-		if c.osArgs.KubeConfig != "" {
-			kubeconfig = c.osArgs.KubeConfig
-		}
-		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
-	} else {
-		restConfig, err = rest.InClusterConfig()
-	}
-	logger.Panicf("Unable to get kubernetes client config: %s", err)
-
-	c.k8s = k8s.New(restConfig, c.osArgs, c.eventChan)
-	x := c.k8s.GetClient().Discovery()
-	if k8sVersion, err := x.ServerVersion(); err != nil {
-		logger.Panicf("Unable to get Kubernetes version: %v\n", err)
-	} else {
-		logger.Printf("Running on Kubernetes version: %s %s", k8sVersion.String(), k8sVersion.Platform)
-	}
-
-	// Monitor k8s events
-	var chanSize int64 = int64(watch.DefaultChanSize * 6)
-	if c.osArgs.ChannelSize > 0 {
-		chanSize = c.osArgs.ChannelSize
-	}
-	logger.Infof("Channel size: %d", chanSize)
-	c.eventChan = make(chan k8s.SyncDataEvent, chanSize)
-	go c.monitorChanges()
-	if c.publishService != nil {
-		// Update Ingress status
-		c.ingressChan = make(chan ingress.Sync, chanSize)
-		go ingress.UpdateStatus(c.k8s.GetClient(), c.store, c.osArgs.IngressClass, c.osArgs.EmptyIngressClass, c.ingressChan)
-	}
+	c.SyncData()
 }
 
 // Stop handles shutting down HAProxyController