Use Kubernetes informer and cache in router (#376)

Use Kubernetes client's informer and cache to watch functions and triggers in the router. This replaces the polling loop we were using so far.

glide.lock

@@ -1,5 +1,5 @@
-hash: b73c8c15207c8906bdc2c2713da03353d7c7ab78649f222b067e805b6f4c03dd
+hash: 285f5f29e18f61b5b834a48ce815fd40e023aa55d51acbc2ac8033cd8267635d
-updated: 2017-09-29T00:30:00.245121189+08:00
+updated: 2017-10-14T14:54:26.384349074+08:00
 imports:
 - name: cloud.google.com/go
   version: 3b1ae45394a234c385be014e9a488f2bb6eef821
@@ -87,6 +87,10 @@ imports:
   version: da285caa6daa337ae04a9ac603dbaf9009ffe687
   subpackages:
   - local
+- name: github.com/hashicorp/golang-lru
+  version: a0d98a5f288019575c6d1f4bb1573fef2d1fcdc4
+  subpackages:
+  - simplelru
 - name: github.com/howeyc/gopass
   version: bf9dde6d0d2c004a008c27aaee91170c786f6db8
 - name: github.com/imdario/mergo
@@ -207,6 +211,10 @@ imports:
   version: 3887ee99ecf07df5b447e9b00d9c0b2adaa9f3e4
 - name: gopkg.in/yaml.v2
   version: 53feefa2559fb8dfa8d81baad31be332c97d6c77
+- name: k8s.io/api
+  version: 4b8fc5be9b77d91bbb6525d18591c43699a2b4e5
+  subpackages:
+  - core/v1
 - name: k8s.io/apimachinery
   version: 1fd2e63a9a370677308a42f24fd40c86438afddf
   subpackages:
@@ -236,6 +244,7 @@ imports:
   - pkg/runtime/serializer/versioning
   - pkg/selection
   - pkg/types
+  - pkg/util/cache
   - pkg/util/clock
   - pkg/util/diff
   - pkg/util/errors
@@ -327,6 +336,7 @@ imports:
   - rest/watch
   - third_party/forked/golang/template
   - tools/auth
+  - tools/cache
   - tools/clientcmd
   - tools/clientcmd/api
   - tools/clientcmd/api/latest

glide.yaml

@@ -29,6 +29,10 @@ import:
   - pkg/labels
   - pkg/util/intstr
   - rest
+- package: k8s.io/api
+  version: 4b8fc5be9b77d91bbb6525d18591c43699a2b4e5
+- package: k8s.io/apimachinery
+  version: 1fd2e63a9a370677308a42f24fd40c86438afddf
 - package: github.com/influxdata/influxdb
   version: v1.2.0
   subpackages:

router/functionReferenceResolver.go

@@ -21,6 +21,11 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/rest"
+	k8sCache "k8s.io/client-go/tools/cache"
 
 	"github.com/fission/fission"
 	"github.com/fission/fission/cache"
@@ -35,6 +40,9 @@ type (
 
 		// FunctionReference -> function metadata
 		refCache *cache.Cache
+
+		stopCh chan struct{}
+		store  k8sCache.Store
 	}
 
 	resolveResultType int
@@ -61,10 +69,40 @@ const (
 )
 
 func makeFunctionReferenceResolver(fissionClient *tpr.FissionClient) *functionReferenceResolver {
-	return &functionReferenceResolver{
+	frr := &functionReferenceResolver{
 		fissionClient: fissionClient,
 		refCache:      cache.MakeCache(time.Minute, 0),
 	}
+	return frr
+}
+
+// Sync starts syncing tpr function resources from k8s api server
+func (frr *functionReferenceResolver) Sync(tprClient *rest.RESTClient) {
+	stopCh := make(chan struct{})
+	store, controller := makeK8SCache(tprClient)
+	frr.stopCh = stopCh
+	frr.store = store
+	go controller.Run(stopCh)
+}
+
+// Stop stops tpr resources syncing
+func (frr *functionReferenceResolver) Stop() {
+	frr.stopCh <- struct{}{}
+}
+
+func makeK8SCache(tprClient *rest.RESTClient) (k8sCache.Store, k8sCache.Controller) {
+	watchlist := k8sCache.NewListWatchFromClient(tprClient, "functions", metav1.NamespaceDefault, fields.Everything())
+	listWatch := &k8sCache.ListWatch{
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			return watchlist.List(options)
+		},
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			return watchlist.Watch(options)
+		},
+	}
+	resyncPeriod := 30 * time.Second
+	return k8sCache.NewInformer(listWatch, &tpr.Function{}, resyncPeriod,
+		k8sCache.ResourceEventHandlerFuncs{})
 }
 
 // resolve translates a namespace and a function reference to resolveResult.
