Switch from OF client to OF lister interface

**What**
- Use the informer/lister to get Profiles instead of the REST client,
  this should be more efficient and low impact on the k8s API because it
  maintains an internal cache of the profiles

Signed-off-by: Lucas Roesler <[email protected]>

Author: LucasRoesler

k8s/factory.go

@@ -4,21 +4,27 @@
 package k8s
 
 import (
-	openfaasv1 "github.com/openfaas/faas-netes/pkg/client/clientset/versioned/typed/openfaas/v1"
+	v1 "github.com/openfaas/faas-netes/pkg/client/listers/openfaas/v1"
 	"k8s.io/client-go/kubernetes"
 )
 
+// NamespacedProfiler is a subset of the v1.ProfileLister that is needed for the function factory
+// to support Profiles
+type NamespacedProfiler interface {
+	Profiles(namespace string) v1.ProfileNamespaceLister
+}
+
 // FunctionFactory is handling Kubernetes operations to materialise functions into deployments and services
 type FunctionFactory struct {
 	Client   kubernetes.Interface
 	Config   DeploymentConfig
-	OFClient openfaasv1.OpenfaasV1Interface
+	Profiler NamespacedProfiler
 }
 
-func NewFunctionFactory(clientset kubernetes.Interface, config DeploymentConfig, ofClient openfaasv1.OpenfaasV1Interface) FunctionFactory {
+func NewFunctionFactory(clientset kubernetes.Interface, config DeploymentConfig, profiler NamespacedProfiler) FunctionFactory {
 	return FunctionFactory{
 		Client:   clientset,
 		Config:   config,
-		OFClient: ofClient,
+		Profiler: profiler,
 	}
 }

k8s/profiles.go

@@ -6,7 +6,6 @@ import (
 	"strings"
 
 	v1 "github.com/openfaas/faas-netes/pkg/apis/openfaas/v1"
-	openfaasv1 "github.com/openfaas/faas-netes/pkg/client/clientset/versioned/typed/openfaas/v1"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -53,7 +52,7 @@ func (c profileCMClient) Get(ctx context.Context, namespace string, names ...str
 
 // profileCRDClient implements PolicyClient using the openfaas CRD Profile
 type profileCRDClient struct {
-	client openfaasv1.ProfilesGetter
+	client NamespacedProfiler
 }
 
 func (c profileCRDClient) Get(ctx context.Context, namespace string, names ...string) ([]Profile, error) {
@@ -62,7 +61,8 @@ func (c profileCRDClient) Get(ctx context.Context, namespace string, names ...st
 		// this is where we would consider using an informer/lister. The throughput on this
 		// API. We expect will be similar to the secrets API, since we only use it during
 		// function Deploy
-		profile, err := c.client.Profiles(namespace).Get(ctx, name, metav1.GetOptions{})
+		// Note Lister interfaces do not have context yet
+		profile, err := c.client.Profiles(namespace).Get(name)
 		if err != nil {
 			return nil, err
 		}
@@ -74,7 +74,7 @@ func (c profileCRDClient) Get(ctx context.Context, namespace string, names ...st
 // NewProfileClient returns the ProfilerClient powered by the Profile CRD
 func (f FunctionFactory) NewProfileClient() ProfileClient {
 	// this is where we can replace with an informer/listener in the future
-	return &profileCRDClient{client: f.OFClient}
+	return &profileCRDClient{client: f.Profiler}
 }
 
 // NewProfileClient returns the ProfilerClient powered by ConfigMaps

main.go

@@ -94,8 +94,6 @@ func main() {
 
 	log.Printf("DeploymentConfig: %+v\n", deployConfig)
 
-	factory := k8s.NewFunctionFactory(kubeClient, deployConfig, faasClient.OpenfaasV1())
-
 	// the sync interval does not affect the scale to/from zero feature
 	// auto-scaling is does via the HTTP API that acts on the deployment Spec.Replicas
 	defaultResync := time.Minute * 5
@@ -106,13 +104,20 @@ func main() {
 	faasInformerOpt := informers.WithNamespace(config.DefaultFunctionNamespace)
 	faasInformerFactory := informers.NewSharedInformerFactoryWithOptions(faasClient, defaultResync, faasInformerOpt)
 
+	// this is where we need to swap to the faasInformerFactory
+	profileInformerOpt := informers.WithNamespace(config.ProfilesNamespace)
+	profileInformerFactory := informers.NewSharedInformerFactoryWithOptions(faasClient, defaultResync, profileInformerOpt)
+	profileLister := profileInformerFactory.Openfaas().V1().Profiles().Lister()
+	factory := k8s.NewFunctionFactory(kubeClient, deployConfig, profileLister)
+
 	setup := serverSetup{
-		config:              config,
-		functionFactory:     factory,
-		kubeInformerFactory: kubeInformerFactory,
-		faasInformerFactory: faasInformerFactory,
-		kubeClient:          kubeClient,
-		faasClient:          faasClient,
+		config:                 config,
+		functionFactory:        factory,
+		kubeInformerFactory:    kubeInformerFactory,
+		faasInformerFactory:    faasInformerFactory,
+		profileInformerFactory: profileInformerFactory,
+		kubeClient:             kubeClient,
+		faasClient:             faasClient,
 	}
 
 	if !operator {
@@ -142,9 +147,11 @@ func runController(setup serverSetup) {
 
 	log.Println("waiting for openfaas CRD cache sync")
 	faasInformerFactory.WaitForCacheSync(stopCh)
+	setup.profileInformerFactory.WaitForCacheSync(stopCh)
 	log.Println("cache sync complete")
 	go faasInformerFactory.Start(stopCh)
 	go kubeInformerFactory.Start(stopCh)
+	go setup.profileInformerFactory.Start(stopCh)
 
 	lister := endpointsInformer.Lister()
 	functionLookup := k8s.NewFunctionLookup(config.DefaultFunctionNamespace, lister)
@@ -190,6 +197,7 @@ func runOperator(setup serverSetup) {
 
 	glog.Infof("Waiting for cache sync in main")
 	kubeInformerFactory.WaitForCacheSync(stopCh)
+	setup.profileInformerFactory.WaitForCacheSync(stopCh)
 	glog.Infof("Cache sync done")
 
 	ctrl := controller.NewController(
@@ -204,6 +212,7 @@ func runOperator(setup serverSetup) {
 
 	go faasInformerFactory.Start(stopCh)
 	go kubeInformerFactory.Start(stopCh)
+	go setup.profileInformerFactory.Start(stopCh)
 
 	go srv.Start()
 	if err := ctrl.Run(1, stopCh); err != nil {
@@ -214,12 +223,13 @@ func runOperator(setup serverSetup) {
 // serverSetup is a container for the config and clients needed to start the
 // faas-netes controller or operator
 type serverSetup struct {
-	config              types.BootstrapConfig
-	kubeClient          *kubernetes.Clientset
-	faasClient          *clientset.Clientset
-	functionFactory     k8s.FunctionFactory
-	kubeInformerFactory kubeinformers.SharedInformerFactory
-	faasInformerFactory informers.SharedInformerFactory
+	config                 types.BootstrapConfig
+	kubeClient             *kubernetes.Clientset
+	faasClient             *clientset.Clientset
+	functionFactory        k8s.FunctionFactory
+	kubeInformerFactory    kubeinformers.SharedInformerFactory
+	faasInformerFactory    informers.SharedInformerFactory
+	profileInformerFactory informers.SharedInformerFactory
 }
 
 func setupLogging() {
@@ -231,7 +241,7 @@ func setupLogging() {
 		f2 := klogFlags.Lookup(f1.Name)
 		if f2 != nil {
 			value := f1.Value.String()
-			f2.Value.Set(value)
+			_ = f2.Value.Set(value)
 		}
 	})
 }

pkg/controller/deployment.go

@@ -137,7 +137,7 @@ func newDeployment(
 	if err != nil {
 		// TODO: a simple warning doesn't seem strong enough if a profile can't be found or there is
 		// some other error
-		glog.Warningf("Function %s can not retrieve required Profiles: %v", function.Spec.Name, err)
+		glog.Warningf("Function %s can not retrieve required Profiles in %s: %v", function.Spec.Name, profileNamespace, err)
 	}
 	for _, profile := range profileList {
 		factory.RemoveProfile(profile, deploymentSpec)
@@ -151,7 +151,7 @@ func newDeployment(
 	if err != nil {
 		// TODO: a simple warning doesn't seem strong enough if a profile can't be found or there is
 		// some other error
-		glog.Warningf("Function %s can not retrieve required Profiles: %v", function.Spec.Name, err)
+		glog.Warningf("Function %s can not retrieve required Profiles in %s: %v", function.Spec.Name, profileNamespace, err)
 	}
 	// TODO: remove this or refactor to just print names
 	glog.Infof("Function %s: Applying profiles %+v", function.Spec.Name, profileList)