From 7bb397dfeecde16c7cd81ee34a89f50e10bb1cb6 Mon Sep 17 00:00:00 2001 From: Vishal Date: Thu, 16 Nov 2017 03:27:19 +0530 Subject: [PATCH] Changed podName to a generic objectReference in fscache (#391) Changed podName to a generic objectReference in function service cache implementation. --- poolmgr/functionServiceCache.go | 86 +++++++++++++++------------- poolmgr/functionServiceCache_test.go | 14 +++-- poolmgr/gp.go | 47 +++++++++------ 3 files changed, 86 insertions(+), 61 deletions(-) diff --git a/poolmgr/functionServiceCache.go b/poolmgr/functionServiceCache.go index c4ea0ccb..4bf79be2 100644 --- a/poolmgr/functionServiceCache.go +++ b/poolmgr/functionServiceCache.go @@ -21,6 +21,7 @@ import ( "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api" "github.com/fission/fission" "github.com/fission/fission/cache" @@ -28,43 +29,50 @@ import ( ) type fscRequestType int +type backendType int const ( TOUCH fscRequestType = iota LISTOLD LOG - DELETE_BY_POD + DELETE_BY_OBJECT +) + +const ( + POOLMGR backendType = iota + NEWDEPLOY ) 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. - podName string // pod name (within the function namespace) + 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 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 - byPod *cache.Cache // podname -> function : map[string]metav1.ObjectMeta + 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 requestChannel chan *fscRequest } fscRequest struct { - requestType fscRequestType - address string - podName string - age time.Duration - env *metav1.ObjectMeta // used for ListOld - responseChannel chan *fscResponse + requestType fscRequestType + address string + kubernetesObject api.ObjectReference + age time.Duration + env *metav1.ObjectMeta // used for ListOld + responseChannel chan *fscResponse } fscResponse struct { - podNames []string - deleted bool + objects []api.ObjectReference + deleted bool error } ) @@ -73,7 +81,7 @@ func MakeFunctionServiceCache() *functionServiceCache { fsc := &functionServiceCache{ byFunction: cache.MakeCache(0, 0), byAddress: cache.MakeCache(0, 0), - byPod: cache.MakeCache(0, 0), + byKubeObject: cache.MakeCache(0, 0), requestChannel: make(chan *fscRequest), } go fsc.service() @@ -90,9 +98,9 @@ func (fsc *functionServiceCache) service() { resp.error = fsc._touchByAddress(req.address) case LISTOLD: // get svcs idle for > req.age - byPodCopy := fsc.byPod.Copy() - pods := make([]string, 0) - for podNameI, mI := range byPodCopy { + 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 { @@ -102,21 +110,21 @@ func (fsc *functionServiceCache) service() { if fsvc.environment.Metadata.UID == req.env.UID && time.Now().Sub(fsvc.atime) > req.age { - podName := podNameI.(string) - pods = append(pods, podName) + obj := objI.(api.ObjectReference) + kubeObjects = append(kubeObjects, obj) } } } - resp.podNames = pods + resp.objects = kubeObjects 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", key, fsvc.podName) + log.Printf("%v\t%v\t%v", key, fsvc.kubernetesObject.Kind, fsvc.kubernetesObject.Name) } - case DELETE_BY_POD: - resp.deleted, resp.error = fsc._deleteByPod(req.podName, req.age) + case DELETE_BY_OBJECT: + resp.deleted, resp.error = fsc._deleteByKubeObject(req.kubernetesObject, req.age) } req.responseChannel <- resp } @@ -157,7 +165,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { fsvc.ctime = now fsvc.atime = now - // Add to byAddress and byPod caches. Ignore NameExists errors + // Add to byAddress and byKubernetesObject caches. Ignore NameExists errors // because of multiple-specialization. See issue #331. err, _ = fsc.byAddress.Set(fsvc.address, *fsvc.function) if err != nil { @@ -169,7 +177,7 @@ func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { log.Printf("error caching fsvc: %v", err) return err, nil } - err, _ = fsc.byPod.Set(fsvc.podName, *fsvc.function) + err, _ = fsc.byKubeObject.Set(fsvc.kubernetesObject, *fsvc.function) if err != nil { if fe, ok := err.(fission.Error); ok { if fe.Code == fission.ErrorNameExists { @@ -208,22 +216,22 @@ func (fsc *functionServiceCache) _touchByAddress(address string) error { return nil } -func (fsc *functionServiceCache) DeleteByPod(podName string, minAge time.Duration) (bool, error) { +func (fsc *functionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ - requestType: DELETE_BY_POD, - podName: podName, - age: minAge, - responseChannel: responseChannel, + requestType: DELETE_BY_OBJECT, + kubernetesObject: obj, + age: minAge, + responseChannel: responseChannel, } resp := <-responseChannel return resp.deleted, resp.error } -// _deleteByPod deletes the entry keyed by podName, but only if it is +// _deleteByKubeObject deletes the entry keyed by Kubernetes Object, but only if it is // at least minAge old. -func (fsc *functionServiceCache) _deleteByPod(podName string, minAge time.Duration) (bool, error) { - mI, err := fsc.byPod.Get(podName) +func (fsc *functionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) { + mI, err := fsc.byKubeObject.Get(obj) if err != nil { return false, err } @@ -240,11 +248,11 @@ func (fsc *functionServiceCache) _deleteByPod(podName string, minAge time.Durati fsc.byFunction.Delete(crd.CacheKey(&m)) fsc.byAddress.Delete(fsvc.address) - fsc.byPod.Delete(podName) + fsc.byKubeObject.Delete(obj) return true, nil } -func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]string, error) { +func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) { responseChannel := make(chan *fscResponse) fsc.requestChannel <- &fscRequest{ requestType: LISTOLD, @@ -253,7 +261,7 @@ func (fsc *functionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Durati responseChannel: responseChannel, } resp := <-responseChannel - return resp.podNames, resp.error + return resp.objects, resp.error } func (fsc *functionServiceCache) Log() { diff --git a/poolmgr/functionServiceCache_test.go b/poolmgr/functionServiceCache_test.go index 7c1bd8d4..b3e81529 100644 --- a/poolmgr/functionServiceCache_test.go +++ b/poolmgr/functionServiceCache_test.go @@ -6,6 +6,7 @@ import ( "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/pkg/api" "github.com/fission/fission" "github.com/fission/fission/crd" @@ -39,9 +40,14 @@ func TestFunctionServiceCache(t *testing.T) { }, }, address: "xxx", - podName: "yyy", - ctime: now, - atime: now, + kubernetesObject: api.ObjectReference{ + Kind: "pod", + Name: "xxx", + APIVersion: "v1", + Namespace: "fission-function", + }, + ctime: now, + atime: now, } err, _ := fsc.Add(*fsvc) if err != nil { @@ -67,7 +73,7 @@ func TestFunctionServiceCache(t *testing.T) { log.Panicf("Failed to touch fsvc: %v", err) } - deleted, err := fsc.DeleteByPod(fsvc.podName, 0) + deleted, err := fsc.DeleteByKubeObject(fsvc.kubernetesObject, 0) if err != nil { fsc.Log() log.Panicf("Failed to delete fsvc: %v", err) diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 613f9e1f..163fef84 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -36,6 +36,7 @@ import ( "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/pkg/api" apiv1 "k8s.io/client-go/pkg/api/v1" "k8s.io/client-go/pkg/apis/extensions/v1beta1" @@ -556,13 +557,23 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*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, + } + fsvc := &funcSvc{ - function: m, - environment: gp.env, - address: svcHost, - podName: pod.ObjectMeta.Name, - ctime: time.Now(), - atime: time.Now(), + function: m, + environment: gp.env, + address: svcHost, + kubernetesObject: kubeObjRef, + backend: POOLMGR, + ctime: time.Now(), + atime: time.Now(), } err, existingFsvc := gp.fsCache.Add(*fsvc) @@ -571,9 +582,9 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) { if fe.Code == fission.ErrorNameExists { // Some other thread beat us to it -- return the other thread's fsvc and clean up // our own. - log.Printf("func svc already exists: %v", existingFsvc.podName) + log.Printf("func svc already exists: %v", existingFsvc.kubernetesObject.Name) go func() { - gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(fsvc.podName, nil) + gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(fsvc.kubernetesObject.Name, nil) }() return existingFsvc, nil } @@ -583,19 +594,19 @@ func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*funcSvc, error) { return fsvc, nil } -func (gp *GenericPool) CleanupFunctionService(podName string) error { +func (gp *GenericPool) CleanupFunctionService(obj api.ObjectReference) error { // remove ourselves from fsCache (only if we're still old) - deleted, err := gp.fsCache.DeleteByPod(podName, gp.idlePodReapTime) + deleted, err := gp.fsCache.DeleteByKubeObject(obj, gp.idlePodReapTime) if err != nil { return err } if !deleted { - log.Printf("Not deleting %v, in use", podName) + log.Printf("Not deleting %v, in use", obj.Name) return nil } - pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(podName, metav1.GetOptions{}) + pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(obj.Name, metav1.GetOptions{}) if err != nil { return err } @@ -613,7 +624,7 @@ func (gp *GenericPool) CleanupFunctionService(podName string) error { } // delete pod - err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(podName, nil) + err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(obj.Name, nil) if err != nil { return err } @@ -624,16 +635,16 @@ func (gp *GenericPool) CleanupFunctionService(podName string) error { func (gp *GenericPool) idlePodReaper() { for { time.Sleep(time.Minute) - podNames, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime) + objects, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime) if err != nil { log.Printf("Error reaping idle pods: %v", err) continue } - for _, podName := range podNames { - log.Printf("Reaping idle pod '%v'", podName) - err := gp.CleanupFunctionService(podName) + for _, obj := range objects { + log.Printf("Reaping idle pod '%v'", obj.Name) + err := gp.CleanupFunctionService(obj) if err != nil { - log.Printf("Error deleting idle pod '%v': %v", podName, err) + log.Printf("Error deleting idle pod '%v': %v", obj.Name, err) } } }