From f3eba4b825dfc62866126de5fe87c0e87303c4f1 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 16:28:49 -0700 Subject: [PATCH 01/14] Call poolmgr from router while using a cached service Poolmgr needs to know usage statistics for a function's pod. This change asynchronously taps poolmgr API when router uses a service. Poolmgr can use this information to control pod expiry. It may also be useful later as one of the metrics for autoscaling. --- poolmgr/api.go | 39 +++++++++++++++++++++++++++++++++++++-- poolmgr/client/client.go | 17 +++++++++++++++++ router/functionHandler.go | 13 +++++++++++++ 3 files changed, 67 insertions(+), 2 deletions(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index 9206cc92..3205dc77 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -23,6 +23,7 @@ import ( "log" "net/http" "os" + "strings" "time" "github.com/gorilla/handlers" @@ -45,7 +46,8 @@ type funcSvc struct { type API struct { poolMgr *GenericPoolManager functionEnv *cache.Cache // map[fission.Metadata]fission.Environment - functionService *cache.Cache // map[fission.Metadata]funcSvc + functionService *cache.Cache // map[fission.Metadata]*funcSvc + urlFuncSvc *cache.Cache // map[string]*funcSvc controller *controllerclient.Client } @@ -54,6 +56,7 @@ func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API poolMgr: gpm, functionEnv: cache.MakeCache(), functionService: cache.MakeCache(), + urlFuncSvc: cache.MakeCache(), controller: controller, } } @@ -120,6 +123,7 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { if err == nil { // Ok: return svc name svc := result.(*funcSvc) + svc.atime = time.Now() return svc.serviceName, nil } @@ -151,15 +155,46 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { err = api.functionService.Set(m, funcSvc) if err != nil { // log and ignore error - log.Printf("Error saving function service: %v", err) + log.Printf("Error caching function service: %v", err) + } + + // cache by svc hostname, for tapService() + err = api.urlFuncSvc.Set(funcSvc.serviceName, funcSvc) + if err != nil { + // log and ignore error + log.Printf("Error caching function service by name: %v", err) } return funcSvc.serviceName, nil } +// find funcSvc and update its atime +func (api *API) tapService(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request", 500) + return + } + svcName := string(body) + svcHost := strings.TrimPrefix(svcName, "http://") + + log.Printf("tap svc: %v", svcHost) + + funcSvcI, err := api.urlFuncSvc.Get(svcHost) + if err != nil { + http.Error(w, "Not found", 404) + return + } + + (funcSvcI.(*funcSvc)).atime = time.Now() + + w.WriteHeader(http.StatusOK) +} + func (api *API) Serve(port int) { r := mux.NewRouter() r.HandleFunc("/v1/getServiceForFunction", api.getServiceForFunctionApi).Methods("POST") + r.HandleFunc("/v1/tapService", api.tapService).Methods("POST") address := fmt.Sprintf(":%v", port) log.Printf("starting poolmgr at port %v", port) diff --git a/poolmgr/client/client.go b/poolmgr/client/client.go index 3556f059..f716dda2 100644 --- a/poolmgr/client/client.go +++ b/poolmgr/client/client.go @@ -24,6 +24,7 @@ import ( "encoding/json" "github.com/platform9/fission" "io/ioutil" + "net/url" ) type Client struct { @@ -58,3 +59,19 @@ func (c *Client) GetServiceForFunction(metadata *fission.Metadata) (string, erro return string(svcName), nil } + +func (c *Client) TapService(serviceUrl *url.URL) error { + url := c.poolmgrUrl + "/v1/tapService" + + serviceUrlStr := serviceUrl.String() + + resp, err := http.Post(url, "application/octet-stream", bytes.NewReader([]byte(serviceUrlStr))) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return fission.MakeErrorFromHTTP(resp) + } + return nil +} diff --git a/router/functionHandler.go b/router/functionHandler.go index ca7c88ff..3d205db9 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -80,7 +80,16 @@ func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er return http.DefaultTransport.RoundTrip(req) } +func (fh *functionHandler) tapService(serviceUrl *url.URL) { + err := fh.poolmgr.TapService(serviceUrl) + if err != nil { + log.Printf("tap service error: %v", serviceUrl.String()) + } +} + func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *http.Request) { + + // cache lookup serviceUrl, err := fh.fmap.lookup(&fh.Function) if err != nil { // Cache miss: request the Pool Manager to make a new service. @@ -99,6 +108,10 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request * // add it to the map fh.fmap.assign(&fh.Function, serviceUrl) + } else { + // if we're using our cache, asynchronously tell + // poolmgr we're using this service + go fh.tapService(serviceUrl) } // Proxy off our request to the serviceUrl, and send the response back. From 5cfe4b7c1d15c1c9808365320d3f783ba6090e15 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 17:20:09 -0700 Subject: [PATCH 02/14] Log and ignore errors on eager pool creation --- poolmgr/gpm.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 5c4213f1..48231c1b 100644 --- a/poolmgr/gpm.go +++ b/poolmgr/gpm.go @@ -108,7 +108,10 @@ func (gpm *GenericPoolManager) eagerPoolCreator() { // to keep these eagerly created pools smaller than the ones created when there are // actual function calls. for _, env := range envs { - gpm.GetPool(&env) + _, err := gpm.GetPool(&env) + if err != nil { + log.Printf("eager-create pool failed: %v", err) + } } } } From f639bc3558ac62d8f28bb8903807f27c8f22dc05 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 17:21:00 -0700 Subject: [PATCH 03/14] Use fission.Cache for functionServiceMap --- router/functionHandler.go | 3 ++ router/functionServiceMap.go | 89 ++++++------------------------------ 2 files changed, 17 insertions(+), 75 deletions(-) diff --git a/router/functionHandler.go b/router/functionHandler.go index 3d205db9..9dc13a27 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -81,6 +81,9 @@ func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er } func (fh *functionHandler) tapService(serviceUrl *url.URL) { + if fh.poolmgr == nil { + return + } err := fh.poolmgr.TapService(serviceUrl) if err != nil { log.Printf("tap service error: %v", serviceUrl.String()) diff --git a/router/functionServiceMap.go b/router/functionServiceMap.go index 671c19bc..e858930a 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -17,97 +17,36 @@ limitations under the License. package router import ( - "errors" "log" "net/url" "github.com/platform9/fission" + "github.com/platform9/fission/cache" ) -type requestType int - -const ( - LOOKUP requestType = iota // lookup the map - ASSIGN // assign function - NEXT_GEN // increment current generation - SWEEP // delete all but the current generation -) - -type functionServiceMapResponse struct { - serviceUrl url.URL - error -} -type functionServiceMapRequest struct { - Function fission.Metadata - serviceUrl url.URL - requestType - responseChannel chan<- functionServiceMapResponse -} -type functionServiceMapEntry struct { - serviceUrl url.URL - generation uint64 -} - type functionServiceMap struct { - // map (funcname, uid) -> url - svc map[fission.Metadata]functionServiceMapEntry - currentGeneration uint64 - requestChannel chan *functionServiceMapRequest + svc *cache.Cache // map[fission.Metadata]*url.URL } func makeFunctionServiceMap() *functionServiceMap { - fmap := &functionServiceMap{} - fmap.requestChannel = make(chan *functionServiceMapRequest) - fmap.svc = make(map[fission.Metadata]functionServiceMapEntry) - go fmap.functionServiceMapWork() - return fmap -} - -func (fmap *functionServiceMap) functionServiceMapWork() { - for { - req := <-fmap.requestChannel - switch req.requestType { - case LOOKUP: - e, present := fmap.svc[req.Function] - if present { - req.responseChannel <- functionServiceMapResponse{serviceUrl: e.serviceUrl} - } else { - req.responseChannel <- functionServiceMapResponse{error: errors.New("not found")} - } - case ASSIGN: - fmap.svc[req.Function] = - functionServiceMapEntry{serviceUrl: req.serviceUrl, generation: fmap.currentGeneration} - // no response - case NEXT_GEN: - fmap.currentGeneration++ - // no response - case SWEEP: - log.Panic("not implemented") - default: - log.Panic("bad request") - } + return &functionServiceMap{ + svc: cache.MakeCache(), } } func (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) { - respChannel := make(chan functionServiceMapResponse) - fmap.requestChannel <- &functionServiceMapRequest{Function: *f, requestType: LOOKUP, responseChannel: respChannel} - resp := <-respChannel - if resp.error != nil { - return nil, resp.error - } else { - return &resp.serviceUrl, nil + item, err := fmap.svc.Get(*f) + if err != nil { + return nil, err } + u := item.(*url.URL) + return u, nil } func (fmap *functionServiceMap) assign(f *fission.Metadata, serviceUrl *url.URL) { - fmap.requestChannel <- &functionServiceMapRequest{Function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN} -} - -func (fmap *functionServiceMap) nextGen() { - fmap.requestChannel <- &functionServiceMapRequest{requestType: NEXT_GEN} -} - -func (fmap *functionServiceMap) sweep() { - fmap.requestChannel <- &functionServiceMapRequest{requestType: SWEEP} + //fmap.requestChannel <- &functionServiceMapRequest{Function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN} + err := fmap.svc.Set(*f, serviceUrl) + if err != nil { + log.Printf("error caching svc for function: %v", err) + } } From 094da5c69a6ca7587243d21190b4c7bf53efde90 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 22:12:34 -0700 Subject: [PATCH 04/14] Add expiry time to cache MakeCache() now takes a time.Duration after which entries are considered invalid, and not returned from a Get(). --- cache/cache.go | 47 ++++++++++++++++++++++++++++++++++++++++----- cache/cache_test.go | 31 +++++++++++++++++++----------- 2 files changed, 62 insertions(+), 16 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index 87c24f6d..c327c3fa 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -29,6 +29,7 @@ const ( GET requestType = iota SET DELETE + EXPIRE ) type ( @@ -39,6 +40,7 @@ type ( } Cache struct { cache map[interface{}]Value + expiryTime time.Duration requestChannel chan *request } @@ -54,12 +56,26 @@ type ( } ) -func MakeCache() *Cache { +func (c *Cache) IsOld(v *Value) bool { + if c.expiryTime == 0 { + return false + } + if time.Now().Sub(v.atime) > c.expiryTime { + return true + } + return false +} + +func MakeCache(expiryTime time.Duration) *Cache { c := &Cache{ cache: make(map[interface{}]Value), + expiryTime: expiryTime, requestChannel: make(chan *request), } go c.service() + if expiryTime != time.Duration(0) { + go c.expiryService() + } return c } @@ -73,11 +89,16 @@ func (c *Cache) service() { if !ok { resp.error = fission.MakeError(fission.ErrorNotFound, fmt.Sprintf("key '%v' not found", req.key)) + } else if c.IsOld(&val) { + resp.error = fission.MakeError(fission.ErrorNotFound, + fmt.Sprintf("key '%v' expired (atime %v)", req.key, val.atime)) + delete(c.cache, req.key) + } else { + // update atime + val.atime = time.Now() + c.cache[req.key] = val + resp.value = val.value } - val.atime = time.Now() - c.cache[req.key] = val - - resp.value = val.value req.responseChannel <- resp case SET: now := time.Now() @@ -90,6 +111,13 @@ func (c *Cache) service() { case DELETE: delete(c.cache, req.key) req.responseChannel <- resp + case EXPIRE: + for k, v := range c.cache { + if c.IsOld(&v) { + delete(c.cache, k) + } + } + // no response default: resp.error = fission.MakeError(fission.ErrorInvalidArgument, fmt.Sprintf("invalid request type: %v", req.requestType)) @@ -131,3 +159,12 @@ func (c *Cache) Delete(key interface{}) error { resp := <-respChannel return resp.error } + +func (c *Cache) expiryService() { + for { + time.Sleep(time.Minute) + c.requestChannel <- &request{ + requestType: EXPIRE, + } + } +} diff --git a/cache/cache_test.go b/cache/cache_test.go index 52ba9657..fc4baa36 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -18,30 +18,39 @@ package cache import "testing" import "log" +import "time" + +func checkErr(err error) { + if err != nil { + log.Panicf("err: %v", err) + } +} func TestCache(t *testing.T) { - c := MakeCache() + c := MakeCache(100 * time.Millisecond) err := c.Set("a", "b") - if err != nil { - log.Panicf("error: %v", err) - } + checkErr(err) val, err := c.Get("a") - if err != nil { - log.Panicf("error: %v", err) - } + checkErr(err) if val != "b" { log.Panicf("value %v", val) } err = c.Delete("a") - if err != nil { - log.Panicf("error: %v", err) - } + checkErr(err) _, err = c.Get("a") if err == nil { - log.Panicf("error: %v", err) + log.Panicf("found deleted element") + } + + err = c.Set("expires", "42") + checkErr(err) + time.Sleep(150 * time.Millisecond) + _, err = c.Get("expires") + if err == nil { + log.Panicf("found expired element") } } From d512159cb3a97f3a407ccb4eb8e97be6a8ee85c9 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 22:14:18 -0700 Subject: [PATCH 05/14] Fix poolmgr cache bug Poolmgr cache was useless because it was keying by *fission.Metadata instead of the metadata itself. --- poolmgr/api.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index 3205dc77..aaa2b0c0 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -54,9 +54,9 @@ type API struct { func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API { return &API{ poolMgr: gpm, - functionEnv: cache.MakeCache(), - functionService: cache.MakeCache(), - urlFuncSvc: cache.MakeCache(), + functionEnv: cache.MakeCache(0), + functionService: cache.MakeCache(time.Minute), + urlFuncSvc: cache.MakeCache(time.Minute), controller: controller, } } @@ -90,7 +90,7 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error var env *fission.Environment // Cached ? - result, err := api.functionEnv.Get(m) + result, err := api.functionEnv.Get(*m) if err == nil { env = result.(*fission.Environment) return env, nil @@ -111,7 +111,7 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error } // cache for future - api.functionEnv.Set(m, env) + api.functionEnv.Set(*m, env) return env, nil } @@ -119,7 +119,7 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { // Check function -> svc map log.Printf("[%v] Checking for cached function service", m.Name) - result, err := api.functionService.Get(m) + result, err := api.functionService.Get(*m) if err == nil { // Ok: return svc name svc := result.(*funcSvc) @@ -152,7 +152,7 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { } // add to cache - err = api.functionService.Set(m, funcSvc) + err = api.functionService.Set(*m, funcSvc) if err != nil { // log and ignore error log.Printf("Error caching function service: %v", err) From a000d4cc3208133db41952a0c9b7b66b87f21d28 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 3 Nov 2016 22:15:44 -0700 Subject: [PATCH 06/14] Update router for cache expiry --- router/functionServiceMap.go | 14 +++++++------- router/router.go | 5 +++-- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/router/functionServiceMap.go b/router/functionServiceMap.go index e858930a..cbc4f999 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -19,23 +19,24 @@ package router import ( "log" "net/url" + "time" "github.com/platform9/fission" "github.com/platform9/fission/cache" ) type functionServiceMap struct { - svc *cache.Cache // map[fission.Metadata]*url.URL + cache *cache.Cache // map[fission.Metadata]*url.URL } -func makeFunctionServiceMap() *functionServiceMap { +func makeFunctionServiceMap(expiry time.Duration) *functionServiceMap { return &functionServiceMap{ - svc: cache.MakeCache(), + cache: cache.MakeCache(expiry), } } func (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) { - item, err := fmap.svc.Get(*f) + item, err := fmap.cache.Get(*f) if err != nil { return nil, err } @@ -44,9 +45,8 @@ func (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) { } func (fmap *functionServiceMap) assign(f *fission.Metadata, serviceUrl *url.URL) { - //fmap.requestChannel <- &functionServiceMapRequest{Function: *f, serviceUrl: *serviceUrl, requestType: ASSIGN} - err := fmap.svc.Set(*f, serviceUrl) + err := fmap.cache.Set(*f, serviceUrl) if err != nil { - log.Printf("error caching svc for function: %v", err) + log.Printf("error caching service url for function: %v", err) } } diff --git a/router/router.go b/router/router.go index 90e8dd05..068b3ec9 100644 --- a/router/router.go +++ b/router/router.go @@ -41,13 +41,14 @@ package router import ( "fmt" + "log" "net/http" + "time" "github.com/gorilla/mux" controllerClient "github.com/platform9/fission/controller/client" poolmgrClient "github.com/platform9/fission/poolmgr/client" - "log" ) // request url ---[mux]---> Function(name,uid) ----[fmap]----> k8s service url @@ -68,7 +69,7 @@ func serve(port int, httpTriggerSet *HTTPTriggerSet) { } func Start(port int, controllerUrl string, poolmgrUrl string) { - fmap := makeFunctionServiceMap() + fmap := makeFunctionServiceMap(time.Minute) controller := controllerClient.MakeClient(controllerUrl) poolmgr := poolmgrClient.MakeClient(poolmgrUrl) From d805928f3ef0f05f2a130602f2bab75f6733caf6 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 4 Nov 2016 10:29:20 -0700 Subject: [PATCH 07/14] Cache: add method to get a copy of the map --- cache/cache.go | 24 +++++++++++++++++++++--- cache/cache_test.go | 7 +++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index c327c3fa..cc9d37b4 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -30,6 +30,7 @@ const ( SET DELETE EXPIRE + COPY ) type ( @@ -52,15 +53,16 @@ type ( } response struct { error - value interface{} + mapCopy map[interface{}]interface{} + value interface{} } ) func (c *Cache) IsOld(v *Value) bool { - if c.expiryTime == 0 { + if c.expiryTime == time.Duration(0) { return false } - if time.Now().Sub(v.atime) > c.expiryTime { + if time.Now().Sub(v.ctime) > c.expiryTime { return true } return false @@ -118,6 +120,12 @@ func (c *Cache) service() { } } // no response + case COPY: + resp.mapCopy = make(map[interface{}]interface{}) + for k, v := range c.cache { + resp.mapCopy[k] = v + } + req.responseChannel <- resp default: resp.error = fission.MakeError(fission.ErrorInvalidArgument, fmt.Sprintf("invalid request type: %v", req.requestType)) @@ -160,6 +168,16 @@ func (c *Cache) Delete(key interface{}) error { return resp.error } +func (c *Cache) Copy() map[interface{}]interface{} { + respChannel := make(chan *response) + c.requestChannel <- &request{ + requestType: COPY, + responseChannel: respChannel, + } + resp := <-respChannel + return resp.mapCopy +} + func (c *Cache) expiryService() { for { time.Sleep(time.Minute) diff --git a/cache/cache_test.go b/cache/cache_test.go index fc4baa36..e307feed 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -31,6 +31,8 @@ func TestCache(t *testing.T) { err := c.Set("a", "b") checkErr(err) + err = c.Set("p", "q") + checkErr(err) val, err := c.Get("a") checkErr(err) @@ -38,6 +40,11 @@ func TestCache(t *testing.T) { log.Panicf("value %v", val) } + cc := c.Copy() + if len(cc) != 2 { + log.Panicf("expected 2 items") + } + err = c.Delete("a") checkErr(err) From 35a5c947312dca35fadeaad53d557a3e65095ce2 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 4 Nov 2016 10:29:44 -0700 Subject: [PATCH 08/14] Add a reaped bit to funcSvc struct --- poolmgr/api.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index aaa2b0c0..da7ee496 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -35,10 +35,12 @@ import ( ) type funcSvc struct { - function *fission.Metadata // function this thing is for + function *fission.Metadata // function this pod/service is for environment *fission.Environment // env it was obtained from serviceName string // name of k8s svc + reaped bool // if true, the pod has been deleted + ctime time.Time atime time.Time } From 78687d24b38e0828ec54e0aa8dc8aa0c6f874eba Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 4 Nov 2016 10:33:09 -0700 Subject: [PATCH 09/14] Idle pod reaper Delete pods that are unused for more than a certain timeout. This change isn't complete -- other funcSvc caches need to be invalidated on delete. This needs a bit of refactoring. --- poolmgr/gp.go | 41 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 2a73a164..5053172c 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -45,7 +45,10 @@ type ( namespace string // namespace to keep our resources podReadyTimeout time.Duration // timeout for generic pods to become ready controllerUrl string - useSvc bool + idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted + + useSvc bool // create service for + podFuncSvc *cache.Cache // map[pod.ObjectMeta.Name]*funcSvc kubernetesClient *kubernetes.Clientset requestChannel chan *choosePodRequest @@ -70,15 +73,20 @@ func MakeGenericPool( namespace string) (*GenericPool, error) { log.Printf("Creating pool for environment %v", env.Metadata) + // TODO: in general we need to provide the user a way to configure pools. Initial + // replicas, autoscaling params, various timeouts, etc. gp := &GenericPool{ env: env, - replicas: initialReplicas, + replicas: initialReplicas, // TODO make this an env param instead? requestChannel: make(chan *choosePodRequest), kubernetesClient: kubernetesClient, namespace: namespace, - podReadyTimeout: 5 * time.Minute, + podReadyTimeout: 5 * time.Minute, // TODO make this an env param? controllerUrl: controllerUrl, - useSvc: false, + idlePodReapTime: 3 * time.Minute, // TODO make this configurable + + useSvc: false, + podFuncSvc: cache.MakeCache(0), } // create the pool @@ -95,6 +103,9 @@ func MakeGenericPool( } go gp.choosePodService() + + go gp.idlePodReaper() + return gp, nil } @@ -412,5 +423,27 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { ctime: time.Now(), atime: time.Now(), } + gp.podFuncSvc[pod.ObjectMeta.Name].Set(fsvc) return fsvc, nil } + +func (gp *GenericPool) idlePodReaper() { + for { + podmap := gp.podFuncSvc.Copy() + for podNameI, funcSvcI := range podmap { + podName := podNameI.(string) + funcSvc := funcSvcI.(*funcSvc) + lastAccessTime := funcSvc.atime + if time.Now().Sub(lastAccessTime) < gp.idlePodReapTime { + continue + } + + log.Printf("Reaping idle pod %v (last used at %v)", podName, lastAccessTime) + err := gp.kubernetesClient.Core().Pods(gp.namespace).Delete(podName) + if err != nil { + log.Printf("Error reaping pod: %v", err) + continue + } + } + } +} From 68f860dcb3143c63497bc43361e0dcbbda55a183 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 4 Nov 2016 10:36:54 -0700 Subject: [PATCH 10/14] Update unit tests for functionServiceMap change --- router/functionHandler_test.go | 2 +- router/functionServiceMap_test.go | 2 +- router/router_test.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/router/functionHandler_test.go b/router/functionHandler_test.go index 2d7537a8..c25a50d1 100644 --- a/router/functionHandler_test.go +++ b/router/functionHandler_test.go @@ -50,7 +50,7 @@ func TestFunctionProxying(t *testing.T) { log.Printf("Created backend svc at %v", backendURL) fn := &fission.Metadata{Name: "foo", Uid: "xxx"} - fmap := makeFunctionServiceMap() + fmap := makeFunctionServiceMap(0) fmap.assign(fn, backendURL) fh := &functionHandler{fmap: fmap, Function: *fn} diff --git a/router/functionServiceMap_test.go b/router/functionServiceMap_test.go index 9a57cc56..55cd98e4 100644 --- a/router/functionServiceMap_test.go +++ b/router/functionServiceMap_test.go @@ -24,7 +24,7 @@ import ( ) func TestFunctionServiceMap(t *testing.T) { - m := makeFunctionServiceMap() + m := makeFunctionServiceMap(0) fn := &fission.Metadata{Name: "foo", Uid: "012"} u, err := url.Parse("/foo012") if err != nil { diff --git a/router/router_test.go b/router/router_test.go index 521c6cd4..0df37ea2 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -25,7 +25,7 @@ import ( ) func TestRouter(t *testing.T) { - fmap := makeFunctionServiceMap() + fmap := makeFunctionServiceMap(0) fn := &fission.Metadata{Name: "foo", Uid: "xxx"} testResponseString := "hi" From 9dcadd5f4ae4512af6ce77bbc79dcd15dad3e344 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 5 Nov 2016 21:41:57 -0700 Subject: [PATCH 11/14] Support cache expiry based on create/last access time --- cache/cache.go | 57 +++++++++++++++++++++++++++++++------------------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index cc9d37b4..46ce3969 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -19,6 +19,7 @@ package cache import ( "time" + "errors" "fmt" "github.com/platform9/fission" ) @@ -40,8 +41,9 @@ type ( value interface{} } Cache struct { - cache map[interface{}]Value - expiryTime time.Duration + cache map[interface{}]*Value + ctimeExpiry time.Duration + atimeExpiry time.Duration requestChannel chan *request } @@ -53,29 +55,33 @@ type ( } response struct { error - mapCopy map[interface{}]interface{} - value interface{} + existingValue interface{} + mapCopy map[interface{}]interface{} + value interface{} } ) func (c *Cache) IsOld(v *Value) bool { - if c.expiryTime == time.Duration(0) { - return false - } - if time.Now().Sub(v.ctime) > c.expiryTime { + if (c.ctimeExpiry != time.Duration(0)) && (time.Now().Sub(v.ctime) > c.ctimeExpiry) { return true } + + if (c.atimeExpiry != time.Duration(0)) && (time.Now().Sub(v.atime) > c.atimeExpiry) { + return true + } + return false } -func MakeCache(expiryTime time.Duration) *Cache { +func MakeCache(ctimeExpiry, atimeExpiry time.Duration) *Cache { c := &Cache{ - cache: make(map[interface{}]Value), - expiryTime: expiryTime, + cache: make(map[interface{}]*Value), + ctimeExpiry: ctimeExpiry, + atimeExpiry: atimeExpiry, requestChannel: make(chan *request), } go c.service() - if expiryTime != time.Duration(0) { + if ctimeExpiry != time.Duration(0) || atimeExpiry != time.Duration(0) { go c.expiryService() } return c @@ -91,7 +97,7 @@ func (c *Cache) service() { if !ok { resp.error = fission.MakeError(fission.ErrorNotFound, fmt.Sprintf("key '%v' not found", req.key)) - } else if c.IsOld(&val) { + } else if c.IsOld(val) { resp.error = fission.MakeError(fission.ErrorNotFound, fmt.Sprintf("key '%v' expired (atime %v)", req.key, val.atime)) delete(c.cache, req.key) @@ -104,10 +110,17 @@ func (c *Cache) service() { req.responseChannel <- resp case SET: now := time.Now() - c.cache[req.key] = Value{ - value: req.value, - ctime: now, - atime: now, + if _, ok := c.cache[req.key]; ok { + val := c.cache[req.key] + val.atime = time.Now() + resp.existingValue = val.value + resp.error = errors.New("value already exists") + } else { + c.cache[req.key] = &Value{ + value: req.value, + ctime: now, + atime: now, + } } req.responseChannel <- resp case DELETE: @@ -115,7 +128,7 @@ func (c *Cache) service() { req.responseChannel <- resp case EXPIRE: for k, v := range c.cache { - if c.IsOld(&v) { + if c.IsOld(v) { delete(c.cache, k) } } @@ -123,7 +136,7 @@ func (c *Cache) service() { case COPY: resp.mapCopy = make(map[interface{}]interface{}) for k, v := range c.cache { - resp.mapCopy[k] = v + resp.mapCopy[k] = v.value } req.responseChannel <- resp default: @@ -145,7 +158,9 @@ func (c *Cache) Get(key interface{}) (interface{}, error) { return resp.value, resp.error } -func (c *Cache) Set(key interface{}, value interface{}) error { +// if key exists in the cache, the new value is NOT set; instead an +// error and the old value are returned +func (c *Cache) Set(key interface{}, value interface{}) (error, interface{}) { respChannel := make(chan *response) c.requestChannel <- &request{ requestType: SET, @@ -154,7 +169,7 @@ func (c *Cache) Set(key interface{}, value interface{}) error { responseChannel: respChannel, } resp := <-respChannel - return resp.error + return resp.error, resp.existingValue } func (c *Cache) Delete(key interface{}) error { From 8491aaf7fcda73bb2d4abfceb2914e5b480fd2bb Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 5 Nov 2016 21:42:39 -0700 Subject: [PATCH 12/14] Minor router update for cache interface change --- router/functionServiceMap.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/router/functionServiceMap.go b/router/functionServiceMap.go index cbc4f999..54cc8892 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -31,7 +31,7 @@ type functionServiceMap struct { func makeFunctionServiceMap(expiry time.Duration) *functionServiceMap { return &functionServiceMap{ - cache: cache.MakeCache(expiry), + cache: cache.MakeCache(expiry, 0), } } @@ -45,8 +45,9 @@ func (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) { } func (fmap *functionServiceMap) assign(f *fission.Metadata, serviceUrl *url.URL) { - err := fmap.cache.Set(*f, serviceUrl) + err, _ := fmap.cache.Set(*f, serviceUrl) if err != nil { log.Printf("error caching service url for function: %v", err) + // ignore error } } From 5650fca3fddcc7a4293d69cbe0669a436854671d Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 5 Nov 2016 21:42:54 -0700 Subject: [PATCH 13/14] Refactor caching in poolmgr Move separate caches to one cache -- functionServiceCache. It can be looked up by function, can update atime by address, and can be deleted by podname. This removes the other caches. Some of the concurrency logic is still a bit hairy; it might be better not to use fission.Cache. --- poolmgr/api.go | 71 ++++------ poolmgr/functionServiceCache.go | 196 +++++++++++++++++++++++++++ poolmgr/functionServiceCache_test.go | 72 ++++++++++ poolmgr/gp.go | 73 +++++----- poolmgr/gp_test.go | 93 ------------- poolmgr/gpm.go | 6 +- poolmgr/poolmgr.go | 6 +- 7 files changed, 340 insertions(+), 177 deletions(-) create mode 100644 poolmgr/functionServiceCache.go create mode 100644 poolmgr/functionServiceCache_test.go delete mode 100644 poolmgr/gp_test.go diff --git a/poolmgr/api.go b/poolmgr/api.go index da7ee496..c68dd94e 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -37,29 +37,29 @@ import ( type funcSvc struct { function *fission.Metadata // function this pod/service is for environment *fission.Environment // env it was obtained from - serviceName string // name of k8s svc - - reaped bool // if true, the pod has been deleted + address string // Host:Port or IP:Port that the service can be reached at. + podName string // pod name (within the function namespace) ctime time.Time atime time.Time } type API struct { - poolMgr *GenericPoolManager - functionEnv *cache.Cache // map[fission.Metadata]fission.Environment - functionService *cache.Cache // map[fission.Metadata]*funcSvc - urlFuncSvc *cache.Cache // map[string]*funcSvc - controller *controllerclient.Client + poolMgr *GenericPoolManager + functionEnv *cache.Cache // map[fission.Metadata]fission.Environment + fsCache *functionServiceCache + controller *controllerclient.Client + + //functionService *cache.Cache // map[fission.Metadata]*funcSvc + //urlFuncSvc *cache.Cache // map[string]*funcSvc } -func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client) *API { +func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client, fsCache *functionServiceCache) *API { return &API{ - poolMgr: gpm, - functionEnv: cache.MakeCache(0), - functionService: cache.MakeCache(time.Minute), - urlFuncSvc: cache.MakeCache(time.Minute), - controller: controller, + poolMgr: gpm, + functionEnv: cache.MakeCache(time.Minute, 0), + fsCache: fsCache, + controller: controller, } } @@ -121,53 +121,40 @@ func (api *API) getFunctionEnv(m *fission.Metadata) (*fission.Environment, error func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { // Check function -> svc map log.Printf("[%v] Checking for cached function service", m.Name) - result, err := api.functionService.Get(*m) + fsvc, err := api.fsCache.GetByFunction(m) if err == nil { - // Ok: return svc name - svc := result.(*funcSvc) - svc.atime = time.Now() - return svc.serviceName, nil + // Cached, return svc name + return fsvc.address, nil } - // None exists, so create a new funcSvc: + api.fsCache.Log() + + // None exists, so create a new funcSvc: log.Printf("[%v] No cached function service found, creating one", m.Name) - // from Func -> get Env + // from Func -> get Env log.Printf("[%v] getting environment for function", m.Name) env, err := api.getFunctionEnv(m) if err != nil { return "", err } - // from Env -> get GenericPool + // from Env -> get GenericPool log.Printf("[%v] getting generic pool for env", m.Name) pool, err := api.poolMgr.GetPool(env) if err != nil { return "", err } - // from GenericPool -> get one function container + // from GenericPool -> get one function container + // (this also adds to the cache) log.Printf("[%v] getting function service from pool", m.Name) funcSvc, err := pool.GetFuncSvc(m) if err != nil { return "", err } - // add to cache - err = api.functionService.Set(*m, funcSvc) - if err != nil { - // log and ignore error - log.Printf("Error caching function service: %v", err) - } - - // cache by svc hostname, for tapService() - err = api.urlFuncSvc.Set(funcSvc.serviceName, funcSvc) - if err != nil { - // log and ignore error - log.Printf("Error caching function service by name: %v", err) - } - - return funcSvc.serviceName, nil + return funcSvc.address, nil } // find funcSvc and update its atime @@ -180,16 +167,12 @@ func (api *API) tapService(w http.ResponseWriter, r *http.Request) { svcName := string(body) svcHost := strings.TrimPrefix(svcName, "http://") - log.Printf("tap svc: %v", svcHost) - - funcSvcI, err := api.urlFuncSvc.Get(svcHost) + err = api.fsCache.TouchByAddress(svcHost) if err != nil { + log.Printf("funcSvc tap error: %v", err) http.Error(w, "Not found", 404) return } - - (funcSvcI.(*funcSvc)).atime = time.Now() - w.WriteHeader(http.StatusOK) } diff --git a/poolmgr/functionServiceCache.go b/poolmgr/functionServiceCache.go new file mode 100644 index 00000000..74db5046 --- /dev/null +++ b/poolmgr/functionServiceCache.go @@ -0,0 +1,196 @@ +/* +Copyright 2016 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package poolmgr + +import ( + "log" + "time" + + "github.com/platform9/fission" + "github.com/platform9/fission/cache" +) + +type fscRequestType int + +const ( + TOUCH fscRequestType = iota + LISTOLD + LOG +) + +type ( + functionServiceCache struct { + byFunction *cache.Cache // function -> funcSvc : map[fission.Metadata]*funcSvc + byAddress *cache.Cache // address -> function : map[string]fission.Metadata + byPod *cache.Cache // podname -> function : map[string]fission.Metadata + + requestChannel chan *fscRequest + } + fscRequest struct { + requestType fscRequestType + address string + age time.Duration + responseChannel chan *fscResponse + } + fscResponse struct { + podNames []string + error + } +) + +func MakeFunctionServiceCache() *functionServiceCache { + fsc := &functionServiceCache{ + byFunction: cache.MakeCache(0, 0), + byAddress: cache.MakeCache(0, 0), + byPod: cache.MakeCache(0, 0), + requestChannel: make(chan *fscRequest), + } + go fsc.service() + return fsc +} + +func (fsc *functionServiceCache) service() { + for { + req := <-fsc.requestChannel + resp := &fscResponse{} + switch req.requestType { + case TOUCH: + // update atime for this function svc + resp.error = fsc._touchByAddress(req.address) + case LISTOLD: + // 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) + } + } + resp.podNames = pods + case LOG: + funcCopy := fsc.byFunction.Copy() + log.Printf("Cache has %v entries", len(funcCopy)) + for mI, fsvcI := range funcCopy { + m := mI.(fission.Metadata) + fsvc := fsvcI.(*funcSvc) + log.Printf("%v:%v\t%v", m.Name, m.Uid, fsvc.podName) + } + } + req.responseChannel <- resp + } +} + +func (fsc *functionServiceCache) GetByFunction(m *fission.Metadata) (*funcSvc, error) { + fsvcI, err := fsc.byFunction.Get(*m) + if err != nil { + return nil, err + } + // update atime + fsvc := fsvcI.(*funcSvc) + fsvc.atime = time.Now() + + fsvcCopy := *fsvc + return &fsvcCopy, nil +} + +func (fsc *functionServiceCache) Add(fsvc funcSvc) (error, *funcSvc) { + err, existing := fsc.byFunction.Set(*fsvc.function, &fsvc) + if err != nil { + if existing != nil { + f := existing.(*funcSvc) + err2 := fsc.TouchByAddress(f.address) + if err2 != nil { + return err2, nil + } + fCopy := *f + return err, &fCopy + } + return err, nil + } + now := time.Now() + fsvc.ctime = now + fsvc.atime = now + + err, _ = fsc.byAddress.Set(fsvc.address, *fsvc.function) + if err != nil { + log.Printf("error caching fsvc: %v", err) + return err, nil + } + err, _ = fsc.byPod.Set(fsvc.podName, *fsvc.function) + if err != nil { + log.Printf("error caching fsvc: %v", err) + return err, nil + } + return nil, nil +} + +func (fsc *functionServiceCache) TouchByAddress(address string) error { + responseChannel := make(chan *fscResponse) + fsc.requestChannel <- &fscRequest{ + requestType: TOUCH, + address: address, + responseChannel: responseChannel, + } + resp := <-responseChannel + return resp.error +} + +func (fsc *functionServiceCache) _touchByAddress(address string) error { + mI, err := fsc.byAddress.Get(address) + if err != nil { + return err + } + m := mI.(fission.Metadata) + fsvcI, err := fsc.byFunction.Get(m) + if err != nil { + return err + } + fsvc := fsvcI.(*funcSvc) + fsvc.atime = time.Now() + return nil +} + +func (fsc *functionServiceCache) DeleteByPod(podName string) error { + mI, err := fsc.byPod.Get(podName) + if err != nil { + return err + } + m := mI.(fission.Metadata) + fsvcI, err := fsc.byFunction.Get(m) + if err != nil { + return err + } + fsvc := fsvcI.(*funcSvc) + + fsc.byFunction.Delete(m) + fsc.byAddress.Delete(fsvc.address) + fsc.byPod.Delete(podName) + return nil +} + +func (fsc *functionServiceCache) Log() { + log.Printf("--- FunctionService Cache Contents") + responseChannel := make(chan *fscResponse) + fsc.requestChannel <- &fscRequest{ + requestType: LOG, + responseChannel: responseChannel, + } + <-responseChannel + log.Printf("--- FunctionService Cache Contents End") +} diff --git a/poolmgr/functionServiceCache_test.go b/poolmgr/functionServiceCache_test.go new file mode 100644 index 00000000..baaff0d7 --- /dev/null +++ b/poolmgr/functionServiceCache_test.go @@ -0,0 +1,72 @@ +package poolmgr + +import ( + "log" + "testing" + + "github.com/platform9/fission" + "time" +) + +func TestFunctionServiceCache(t *testing.T) { + fsc := MakeFunctionServiceCache() + if fsc == nil { + log.Panicf("error creating cache") + } + + var fsvc *funcSvc + now := time.Now() + + fsvc = &funcSvc{ + function: &fission.Metadata{ + Name: "foo", + Uid: "1212", + }, + environment: &fission.Environment{ + Metadata: fission.Metadata{ + Name: "foo-env", + Uid: "2323", + }, + RunContainerImageUrl: "fission/foo-env", + }, + address: "xxx", + podName: "yyy", + ctime: now, + atime: now, + } + err, _ := fsc.Add(*fsvc) + if err != nil { + fsc.Log() + log.Panicf("Failed to add fsvc: %v", err) + } + + f, err := fsc.GetByFunction(fsvc.function) + if err != nil { + fsc.Log() + log.Panicf("Failed to get fsvc: %v", err) + } + fsvc.atime = f.atime + fsvc.ctime = f.ctime + if *f != *fsvc { + fsc.Log() + log.Panicf("Incorrect fsvc \n(expected: %#v)\n (found: %#v)", fsvc, f) + } + + err = fsc.TouchByAddress(fsvc.address) + if err != nil { + fsc.Log() + log.Panicf("Failed to touch fsvc: %v", err) + } + + err = fsc.DeleteByPod(fsvc.podName) + if err != nil { + fsc.Log() + log.Panicf("Failed to delete fsvc: %v", err) + } + + _, err = fsc.GetByFunction(fsvc.function) + if err == nil { + fsc.Log() + log.Panicf("found fsvc while expecting empty cache", err) + } +} diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 5053172c..7e92b8c3 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -25,8 +25,10 @@ import ( "net" "net/http" "net/url" + "strings" "time" + "github.com/dchest/uniuri" "k8s.io/client-go/1.4/kubernetes" "k8s.io/client-go/1.4/pkg/api" "k8s.io/client-go/1.4/pkg/api/v1" @@ -39,17 +41,16 @@ import ( type ( GenericPool struct { - env *fission.Environment - replicas int32 // num containers - deployment *v1beta1.Deployment // kubernetes deployment - namespace string // namespace to keep our resources - podReadyTimeout time.Duration // timeout for generic pods to become ready - controllerUrl string - idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted - - useSvc bool // create service for - podFuncSvc *cache.Cache // map[pod.ObjectMeta.Name]*funcSvc - + env *fission.Environment + replicas int32 // num containers + deployment *v1beta1.Deployment // kubernetes deployment + namespace string // namespace to keep our resources + podReadyTimeout time.Duration // timeout for generic pods to become ready + controllerUrl string + idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted + fsCache *functionServiceCache // cache funcSvc's by function, address and podname + useSvc bool // create service + poolInstanceId string // small random string to uniquify pod names kubernetesClient *kubernetes.Clientset requestChannel chan *choosePodRequest } @@ -70,7 +71,8 @@ func MakeGenericPool( kubernetesClient *kubernetes.Clientset, env *fission.Environment, initialReplicas int32, - namespace string) (*GenericPool, error) { + namespace string, + fsCache *functionServiceCache) (*GenericPool, error) { log.Printf("Creating pool for environment %v", env.Metadata) // TODO: in general we need to provide the user a way to configure pools. Initial @@ -84,9 +86,10 @@ func MakeGenericPool( podReadyTimeout: 5 * time.Minute, // TODO make this an env param? controllerUrl: controllerUrl, idlePodReapTime: 3 * time.Minute, // TODO make this configurable + fsCache: fsCache, + poolInstanceId: uniuri.NewLen(8), - useSvc: false, - podFuncSvc: cache.MakeCache(0), + useSvc: false, } // create the pool @@ -271,7 +274,8 @@ func (gp *GenericPool) specializePod(metadata *fission.Metadata) (*v1.Pod, error // A pool is a deployment of generic containers for an env. This // creates the pool but doesn't wait for any pods to be ready. func (gp *GenericPool) createPool() error { - poolDeploymentName := fmt.Sprintf("%v-%v", gp.env.Metadata.Name, gp.env.Metadata.Uid) + poolDeploymentName := fmt.Sprintf("%v-%v-%v", + gp.env.Metadata.Name, gp.env.Metadata.Uid, strings.ToLower(gp.poolInstanceId)) podLabels := map[string]string{ "pool": poolDeploymentName, @@ -419,31 +423,28 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { fsvc := &funcSvc{ function: m, environment: gp.env, - serviceName: svcHost, + address: svcHost, + podName: pod.ObjectMeta.Name, ctime: time.Now(), atime: time.Now(), } - gp.podFuncSvc[pod.ObjectMeta.Name].Set(fsvc) + + err, existingFsvc := gp.fsCache.Add(*fsvc) + if err != nil { + // Some other thread beat us to it -- return the other thread's fsvc and clean up + // 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) + return existingFsvc, nil + } return fsvc, nil } -func (gp *GenericPool) idlePodReaper() { - for { - podmap := gp.podFuncSvc.Copy() - for podNameI, funcSvcI := range podmap { - podName := podNameI.(string) - funcSvc := funcSvcI.(*funcSvc) - lastAccessTime := funcSvc.atime - if time.Now().Sub(lastAccessTime) < gp.idlePodReapTime { - continue - } - - log.Printf("Reaping idle pod %v (last used at %v)", podName, lastAccessTime) - err := gp.kubernetesClient.Core().Pods(gp.namespace).Delete(podName) - if err != nil { - log.Printf("Error reaping pod: %v", err) - continue - } - } - } +func (gp *GenericPool) CleanupFunctionService(fsvc *funcSvc) { + // delete pod + // remove ourselves from fsCache +} + +func (gp *GenericPool) idlePodReaper() { } diff --git a/poolmgr/gp_test.go b/poolmgr/gp_test.go deleted file mode 100644 index 90b1c478..00000000 --- a/poolmgr/gp_test.go +++ /dev/null @@ -1,93 +0,0 @@ -package poolmgr - -import ( - "github.com/platform9/fission" - - "fmt" - "k8s.io/kubernetes/pkg/api" - "k8s.io/kubernetes/pkg/client/unversioned" - "k8s.io/kubernetes/pkg/client/unversioned/clientcmd" - "log" - "net/http" - "testing" -) - -func getKubeClient() *unversioned.Client { - loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() - configOverrides := &clientcmd.ConfigOverrides{} - kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) - config, err := kubeConfig.ClientConfig() - if err != nil { - panic("failed loading client config") - } - client := unversioned.NewOrDie(config) - return client -} - -type staticHandler struct { - resp string -} - -func (s *staticHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - w.Write([]byte(s.resp)) -} - -// staticHttpServer starts an http server at port and responds to any -// request with the given response. Use this to mock the controller -// raw function fetch HTTP endpoint. -func staticHttpServer(port int, response string) { - s := &staticHandler{resp: response} - log.Fatal(http.ListenAndServe(fmt.Sprintf(":%v", port), s)) -} - -func TestGenericPool(t *testing.T) { - namespace := "fission-test" - - client := getKubeClient() - - _, err := client.Namespaces().Create(&api.Namespace{ - ObjectMeta: api.ObjectMeta{ - Name: namespace, - Labels: map[string]string{}, - }, - }) - if err != nil { - log.Panicf("failed to create namespace: %v", err) - } - - // destroys everything in the namespace - defer client.Namespaces().Delete(namespace) - - env := &fission.Environment{ - Metadata: fission.Metadata{ - Name: "test-env", - Uid: "", - }, - RunContainerImageUrl: "fission/testing", - } - - gp, err := MakeGenericPool(client, env, 3, namespace) - if err != nil { - log.Panicf("failed to make generic pool: %v", err) - } - log.Printf("Pool created") - - // test specialization - - testFunc := ` -module.exports = function (context, callback) { - callback(200, "Hello, world!"); -} -` - go staticHttpServer(2222, testFunc) - - m := fission.Metadata{ - Name: "foo", - Uid: "xxx-yyy", - } - fsvc, err := gp.GetFuncSvc(&m) - if err != nil { - log.Fatalf("Error getting function svc: %v", err) - } - log.Printf("fsvc: %v", fsvc) -} diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 48231c1b..263c884b 100644 --- a/poolmgr/gpm.go +++ b/poolmgr/gpm.go @@ -33,6 +33,7 @@ type ( namespace string controllerUrl string controllerClient *client.Client + fsCache *functionServiceCache requestChannel chan *request } @@ -46,13 +47,14 @@ type ( } ) -func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.Clientset, namespace string) *GenericPoolManager { +func MakeGenericPoolManager(controllerUrl string, kubernetesClient *kubernetes.Clientset, namespace string, fsCache *functionServiceCache) *GenericPoolManager { gpm := &GenericPoolManager{ pools: make(map[fission.Environment]*GenericPool), kubernetesClient: kubernetesClient, namespace: namespace, controllerUrl: controllerUrl, controllerClient: client.MakeClient(controllerUrl), + fsCache: fsCache, requestChannel: make(chan *request), } go gpm.service() @@ -68,7 +70,7 @@ func (gpm *GenericPoolManager) service() { var err error pool, ok := gpm.pools[*req.env] if !ok { - pool, err = MakeGenericPool(gpm.controllerUrl, gpm.kubernetesClient, req.env, 3, gpm.namespace) + pool, err = MakeGenericPool(gpm.controllerUrl, gpm.kubernetesClient, req.env, 3, gpm.namespace, gpm.fsCache) if err != nil { req.responseChannel <- &response{error: err} continue diff --git a/poolmgr/poolmgr.go b/poolmgr/poolmgr.go index 6ac3ba4b..876292c8 100644 --- a/poolmgr/poolmgr.go +++ b/poolmgr/poolmgr.go @@ -56,9 +56,11 @@ func StartPoolmgr(controllerUrl string, namespace string, port int) error { return err } - gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace) + fsCache := MakeFunctionServiceCache() + + gpm := MakeGenericPoolManager(controllerUrl, kubernetesClient, namespace, fsCache) + api := MakeAPI(gpm, controllerClient, fsCache) - api := MakeAPI(gpm, controllerClient) go api.Serve(port) return nil From c7692978f9f0418c088ee19b7f763487385f5bfa Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 5 Nov 2016 23:10:52 -0700 Subject: [PATCH 14/14] 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) + } + } + } }