From c7692978f9f0418c088ee19b7f763487385f5bfa Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 5 Nov 2016 23:10:52 -0700 Subject: [PATCH] Reap idle pods Idle pod reaper wakes up once a minute, looks in the functionServiceCache for pods that haven't been accessed for more than idlePodReapTime, and deletes all such pods. (This doesn't yet delete pods that may be leaked from previously terminated poolmgrs; it only works with pods created by the current poolmgr instance.) --- poolmgr/api.go | 2 -- poolmgr/functionServiceCache.go | 58 ++++++++++++++++++++++++++++----- poolmgr/gp.go | 37 +++++++++++++++++++-- 3 files changed, 83 insertions(+), 14 deletions(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index c68dd94e..d4fc4ac3 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -127,8 +127,6 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { return fsvc.address, nil } - api.fsCache.Log() - // None exists, so create a new funcSvc: log.Printf("[%v] No cached function service found, creating one", m.Name) diff --git a/poolmgr/functionServiceCache.go b/poolmgr/functionServiceCache.go index 74db5046..c3a248d6 100644 --- a/poolmgr/functionServiceCache.go +++ b/poolmgr/functionServiceCache.go @@ -30,6 +30,7 @@ const ( TOUCH fscRequestType = iota LISTOLD LOG + DELETE_BY_POD ) type ( @@ -43,11 +44,13 @@ type ( fscRequest struct { requestType fscRequestType address string + podName string age time.Duration responseChannel chan *fscResponse } fscResponse struct { podNames []string + deleted bool error } ) @@ -75,11 +78,17 @@ func (fsc *functionServiceCache) service() { // get svcs idle for > req.age byPodCopy := fsc.byPod.Copy() pods := make([]string, 0) - for podNameI, fsvcI := range byPodCopy { - fsvc := fsvcI.(*funcSvc) - if time.Now().Sub(fsvc.atime) > req.age { - podName := podNameI.(string) - pods = append(pods, podName) + for podNameI, mI := range byPodCopy { + m := mI.(fission.Metadata) + fsvcI, err := fsc.byFunction.Get(m) + if err != nil { + resp.error = err + } else { + fsvc := fsvcI.(*funcSvc) + if time.Now().Sub(fsvc.atime) > req.age { + podName := podNameI.(string) + pods = append(pods, podName) + } } } resp.podNames = pods @@ -91,6 +100,8 @@ func (fsc *functionServiceCache) service() { fsvc := fsvcI.(*funcSvc) log.Printf("%v:%v\t%v", m.Name, m.Uid, fsvc.podName) } + case DELETE_BY_POD: + resp.deleted, resp.error = fsc._deleteByPod(req.podName, req.age) } req.responseChannel <- resp } @@ -166,22 +177,51 @@ func (fsc *functionServiceCache) _touchByAddress(address string) error { return nil } -func (fsc *functionServiceCache) DeleteByPod(podName string) error { +func (fsc *functionServiceCache) DeleteByPod(podName string, minAge time.Duration) (bool, error) { + responseChannel := make(chan *fscResponse) + fsc.requestChannel <- &fscRequest{ + requestType: DELETE_BY_POD, + podName: podName, + age: minAge, + responseChannel: responseChannel, + } + resp := <-responseChannel + return resp.deleted, resp.error +} + +// _deleteByPod deletes the entry keyed by podName, 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) if err != nil { - return err + return false, err } m := mI.(fission.Metadata) fsvcI, err := fsc.byFunction.Get(m) if err != nil { - return err + return false, err } fsvc := fsvcI.(*funcSvc) + if time.Now().Sub(fsvc.atime) < minAge { + return false, nil + } + fsc.byFunction.Delete(m) fsc.byAddress.Delete(fsvc.address) fsc.byPod.Delete(podName) - return nil + return true, nil +} + +func (fsc *functionServiceCache) ListOld(age time.Duration) ([]string, error) { + responseChannel := make(chan *fscResponse) + fsc.requestChannel <- &fscRequest{ + requestType: LISTOLD, + age: age, + responseChannel: responseChannel, + } + resp := <-responseChannel + return resp.podNames, resp.error } func (fsc *functionServiceCache) Log() { diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 7e92b8c3..af4f1aa0 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -435,16 +435,47 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { // our own. TODO: this is grossly inefficient, improve it with some sort of state // machine log.Printf("func svc already exists: %v", existingFsvc.podName) - go gp.CleanupFunctionService(fsvc) + go gp.CleanupFunctionService(fsvc.podName) return existingFsvc, nil } return fsvc, nil } -func (gp *GenericPool) CleanupFunctionService(fsvc *funcSvc) { +func (gp *GenericPool) CleanupFunctionService(podName string) error { + // remove ourselves from fsCache (only if we're still old) + deleted, err := gp.fsCache.DeleteByPod(podName, gp.idlePodReapTime) + if err != nil { + return err + } + + if !deleted { + log.Printf("Not deleting %v, in use", podName) + return nil + } + // delete pod - // remove ourselves from fsCache + err = gp.kubernetesClient.Core().Pods(gp.namespace).Delete(podName, nil) + if err != nil { + return err + } + + return nil } func (gp *GenericPool) idlePodReaper() { + for { + time.Sleep(time.Minute) + podNames, err := gp.fsCache.ListOld(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) + if err != nil { + log.Printf("Error deleting idle pod '%v': %v", podName, err) + } + } + } }