diff --git a/cache/cache.go b/cache/cache.go index 87c24f6d..46ce3969 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -19,6 +19,7 @@ package cache import ( "time" + "errors" "fmt" "github.com/platform9/fission" ) @@ -29,6 +30,8 @@ const ( GET requestType = iota SET DELETE + EXPIRE + COPY ) type ( @@ -38,7 +41,9 @@ type ( value interface{} } Cache struct { - cache map[interface{}]Value + cache map[interface{}]*Value + ctimeExpiry time.Duration + atimeExpiry time.Duration requestChannel chan *request } @@ -50,16 +55,35 @@ type ( } response struct { error - value interface{} + existingValue interface{} + mapCopy map[interface{}]interface{} + value interface{} } ) -func MakeCache() *Cache { +func (c *Cache) IsOld(v *Value) bool { + 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(ctimeExpiry, atimeExpiry time.Duration) *Cache { c := &Cache{ - cache: make(map[interface{}]Value), + cache: make(map[interface{}]*Value), + ctimeExpiry: ctimeExpiry, + atimeExpiry: atimeExpiry, requestChannel: make(chan *request), } go c.service() + if ctimeExpiry != time.Duration(0) || atimeExpiry != time.Duration(0) { + go c.expiryService() + } return c } @@ -73,23 +97,48 @@ 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() - 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: 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 + case COPY: + resp.mapCopy = make(map[interface{}]interface{}) + for k, v := range c.cache { + resp.mapCopy[k] = v.value + } + req.responseChannel <- resp default: resp.error = fission.MakeError(fission.ErrorInvalidArgument, fmt.Sprintf("invalid request type: %v", req.requestType)) @@ -109,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, @@ -118,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 { @@ -131,3 +182,22 @@ func (c *Cache) Delete(key interface{}) error { resp := <-respChannel 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) + c.requestChannel <- &request{ + requestType: EXPIRE, + } + } +} diff --git a/cache/cache_test.go b/cache/cache_test.go index 52ba9657..e307feed 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -18,30 +18,46 @@ 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) + err = c.Set("p", "q") + 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) + cc := c.Copy() + if len(cc) != 2 { + log.Panicf("expected 2 items") } + err = c.Delete("a") + 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") } } diff --git a/poolmgr/api.go b/poolmgr/api.go index 9206cc92..d4fc4ac3 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -23,6 +23,7 @@ import ( "log" "net/http" "os" + "strings" "time" "github.com/gorilla/handlers" @@ -34,27 +35,31 @@ 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 + 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 - 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(), - functionService: cache.MakeCache(), - controller: controller, + poolMgr: gpm, + functionEnv: cache.MakeCache(time.Minute, 0), + fsCache: fsCache, + controller: controller, } } @@ -87,7 +92,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 @@ -108,7 +113,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 } @@ -116,50 +121,63 @@ 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) - return svc.serviceName, nil + // Cached, return svc name + return fsvc.address, nil } - // None exists, so create a new funcSvc: + // 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 saving function service: %v", err) - } + return funcSvc.address, nil +} - 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://") + + err = api.fsCache.TouchByAddress(svcHost) + if err != nil { + log.Printf("funcSvc tap error: %v", err) + http.Error(w, "Not found", 404) + return + } + 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/poolmgr/functionServiceCache.go b/poolmgr/functionServiceCache.go new file mode 100644 index 00000000..c3a248d6 --- /dev/null +++ b/poolmgr/functionServiceCache.go @@ -0,0 +1,236 @@ +/* +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 + DELETE_BY_POD +) + +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 + podName string + age time.Duration + responseChannel chan *fscResponse + } + fscResponse struct { + podNames []string + deleted bool + 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, 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 + 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) + } + case DELETE_BY_POD: + resp.deleted, resp.error = fsc._deleteByPod(req.podName, req.age) + } + 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, 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 false, err + } + m := mI.(fission.Metadata) + fsvcI, err := fsc.byFunction.Get(m) + if err != nil { + 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 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() { + 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 2a73a164..af4f1aa0 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,14 +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 - useSvc bool - + 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 } @@ -67,18 +71,25 @@ 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 + // 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 + fsCache: fsCache, + poolInstanceId: uniuri.NewLen(8), + + useSvc: false, } // create the pool @@ -95,6 +106,9 @@ func MakeGenericPool( } go gp.choosePodService() + + go gp.idlePodReaper() + return gp, nil } @@ -260,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, @@ -408,9 +423,59 @@ 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(), } + + 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.podName) + return existingFsvc, nil + } return fsvc, nil } + +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 + 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) + } + } + } +} 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 5c4213f1..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 @@ -108,7 +110,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) + } } } } 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 diff --git a/router/functionHandler.go b/router/functionHandler.go index ca7c88ff..9dc13a27 100644 --- a/router/functionHandler.go +++ b/router/functionHandler.go @@ -80,7 +80,19 @@ func (rrt RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er return http.DefaultTransport.RoundTrip(req) } +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()) + } +} + 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 +111,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. 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.go b/router/functionServiceMap.go index 671c19bc..54cc8892 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -17,97 +17,37 @@ limitations under the License. package router import ( - "errors" "log" "net/url" + "time" "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 + cache *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") - } +func makeFunctionServiceMap(expiry time.Duration) *functionServiceMap { + return &functionServiceMap{ + cache: cache.MakeCache(expiry, 0), } } 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.cache.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} + err, _ := fmap.cache.Set(*f, serviceUrl) + if err != nil { + log.Printf("error caching service url for function: %v", err) + // ignore error + } } 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.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) 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"