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