From bed189c715984fd85a7d71144249b66294170496 Mon Sep 17 00:00:00 2001 From: Vishal Date: Wed, 17 Jan 2018 02:35:24 +0530 Subject: [PATCH] Fscache support for multiple kubernetes objects (#435) Modifies the fsCache to support a list of arbitrary kubernetes objects instead of just one pod. This allows better support for the newdeploy backend. Also removes the byKubeObject cache index; cleanup is changed to use the byFunction index, and byKubeObject is no longer needed. --- executor/fscache/functionServiceCache.go | 105 +++++------------- executor/fscache/functionServiceCache_test.go | 32 ++++-- executor/poolmgr/gp.go | 85 +++++++------- 3 files changed, 96 insertions(+), 126 deletions(-) diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 97e17789..3398cd0c 100644 --- a/executor/fscache/functionServiceCache.go +++ b/executor/fscache/functionServiceCache.go @@ -35,7 +35,6 @@ const ( TOUCH fscRequestType = iota LISTOLD LOG - DELETE_BY_OBJECT ) const ( @@ -45,33 +44,32 @@ const ( type ( FuncSvc struct { - Function *metav1.ObjectMeta // function this pod/service is for - Environment *crd.Environment // function's environment - Address string // Host:Port or IP:Port that the function's service can be reached at. - KubernetesObject api.ObjectReference // Kubernetes Object (within the function namespace) - Backend backendType + Function *metav1.ObjectMeta // function this pod/service is for + Environment *crd.Environment // function's environment + Address string // Host:Port or IP:Port that the function's service can be reached at. + KubernetesObjects []api.ObjectReference // Kubernetes Objects (within the function namespace) + Backend backendType Ctime time.Time Atime time.Time } FunctionServiceCache struct { - byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc - byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta - byKubeObject *cache.Cache // obj -> function : map[api.ObjectReference]metav1.ObjectMeta + byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc + byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta requestChannel chan *fscRequest } fscRequest struct { - requestType fscRequestType - address string - kubernetesObject api.ObjectReference - age time.Duration - env *metav1.ObjectMeta // used for ListOld - responseChannel chan *fscResponse + requestType fscRequestType + address string + kubernetesObjects []api.ObjectReference + age time.Duration + env *metav1.ObjectMeta // used for ListOld + responseChannel chan *fscResponse } fscResponse struct { - objects []api.ObjectReference + objects []*FuncSvc deleted bool error } @@ -81,7 +79,6 @@ func MakeFunctionServiceCache() *FunctionServiceCache { fsc := &FunctionServiceCache{ byFunction: cache.MakeCache(0, 0), byAddress: cache.MakeCache(0, 0), - byKubeObject: cache.MakeCache(0, 0), requestChannel: make(chan *fscRequest), } go fsc.service() @@ -98,33 +95,25 @@ func (fsc *FunctionServiceCache) service() { resp.error = fsc._touchByAddress(req.address) case LISTOLD: // get svcs idle for > req.age - byKubeObjectCopy := fsc.byKubeObject.Copy() - kubeObjects := make([]api.ObjectReference, 0) - for objI, mI := range byKubeObjectCopy { - m := mI.(metav1.ObjectMeta) - fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m)) - if err != nil { - resp.error = err - } else { - fsvc := fsvcI.(*FuncSvc) - if fsvc.Environment.Metadata.UID == req.env.UID && - time.Now().Sub(fsvc.Atime) > req.age { - - obj := objI.(api.ObjectReference) - kubeObjects = append(kubeObjects, obj) - } + fscs := fsc.byFunction.Copy() + funcObjects := make([]*FuncSvc, 0) + for _, funcSvc := range fscs { + fsvc := funcSvc.(*FuncSvc) + if fsvc.Environment.Metadata.UID == req.env.UID && + time.Now().Sub(fsvc.Atime) > req.age { + funcObjects = append(funcObjects, fsvc) } } - resp.objects = kubeObjects + resp.objects = funcObjects case LOG: funcCopy := fsc.byFunction.Copy() log.Printf("Cache has %v entries", len(funcCopy)) for key, fsvcI := range funcCopy { fsvc := fsvcI.(*FuncSvc) - log.Printf("%v\t%v\t%v", key, fsvc.KubernetesObject.Kind, fsvc.KubernetesObject.Name) + for _, kubeObj := range fsvc.KubernetesObjects { + log.Printf("%v\t%v\t%v", key, kubeObj.Kind, kubeObj.Name) + } } - case DELETE_BY_OBJECT: - resp.deleted, resp.error = fsc._deleteByKubeObject(req.kubernetesObject, req.age) } req.responseChannel <- resp } @@ -165,7 +154,7 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) { fsvc.Ctime = now fsvc.Atime = now - // Add to byAddress and byKubernetesObject caches. Ignore NameExists errors + // Add to byAddress cache. Ignore NameExists errors // because of multiple-specialization. See issue #331. err, _ = fsc.byAddress.Set(fsvc.Address, *fsvc.Function) if err != nil { @@ -177,16 +166,6 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) { log.Printf("error caching fsvc: %v", err) return err, nil } - err, _ = fsc.byKubeObject.Set(fsvc.KubernetesObject, *fsvc.Function) - if err != nil { - if fe, ok := err.(fission.Error); ok { - if fe.Code == fission.ErrorNameExists { - err = nil - } - } - log.Printf("error caching fsvc: %v", err) - return err, nil - } return nil, nil } @@ -216,44 +195,18 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error { return nil } -func (fsc *FunctionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { - responseChannel := make(chan *fscResponse) - fsc.requestChannel <- &fscRequest{ - requestType: DELETE_BY_OBJECT, - kubernetesObject: obj, - age: minAge, - responseChannel: responseChannel, - } - resp := <-responseChannel - return resp.deleted, resp.error -} - -// _deleteByKubeObject deletes the entry keyed by Kubernetes Object, but only if it is -// at least minAge old. -func (fsc *FunctionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { - mI, err := fsc.byKubeObject.Get(obj) - if err != nil { - return false, err - } - m := mI.(metav1.ObjectMeta) - fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m)) - if err != nil { - return false, err - } - fsvc := fsvcI.(*FuncSvc) - +func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) { if time.Now().Sub(fsvc.Atime) < minAge { return false, nil } - fsc.byFunction.Delete(crd.CacheKey(&m)) + fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) fsc.byAddress.Delete(fsvc.Address) - fsc.byKubeObject.Delete(obj) return true, nil } -func (fsc *FunctionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) { +func (fsc *FunctionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]*FuncSvc, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ requestType: LISTOLD, diff --git a/executor/fscache/functionServiceCache_test.go b/executor/fscache/functionServiceCache_test.go index 21424049..7e161454 100644 --- a/executor/fscache/functionServiceCache_test.go +++ b/executor/fscache/functionServiceCache_test.go @@ -21,6 +21,21 @@ func TestFunctionServiceCache(t *testing.T) { var fsvc *FuncSvc now := time.Now() + objects := []api.ObjectReference{ + { + Kind: "pod", + Name: "xxx", + APIVersion: "v1", + Namespace: "fission-function", + }, + { + Kind: "pod", + Name: "xxx2", + APIVersion: "v1", + Namespace: "fission-function", + }, + } + fsvc = &FuncSvc{ Function: &metav1.ObjectMeta{ Name: "foo", @@ -39,15 +54,10 @@ func TestFunctionServiceCache(t *testing.T) { Builder: fission.Builder{}, }, }, - Address: "xxx", - KubernetesObject: api.ObjectReference{ - Kind: "pod", - Name: "xxx", - APIVersion: "v1", - Namespace: "fission-function", - }, - Ctime: now, - Atime: now, + Address: "xxx", + KubernetesObjects: objects, + Ctime: now, + Atime: now, } err, _ := fsc.Add(*fsvc) if err != nil { @@ -62,7 +72,7 @@ func TestFunctionServiceCache(t *testing.T) { } fsvc.Atime = f.Atime fsvc.Ctime = f.Ctime - if *f != *fsvc { + if f.Address != fsvc.Address { fsc.Log() log.Panicf("Incorrect fsvc \n(expected: %#v)\n (found: %#v)", fsvc, f) } @@ -73,7 +83,7 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("Failed to touch fsvc: %v", err) } - deleted, err := fsc.DeleteByKubeObject(fsvc.KubernetesObject, 0) + deleted, err := fsc.DeleteOld(fsvc, 0) if err != nil { fsc.Log() log.Panicf("Failed to delete fsvc: %v", err) diff --git a/executor/poolmgr/gp.go b/executor/poolmgr/gp.go index 2a63c107..718053ac 100644 --- a/executor/poolmgr/gp.go +++ b/executor/poolmgr/gp.go @@ -580,23 +580,25 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP) } - kubeObjRef := api.ObjectReference{ - Kind: pod.TypeMeta.Kind, - Name: pod.ObjectMeta.Name, - APIVersion: pod.TypeMeta.APIVersion, - Namespace: pod.ObjectMeta.Namespace, - ResourceVersion: pod.ObjectMeta.ResourceVersion, - UID: pod.ObjectMeta.UID, + kubeObjRefs := []api.ObjectReference{ + { + Kind: pod.TypeMeta.Kind, + Name: pod.ObjectMeta.Name, + APIVersion: pod.TypeMeta.APIVersion, + Namespace: pod.ObjectMeta.Namespace, + ResourceVersion: pod.ObjectMeta.ResourceVersion, + UID: pod.ObjectMeta.UID, + }, } fsvc := &fscache.FuncSvc{ - Function: m, - Environment: gp.env, - Address: svcHost, - KubernetesObject: kubeObjRef, - Backend: fscache.POOLMGR, - Ctime: time.Now(), - Atime: time.Now(), + Function: m, + Environment: gp.env, + Address: svcHost, + KubernetesObjects: kubeObjRefs, + Backend: fscache.POOLMGR, + Ctime: time.Now(), + Atime: time.Now(), } err, _ = gp.fsCache.Add(*fsvc) @@ -606,39 +608,41 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error return fsvc, nil } -func (gp *GenericPool) CleanupFunctionService(obj api.ObjectReference) error { +func (gp *GenericPool) CleanupFunctionService(obj *fscache.FuncSvc) error { // remove ourselves from fsCache (only if we're still old) - deleted, err := gp.fsCache.DeleteByKubeObject(obj, gp.idlePodReapTime) + deleted, err := gp.fsCache.DeleteOld(obj, gp.idlePodReapTime) if err != nil { return err } if !deleted { - log.Printf("Not deleting %v, in use", obj.Name) + log.Printf("Not deleting %v, in use", obj.Function) return nil } - pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(obj.Name, metav1.GetOptions{}) - if err != nil { - return err - } + for _, kubeobj := range obj.KubernetesObjects { + pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(kubeobj.Name, metav1.GetOptions{}) - loggerUrl := fmt.Sprintf("http://%s:1234/v1/log/%s", pod.Spec.NodeName, pod.Name) - req, err := http.NewRequest("DELETE", loggerUrl, nil) - resp, err := http.DefaultClient.Do(req) - if err != nil { - log.Printf("Error from %s daemonset logger: %v", pod.Spec.NodeName, err) - } else { - if resp.StatusCode != 200 { - log.Printf("Received not http 200(OK) status from %s daemonset logger: %s", pod.Spec.NodeName, resp.Status) + loggerUrl := fmt.Sprintf("http://%s:1234/v1/log/%s", pod.Spec.NodeName, pod.Name) + req, err := http.NewRequest("DELETE", loggerUrl, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("Error from %s daemonset logger: %v", pod.Spec.NodeName, err) + } else { + if resp.StatusCode != 200 { + log.Printf("Received not http 200(OK) status from %s daemonset logger: %s", pod.Spec.NodeName, resp.Status) + } + resp.Body.Close() } - resp.Body.Close() - } - // delete pod - err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(obj.Name, nil) - if err != nil { - return err + // delete pod + err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(kubeobj.Name, nil) + if err != nil { + return err + } + if err != nil { + return err + } } return nil @@ -647,16 +651,19 @@ func (gp *GenericPool) CleanupFunctionService(obj api.ObjectReference) error { func (gp *GenericPool) idlePodReaper() { for { time.Sleep(time.Minute) - objects, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime) + funcSvcs, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime) if err != nil { log.Printf("Error reaping idle pods: %v", err) continue } - for _, obj := range objects { - log.Printf("Reaping idle pod '%v'", obj.Name) + for _, obj := range funcSvcs { err := gp.CleanupFunctionService(obj) if err != nil { - log.Printf("Error deleting idle pod '%v': %v", obj.Name, err) + log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", obj, err) + log.Printf("Object Name| Object Kind | Object Space") + for _, kubeobj := range obj.KubernetesObjects { + log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace) + } } } }