@@ -106,10 +144,21 @@ func (frr *functionReferenceResolver) resolve(namespace string, fr *fission.Func
 
 // resolveByName simply looks up function by name in a namespace.
 func (frr *functionReferenceResolver) resolveByName(namespace, name string) (*resolveResult, error) {
-	f, err := frr.fissionClient.Functions(namespace).Get(name)
+	// get function from cache
+	obj, isExist, err := frr.store.Get(&tpr.Function{
+		Metadata: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+	})
 	if err != nil {
 		return nil, err
 	}
+	if !isExist {
+		return nil, fmt.Errorf("function %v does not exist", name)
+	}
+
+	f := obj.(*tpr.Function)
 	rr := resolveResult{
 		resolveResultType: resolveResultSingleFunction,
 		functionMetadata:  &f.Metadata,

router/httpTriggers.go

@@ -23,7 +23,11 @@ import (
 
 	"github.com/gorilla/mux"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/rest"
+	k8sCache "k8s.io/client-go/tools/cache"
 
 	"github.com/fission/fission"
 	poolmgrClient "github.com/fission/fission/poolmgr/client"
@@ -38,9 +42,11 @@ type HTTPTriggerSet struct {
 	resolver      *functionReferenceResolver
 	triggers      []tpr.Httptrigger
 	functions     []tpr.Function
+	tprClient     *rest.RESTClient
 }
 
-func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *tpr.FissionClient, poolmgr *poolmgrClient.Client, resolver *functionReferenceResolver) *HTTPTriggerSet {
+func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *tpr.FissionClient,
+	poolmgr *poolmgrClient.Client, resolver *functionReferenceResolver, tprClient *rest.RESTClient) *HTTPTriggerSet {
 	triggers := make([]tpr.Httptrigger, 1)
 	return &HTTPTriggerSet{
 		functionServiceMap: fmap,
@@ -48,6 +54,7 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, fissionClient *tpr.FissionClie
 		fissionClient:      fissionClient,
 		poolmgr:            poolmgr,
 		resolver:           resolver,
+		tprClient:          tprClient,
 	}
 }
 
@@ -135,75 +142,77 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 	// sync all http triggers
 	ts.syncTriggers()
 
-	// Watch controller for updates to triggers and update the router accordingly.
+	watchlist := k8sCache.NewListWatchFromClient(ts.tprClient, "httptriggers", metav1.NamespaceDefault, fields.Everything())
-	rv := ""
+	listWatch := &k8sCache.ListWatch{
-	for {
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-		wi, err := ts.fissionClient.Httptriggers(metav1.NamespaceAll).Watch(metav1.ListOptions{
+			return watchlist.List(options)
-			ResourceVersion: rv,
+		},
-		})
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-		if err != nil {
+			return watchlist.Watch(options)
-			log.Fatalf("Failed to watch http trigger list: %v", err)
+		},
-		}
-
-		for {
-			ev, more := <-wi.ResultChan()
-			if !more {
-				// restart watch from last rv
-				break
-			}
-			if ev.Type == watch.Error {
-				// restart watch from the start
-				rv = ""
-				time.Sleep(time.Second)
-				break
-			}
-			ht := ev.Object.(*tpr.Httptrigger)
-			rv = ht.Metadata.ResourceVersion
-			ts.syncTriggers()
-		}
 	}
+	resyncPeriod := 30 * time.Second
+	_, controller := k8sCache.NewInformer(listWatch, &tpr.Httptrigger{}, resyncPeriod,
+		k8sCache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				ts.syncTriggers()
+			},
+			DeleteFunc: func(obj interface{}) {
+				ts.syncTriggers()
+			},
+			UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+				ts.syncTriggers()
+			},
+		})
+	stop := make(chan struct{})
+	defer func() {
+		stop <- struct{}{}
+	}()
+	controller.Run(stop)
 }
 
 func (ts *HTTPTriggerSet) watchFunctions() {
-	rv := ""
+	ts.syncTriggers()
-	for {
-		wi, err := ts.fissionClient.Functions(metav1.NamespaceAll).Watch(metav1.ListOptions{
-			ResourceVersion: rv,
-		})
-		if err != nil {
-			log.Fatalf("Failed to watch function list: %v", err)
-		}
 
-		for {
+	watchlist := k8sCache.NewListWatchFromClient(ts.tprClient, "functions", metav1.NamespaceDefault, fields.Everything())
-			ev, more := <-wi.ResultChan()
+	listWatch := &k8sCache.ListWatch{
-			if !more {
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				// restart watch from last rv
+			return watchlist.List(options)
-				break
+		},
-			}
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-			if ev.Type == watch.Error {
+			return watchlist.Watch(options)
-				// restart watch from the start
+		},
-				rv = ""
-				time.Sleep(time.Second)
-				break
-			}
-			fn := ev.Object.(*tpr.Function)
-			rv = fn.Metadata.ResourceVersion
-
-			// update resolver function reference cache
-			for key, rr := range ts.resolver.copy() {
-				if key.functionReference.Name == fn.Metadata.Name &&
-					rr.functionMetadata.ResourceVersion != fn.Metadata.ResourceVersion {
-					err := ts.resolver.delete(key.namespace, &key.functionReference)
-					if err != nil {
-						log.Printf("Error deleting functionReferenceResolver cache: %v", err)
-					}
-					break
-				}
-			}
-
-			ts.syncTriggers()
-		}
 	}
