Merge pull request #20 from platform9/reap-idle-pods

Reap idle pods

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{
+			if _, ok := c.cache[req.key]; ok {
-				value: req.value,
+				val := c.cache[req.key]
-				ctime: now,
+				val.atime = time.Now()
-				atime: 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,
+		}
+	}
+}

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 {
+	checkErr(err)
-		log.Panicf("error: %v", err)
+	err = c.Set("p", "q")
-	}
+	checkErr(err)
 
 	val, err := c.Get("a")
-	if err != nil {
+	checkErr(err)
-		log.Panicf("error: %v", err)
-	}
 	if val != "b" {
 		log.Panicf("value %v", val)
 	}
 
-	err = c.Delete("a")
+	cc := c.Copy()
-	if err != nil {
+	if len(cc) != 2 {
-		log.Panicf("error: %v", err)
+		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")
 	}
 }

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
+	poolMgr     *GenericPoolManager
-	functionEnv     *cache.Cache // map[fission.Metadata]fission.Environment
+	functionEnv *cache.Cache // map[fission.Metadata]fission.Environment
-	functionService *cache.Cache // map[fission.Metadata]funcSvc
+	fsCache     *functionServiceCache
-	controller      *controllerclient.Client
+	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,
+		poolMgr:     gpm,
-		functionEnv:     cache.MakeCache(),
+		functionEnv: cache.MakeCache(time.Minute, 0),
-		functionService: cache.MakeCache(),
+		fsCache:     fsCache,
-		controller:      controller,
+		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
+		// Cached, return svc name
-		svc := result.(*funcSvc)
+		return fsvc.address, nil
-		return svc.serviceName, 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
+	return funcSvc.address, nil
-	err = api.functionService.Set(m, funcSvc)
+}
-	if err != nil {
-		// log and ignore error
-		log.Printf("Error saving function service: %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://")
+
+	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)

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
+}

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")
+}

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)
+	}
+}

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
+		env              *fission.Environment
-		replicas        int32               // num containers
+		replicas         int32               // num containers
-		deployment      *v1beta1.Deployment // kubernetes deployment
+		deployment       *v1beta1.Deployment // kubernetes deployment
-		namespace       string              // namespace to keep our resources
+		namespace        string              // namespace to keep our resources
-		podReadyTimeout time.Duration       // timeout for generic pods to become ready
+		podReadyTimeout  time.Duration       // timeout for generic pods to become ready
-		controllerUrl   string
+		controllerUrl    string
-		useSvc          bool
+		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)
+			}
+		}
+	}
+}

poolmgr/gp_test.go

@@ -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)
-}

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)
+			}
 		}
 	}
 }

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

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.

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}

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
+	cache *cache.Cache // map[fission.Metadata]*url.URL
-	svc               map[fission.Metadata]functionServiceMapEntry
-	currentGeneration uint64
-	requestChannel    chan *functionServiceMapRequest
 }
 
-func makeFunctionServiceMap() *functionServiceMap {
+func makeFunctionServiceMap(expiry time.Duration) *functionServiceMap {
-	fmap := &functionServiceMap{}
+	return &functionServiceMap{
-	fmap.requestChannel = make(chan *functionServiceMapRequest)
+		cache: cache.MakeCache(expiry, 0),
-	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 (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) {
-	respChannel := make(chan functionServiceMapResponse)
+	item, err := fmap.cache.Get(*f)
-	fmap.requestChannel <- &functionServiceMapRequest{Function: *f, requestType: LOOKUP, responseChannel: respChannel}
+	if err != nil {
-	resp := <-respChannel
+		return nil, err
-	if resp.error != nil {
-		return nil, resp.error
-	} else {
-		return &resp.serviceUrl, nil
 	}
+	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}
+	err, _ := fmap.cache.Set(*f, serviceUrl)
-}
+	if err != nil {
-
+		log.Printf("error caching service url for function: %v", err)
-func (fmap *functionServiceMap) nextGen() {
+		// ignore error
-	fmap.requestChannel <- &functionServiceMapRequest{requestType: NEXT_GEN}
+	}
-}
-
-func (fmap *functionServiceMap) sweep() {
-	fmap.requestChannel <- &functionServiceMapRequest{requestType: SWEEP}
 }

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 {

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)
 

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"