From 439e7d4535541d9e89525abb192909f57d7e1264 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Thu, 26 Oct 2017 06:10:44 +0800 Subject: [PATCH] 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 | 14 ++- glide.yaml | 4 + router/functionReferenceResolver.go | 53 ++++++++++- router/httpTriggers.go | 137 +++++++++++++++------------- router/router.go | 7 +- router/router_test.go | 2 +- tpr/client.go | 3 + 7 files changed, 150 insertions(+), 70 deletions(-) diff --git a/glide.lock b/glide.lock index 6a7e2b0a..12be4af6 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: b73c8c15207c8906bdc2c2713da03353d7c7ab78649f222b067e805b6f4c03dd -updated: 2017-09-29T00:30:00.245121189+08:00 +hash: 285f5f29e18f61b5b834a48ce815fd40e023aa55d51acbc2ac8033cd8267635d +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 diff --git a/glide.yaml b/glide.yaml index 3cfac534..529d8511 100644 --- a/glide.yaml +++ b/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: diff --git a/router/functionReferenceResolver.go b/router/functionReferenceResolver.go index 088cad54..b7c33cb3 100644 --- a/router/functionReferenceResolver.go +++ b/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, diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 202bfc46..8a3b805a 100644 --- a/router/httpTriggers.go +++ b/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. - rv := "" - for { - wi, err := ts.fissionClient.Httptriggers(metav1.NamespaceAll).Watch(metav1.ListOptions{ - ResourceVersion: rv, - }) - if err != nil { - 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() - } + watchlist := k8sCache.NewListWatchFromClient(ts.tprClient, "httptriggers", 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 + _, 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 := "" - 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) - } + ts.syncTriggers() - 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 - } - 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() - } + watchlist := k8sCache.NewListWatchFromClient(ts.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 + _, 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() { diff --git a/router/router.go b/router/router.go index bc2ef636..2b133344 100644 --- a/router/router.go +++ b/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) } diff --git a/router/router_test.go b/router/router_test.go index e524eeca..b9fd2992 100644 --- a/router/router_test.go +++ b/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{ diff --git a/tpr/client.go b/tpr/client.go index 5ef917f1..55fecb9c 100644 --- a/tpr/client.go +++ b/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 +}