Initialize watcher and cache during DLX creation

Author: weilerN

pkg/dlx/dlx.go

@@ -24,21 +24,28 @@ import (
 	"context"
 	"net/http"
 
+	"github.com/v3io/scaler/pkg/ingresscache"
+	"github.com/v3io/scaler/pkg/kube"
 	"github.com/v3io/scaler/pkg/scalertypes"
 
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
+	"k8s.io/client-go/kubernetes"
 )
 
 type DLX struct {
-	logger  logger.Logger
-	handler Handler
-	server  *http.Server
+	logger        logger.Logger
+	handler       Handler
+	server        *http.Server
+	cache         ingresscache.IngressHostCache
+	watchOperator *kube.IngressWatcher
 }
 
 func NewDLX(parentLogger logger.Logger,
 	resourceScaler scalertypes.ResourceScaler,
-	options scalertypes.DLXOptions) (*DLX, error) {
+	options scalertypes.DLXOptions,
+	kubeClientSet kubernetes.Interface,
+) (*DLX, error) {
 	childLogger := parentLogger.GetChild("dlx")
 	childLogger.InfoWith("Creating DLX",
 		"options", options)
@@ -50,34 +57,57 @@ func NewDLX(parentLogger logger.Logger,
 		return nil, errors.Wrap(err, "Failed to create function starter")
 	}
 
+	cache := ingresscache.NewIngressCache(childLogger)
 	handler, err := NewHandler(childLogger,
 		resourceStarter,
 		resourceScaler,
 		options.TargetNameHeader,
 		options.TargetPathHeader,
 		options.TargetPort,
-		options.MultiTargetStrategy)
+		options.MultiTargetStrategy,
+		cache)
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to create handler")
 	}
 
+	watcherOperator, err := kube.NewIngressWatcher(
+		context.Background(),
+		childLogger,
+		kubeClientSet,
+		cache,
+		options.ResolveTargetsFromIngressCallback,
+		options.ResyncInterval,
+		options.Namespace,
+		options.LabelSelector,
+	)
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to create ingress watcher")
+	}
+
 	return &DLX{
 		logger:  childLogger,
 		handler: handler,
 		server: &http.Server{
 			Addr: options.ListenAddress,
 		},
+		cache:         cache,
+		watchOperator: watcherOperator,
 	}, nil
 }
 
 func (d *DLX) Start() error {
 	d.logger.DebugWith("Starting", "server", d.server.Addr)
+	// watchOperator.Start() is non-blocking and runs informers in background goroutines
+	if err := d.watchOperator.Start(); err != nil {
+		return errors.Wrap(err, "Failed to start ingress watcher")
+	}
 	http.HandleFunc("/", d.handler.HandleFunc)
 	go d.server.ListenAndServe() // nolint: errcheck
 	return nil
 }
 
 func (d *DLX) Stop(context context.Context) error {
 	d.logger.DebugWith("Stopping", "server", d.server.Addr)
+	d.watchOperator.Stop()
 	return d.server.Shutdown(context)
 }

pkg/scalertypes/types.go

@@ -29,6 +29,8 @@ import (
 
 	"github.com/nuclio/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/v3io/scaler/pkg/kube"
 )
 
 type AutoScalerOptions struct {
@@ -55,12 +57,15 @@ type DLXOptions struct {
 	Namespace string
 
 	// comma delimited
-	TargetNameHeader         string
-	TargetPathHeader         string
-	TargetPort               int
-	ListenAddress            string
-	ResourceReadinessTimeout Duration
-	MultiTargetStrategy      MultiTargetStrategy
+	TargetNameHeader                  string
+	TargetPathHeader                  string
+	TargetPort                        int
+	ListenAddress                     string
+	ResourceReadinessTimeout          Duration
+	MultiTargetStrategy               MultiTargetStrategy
+	LabelSelector                     string
+	ResolveTargetsFromIngressCallback kube.ResolveTargetsFromIngressCallback
+	ResyncInterval                    time.Duration
 }
 
 type ResourceScaler interface {