+	resyncPeriod := 30 * time.Second
+	_, controller := k8sCache.NewInformer(listWatch, &tpr.Function{}, resyncPeriod,
+		k8sCache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				ts.syncTriggers()
+			},
+			DeleteFunc: func(obj interface{}) {
+				ts.syncTriggers()
+			},
+			UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+				fn := newObj.(*tpr.Function)
+				// update resolver function reference cache
+				for key, rr := range ts.resolver.copy() {
+					if key.functionReference.Name == fn.Metadata.Name &&
+						rr.functionMetadata.ResourceVersion != fn.Metadata.ResourceVersion {
+						err := ts.resolver.delete(key.namespace, &key.functionReference)
+						if err != nil {
+							log.Printf("Error deleting functionReferenceResolver cache: %v", err)
+						}
+						break
+					}
+				}
+				ts.syncTriggers()
+			},
+		})
+	stop := make(chan struct{})
+	defer func() {
+		stop <- struct{}{}
+	}()
+	controller.Run(stop)
 }
 
 func (ts *HTTPTriggerSet) syncTriggers() {

router/router.go

@@ -77,9 +77,14 @@ func Start(port int, poolmgrUrl string) {
 	if err != nil {
 		log.Fatalf("Error connecting to kubernetes API: %v", err)
 	}
+	restClient := fissionClient.GetTprClient()
 	poolmgr := poolmgrClient.MakeClient(poolmgrUrl)
 	resolver := makeFunctionReferenceResolver(fissionClient)
-	triggers := makeHTTPTriggerSet(fmap, fissionClient, poolmgr, resolver)
+	resolver.Sync(restClient)
+	defer func() {
+		resolver.Stop()
+	}()
+	triggers := makeHTTPTriggerSet(fmap, fissionClient, poolmgr, resolver, restClient)
 	log.Printf("Starting router at port %v\n", port)
 	serve(port, triggers)
 }

router/router_test.go

@@ -58,7 +58,7 @@ func TestRouter(t *testing.T) {
 	frr.refCache.Set(nfr, rr)
 
 	// HTTP trigger set with a trigger for this function
-	triggers := makeHTTPTriggerSet(fmap, nil, nil, frr)
+	triggers := makeHTTPTriggerSet(fmap, nil, nil, frr, nil)
 	triggerUrl := "/foo"
 	triggers.triggers = append(triggers.triggers,
 		tpr.Httptrigger{

tpr/client.go

@@ -209,3 +209,6 @@ func (fc *FissionClient) Packages(ns string) PackageInterface {
 func (fc *FissionClient) WaitForTPRs() {
 	waitForTPRs(fc.tprClient)
 }
+func (fc *FissionClient) GetTprClient() *rest.RESTClient {
+	return fc.tprClient
+}