Executor abstraction (#384)

This change adds a layer of abstraction over poolmgr. Poolmgr is now just one of the ways to turn a function into a service; other implementations will be added. The executor abstraction is a uniform API over all these implementations.

* Executor layer added on top of pool manager

* Removed the external server for executor

* Minor changes to keep existing semantics as much possible

* Separating the executor vs. poolmgr backend functionality and associated data members

* Executor logic separated from Poolmgr backend completely, placeholder for new backend

* Changed references to poolmgr in tests

* Moved poolmgr to it's package, as a side effect moved Cache to its's package (was causing cyclical dependency) and had to make some data structures exposed outside package

* Rebased from master and changed references to tpr -> crd

* Executor layer added on top of pool manager

* Executor logic separated from Poolmgr backend completely, placeholder for new backend

* Changed podName to a generic objectReference in fscache (#391)

Changed podName to a generic objectReference in function service cache implementation.

* Moved poolmgr to it's package, as a side effect moved Cache to its's package (was causing cyclical dependency) and had to make some data structures exposed outside package

* Rebased from master and changed references to tpr -> crd

* Merged from master with latest changes

* Removed stale executor service & deployment from previous merge

* Addressed review comments, still testing some areas

executor/api.go

@@ -0,0 +1,105 @@
+/*
+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 executor
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+
+	"github.com/gorilla/handlers"
+	"github.com/gorilla/mux"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/fission/fission"
+)
+
+func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "Failed to read request", 500)
+		return
+	}
+
+	// get function metadata
+	m := metav1.ObjectMeta{}
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		http.Error(w, "Failed to parse request", 400)
+		return
+	}
+
+	serviceName, err := executor.getServiceForFunction(&m)
+	if err != nil {
+		code, msg := fission.GetHTTPError(err)
+		log.Printf("Error: %v: %v", code, msg)
+		http.Error(w, msg, code)
+		return
+	}
+
+	w.Write([]byte(serviceName))
+}
+
+func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, error) {
+	// Check function -> svc cache
+	log.Printf("[%v] Checking for cached function service", m.Name)
+	fsvc, err := executor.fsCache.GetByFunction(m)
+	if err == nil {
+		// Cached, return svc address
+		return fsvc.Address, nil
+	}
+
+	respChan := make(chan *createFuncServiceResponse)
+	executor.requestChan <- &createFuncServiceRequest{
+		funcMeta: m,
+		respChan: respChan,
+	}
+	resp := <-respChan
+	return resp.funcSvc.Address, resp.err
+}
+
+// find funcSvc and update its atime
+func (executor *Executor) 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 = executor.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 (executor *Executor) Serve(port int) {
+	r := mux.NewRouter()
+	r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST")
+	r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST")
+	address := fmt.Sprintf(":%v", port)
+	log.Printf("starting executor at port %v", port)
+	log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r)))
+}

executor/client/client.go

@@ -0,0 +1,114 @@
+/*
+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 client
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/fission/fission"
+)
+
+type Client struct {
+	executorUrl string
+	tappedByUrl map[string]bool
+	requestChan chan string
+}
+
+func MakeClient(executorUrl string) *Client {
+	c := &Client{
+		executorUrl: strings.TrimSuffix(executorUrl, "/"),
+		tappedByUrl: make(map[string]bool),
+		requestChan: make(chan string),
+	}
+	go c.service()
+	return c
+}
+
+func (c *Client) GetServiceForFunction(metadata *metav1.ObjectMeta) (string, error) {
+	executorUrl := c.executorUrl + "/v2/getServiceForFunction"
+
+	body, err := json.Marshal(metadata)
+	if err != nil {
+		return "", err
+	}
+
+	resp, err := http.Post(executorUrl, "application/json", bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return "", fission.MakeErrorFromHTTP(resp)
+	}
+
+	svcName, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	return string(svcName), nil
+}
+
+func (c *Client) service() {
+	ticker := time.NewTicker(time.Second * 5)
+	for {
+		select {
+		case serviceUrl := <-c.requestChan:
+			c.tappedByUrl[serviceUrl] = true
+		case <-ticker.C:
+			urls := c.tappedByUrl
+			c.tappedByUrl = make(map[string]bool)
+			if len(urls) > 0 {
+				go func() {
+					for u := range c.tappedByUrl {
+						c._tapService(u)
+					}
+					log.Printf("Tapped %v services in batch", len(urls))
+				}()
+				log.Printf("Tapped %v services in batch", len(urls))
+			}
+		}
+	}
+}
+
+func (c *Client) TapService(serviceUrl *url.URL) {
+	c.requestChan <- serviceUrl.String()
+}
+
+func (c *Client) _tapService(serviceUrlStr string) error {
+	executorUrl := c.executorUrl + "/v2/tapService"
+
+	resp, err := http.Post(executorUrl, "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
+}

executor/deploymgr/newdeploy.go

@@ -0,0 +1,22 @@
+/*
+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 deploymgr
+
+func GetFuncSvc() (*funcSvc, error) {
+
+	return nil, nil
+}

executor/executor.go

@@ -0,0 +1,195 @@
+/*
+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 executor
+
+import (
+	"log"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/dchest/uniuri"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/fission/fission/cache"
+	"github.com/fission/fission/crd"
+	"github.com/fission/fission/executor/fscache"
+	"github.com/fission/fission/executor/poolmgr"
+)
+
+type (
+	Executor struct {
+		gpm           *poolmgr.GenericPoolManager
+		functionEnv   *cache.Cache
+		fissionClient *crd.FissionClient
+		fsCache       *fscache.FunctionServiceCache
+
+		requestChan chan *createFuncServiceRequest
+		fsCreateWg  map[string]*sync.WaitGroup
+	}
+	createFuncServiceRequest struct {
+		funcMeta *metav1.ObjectMeta
+		respChan chan *createFuncServiceResponse
+	}
+
+	createFuncServiceResponse struct {
+		funcSvc *fscache.FuncSvc
+		err     error
+	}
+)
+
+func MakeExecutor(gpm *poolmgr.GenericPoolManager, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor {
+	executor := &Executor{
+		gpm:           gpm,
+		functionEnv:   cache.MakeCache(10*time.Second, 0),
+		fissionClient: fissionClient,
+		fsCache:       fsCache,
+
+		requestChan: make(chan *createFuncServiceRequest),
+		fsCreateWg:  make(map[string]*sync.WaitGroup),
+	}
+	go executor.serveCreateFuncServices()
+	return executor
+}
+
+// All non-cached function service requests go through this goroutine
+// serially. It parallelizes requests for different functions, and
+// ensures that for a given function, only one request causes a pod to
+// get specialized. In other words, it ensures that when there's an
+// ongoing request for a certain function, all other requests wait for
+// that request to complete.
+func (executor *Executor) serveCreateFuncServices() {
+	for {
+		req := <-executor.requestChan
+		m := req.funcMeta
+
+		// Cache miss -- is this first one to request the func?
+		wg, found := executor.fsCreateWg[crd.CacheKey(m)]
+		if !found {
+			// create a waitgroup for other requests for
+			// the same function to wait on
+			wg := &sync.WaitGroup{}
+			wg.Add(1)
+			executor.fsCreateWg[crd.CacheKey(m)] = wg
+
+			// launch a goroutine for each request, to parallelize
+			// the specialization of different functions
+			go func() {
+				fsvc, err := executor.createServiceForFunction(m)
+				req.respChan <- &createFuncServiceResponse{
+					funcSvc: fsvc,
+					err:     err,
+				}
+				delete(executor.fsCreateWg, crd.CacheKey(m))
+				wg.Done()
+			}()
+		} else {
+			// There's an existing request for this function, wait for it to finish
+			go func() {
+				log.Printf("Waiting for concurrent request for the same function: %v", m)
+				wg.Wait()
+
+				// get the function service from the cache
+				fsvc, err := executor.fsCache.GetByFunction(m)
+				req.respChan <- &createFuncServiceResponse{
+					funcSvc: fsvc,
+					err:     err,
+				}
+			}()
+		}
+	}
+}
+
+func (executor *Executor) createServiceForFunction(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
+	log.Printf("[%v] No cached function service found, creating one", m.Name)
+
+	env, err := executor.getFunctionEnv(m)
+	if err != nil {
+		return nil, err
+	}
+	// Appropriate backend handles the service creation
+	backend := os.Getenv("EXECUTOR_BACKEND")
+	switch backend {
+	case "DEPLOY":
+		return nil, nil
+	default:
+		pool, err := executor.gpm.GetPool(env)
+		if err != nil {
+			return nil, err
+		}
+		// from GenericPool -> get one function container
+		// (this also adds to the cache)
+		log.Printf("[%v] getting function service from pool", m.Name)
+		fsvc, err := pool.GetFuncSvc(m)
+		if err != nil {
+			return nil, err
+		}
+		return fsvc, nil
+	}
+}
+
+func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment, error) {
+	var env *crd.Environment
+
+	// Cached ?
+	result, err := executor.functionEnv.Get(crd.CacheKey(m))
+	if err == nil {
+		env = result.(*crd.Environment)
+		return env, nil
+	}
+
+	// Cache miss -- get func from controller
+	f, err := executor.fissionClient.Functions(m.Namespace).Get(m.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get env from metadata
+	log.Printf("[%v] getting env", m)
+	env, err = executor.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	// cache for future lookups
+	executor.functionEnv.Set(crd.CacheKey(m), env)
+
+	return env, nil
+}
+
+// StartExecutor Starts executor and the backend components that executor uses such as Poolmgr,
+// deploymgr and potential future backends
+func StartExecutor(fissionNamespace string, functionNamespace string, port int) error {
+	fissionClient, kubernetesClient, _, err := crd.MakeFissionClient()
+	if err != nil {
+		log.Printf("Failed to get kubernetes client: %v", err)
+		return err
+	}
+
+	instanceID := uniuri.NewLen(8)
+	poolmgr.CleanupOldPoolmgrResources(kubernetesClient, functionNamespace, instanceID)
+
+	fsCache := fscache.MakeFunctionServiceCache()
+	gpm := poolmgr.MakeGenericPoolManager(
+		fissionClient, kubernetesClient, fissionNamespace,
+		functionNamespace, fsCache, instanceID)
+
+	api := MakeExecutor(gpm, fissionClient, fsCache)
+	go api.Serve(port)
+
+	return nil
+}

executor/executor_test.go

@@ -0,0 +1,263 @@
+//
+// This test depends on several env vars:
+//
+//   KUBECONFIG has to point at a kube config with a cluster. The test
+//      will use the default context from that config. Be careful,
+//      don't point this at your production environment. The test is
+//      skipped if KUBECONFIG is undefined.
+//
+//   TEST_SPECIALIZE_URL
+//   TEST_FETCHER_URL
+//      These need to point at <node ip>:30001 and <node ip>:30002,
+//      where <node ip> is the address of any node in the test
+//      cluster.
+//
+//   FETCHER_IMAGE
+//      Optional. Set this to a fetcher image; otherwise uses the
+//      default.
+//
+
+// Here's how I run this on my setup, with minikube:
+// TEST_SPECIALIZE_URL=http://192.168.99.100:30002/specialize TEST_FETCHER_URL=http://192.168.99.100:30001 FETCHER_IMAGE=minikube/fetcher:testing KUBECONFIG=/Users/soam/.kube/config go test -v .
+
+package executor
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"math/rand"
+	"net/http"
+	"os"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/kubernetes"
+	apiv1 "k8s.io/client-go/pkg/api/v1"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
+	"github.com/fission/fission/executor/client"
+)
+
+// return the number of pods in the given namespace matching the given labels
+func countPods(kubeClient *kubernetes.Clientset, ns string, labelz map[string]string) int {
+	pods, err := kubeClient.Pods(ns).List(metav1.ListOptions{
+		LabelSelector: labels.Set(labelz).AsSelector().String(),
+	})
+	if err != nil {
+		log.Panicf("Failed to list pods: %v", err)
+	}
+	return len(pods.Items)
+}
+
+func createTestNamespace(kubeClient *kubernetes.Clientset, ns string) {
+	_, err := kubeClient.Namespaces().Create(&apiv1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: ns,
+		},
+	})
+	if err != nil {
+		log.Panicf("failed to create ns %v: %v", ns, err)
+	}
+	log.Printf("Created namespace %v", ns)
+}
+
+// create a nodeport service
+func createSvc(kubeClient *kubernetes.Clientset, ns string, name string, targetPort int, nodePort int32, labels map[string]string) *apiv1.Service {
+	svc, err := kubeClient.Services(ns).Create(&apiv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: apiv1.ServiceSpec{
+			Type: apiv1.ServiceTypeNodePort,
+			Ports: []apiv1.ServicePort{
+				{
+					Protocol:   apiv1.ProtocolTCP,
+					Port:       80,
+					TargetPort: intstr.FromInt(targetPort),
+					NodePort:   nodePort,
+				},
+			},
+			Selector: labels,
+		},
+	})
+	if err != nil {
+		log.Panicf("Failed to create svc: %v", err)
+	}
+	return svc
+}
+
+func httpGet(url string) string {
+	resp, err := http.Get(url)
+	if err != nil {
+		log.Panicf("HTTP Get failed: URL %v: %v", url, err)
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		log.Panicf("HTTP Get failed to read body: URL %v: %v", url, err)
+	}
+	return string(body)
+}
+
+func TestExecutor(t *testing.T) {
+	// run in a random namespace so we can have concurrent tests
+	// on a given cluster
+	rand.Seed(time.Now().UTC().UnixNano())
+	testId := rand.Intn(999)
+	fissionNs := fmt.Sprintf("test-%v", testId)
+	functionNs := fmt.Sprintf("test-function-%v", testId)
+
+	// skip test if no cluster available for testing
+	kubeconfig := os.Getenv("KUBECONFIG")
+	if len(kubeconfig) == 0 {
+		t.Skip("Skipping test, no kubernetes cluster")
+		return
+	}
+
+	// connect to k8s
+	// and get CRD client
+	fissionClient, kubeClient, apiExtClient, err := crd.MakeFissionClient()
+	if err != nil {
+		log.Panicf("failed to connect: %v", err)
+	}
+
+	// create the test's namespaces
+	createTestNamespace(kubeClient, fissionNs)
+	defer kubeClient.Namespaces().Delete(fissionNs, nil)
+
+	createTestNamespace(kubeClient, functionNs)
+	defer kubeClient.Namespaces().Delete(functionNs, nil)
+
+	// make sure CRD types exist on cluster
+	err = crd.EnsureFissionCRDs(apiExtClient)
+	if err != nil {
+		log.Panicf("failed to ensure crds: %v", err)
+	}
+	fissionClient.WaitForCRDs()
+
+	// create an env on the cluster
+	env, err := fissionClient.Environments(fissionNs).Create(&crd.Environment{
+		Metadata: metav1.ObjectMeta{
+			Name:      "nodejs",
+			Namespace: fissionNs,
+		},
+		Spec: fission.EnvironmentSpec{
+			Version: 1,
+			Runtime: fission.Runtime{
+				Image: "fission/node-env",
+			},
+			Builder: fission.Builder{},
+		},
+	})
+	if err != nil {
+		log.Panicf("failed to create env: %v", err)
+	}
+
+	// create poolmgr
+	port := 9999
+	err = StartExecutor(fissionNs, functionNs, port)
+	if err != nil {
+		log.Panicf("failed to start poolmgr: %v", err)
+	}
+
+	// connect poolmgr client
+	poolmgrClient := client.MakeClient(fmt.Sprintf("http://localhost:%v", port))
+
+	// Wait for pool to be created (we don't actually need to do
+	// this, since the API should do the right thing in any case).
+	// waitForPool(functionNs, "nodejs")
+	time.Sleep(6 * time.Second)
+
+	envRef := fission.EnvironmentReference{
+		Namespace: env.Metadata.Namespace,
+		Name:      env.Metadata.Name,
+	}
+
+	deployment := fission.Archive{
+		Type:    fission.ArchiveTypeLiteral,
+		Literal: []byte(`module.exports = async function(context) { return { status: 200, body: "Hello, world!\n" }; }`),
+	}
+
+	// create a package
+	p := &crd.Package{
+		Metadata: metav1.ObjectMeta{
+			Name:      "hello",
+			Namespace: fissionNs,
+		},
+		Spec: fission.PackageSpec{
+			Environment: envRef,
+			Deployment:  deployment,
+		},
+	}
+	p, err = fissionClient.Packages(fissionNs).Create(p)
+	if err != nil {
+		log.Panicf("failed to create package: %v", err)
+	}
+
+	// create a function
+	f := &crd.Function{
+		Metadata: metav1.ObjectMeta{
+			Name:      "hello",
+			Namespace: fissionNs,
+		},
+		Spec: fission.FunctionSpec{
+			Environment: envRef,
+			Package: fission.FunctionPackageRef{
+				PackageRef: fission.PackageRef{
+					Namespace:       p.Metadata.Namespace,
+					Name:            p.Metadata.Name,
+					ResourceVersion: p.Metadata.ResourceVersion,
+				},
+			},
+		},
+	}
+	_, err = fissionClient.Functions(fissionNs).Create(f)
+	if err != nil {
+		log.Panicf("failed to create function: %v", err)
+	}
+
+	// create a service to call fetcher and the env container
+	labels := map[string]string{"functionName": f.Metadata.Name}
+	var fetcherPort int32 = 30001
+	fetcherSvc := createSvc(kubeClient, functionNs, fmt.Sprintf("%v-%v", f.Metadata.Name, "fetcher"), 8000, fetcherPort, labels)
+	defer kubeClient.Services(functionNs).Delete(fetcherSvc.ObjectMeta.Name, nil)
+
+	var funcSvcPort int32 = 30002
+	functionSvc := createSvc(kubeClient, functionNs, f.Metadata.Name, 8888, funcSvcPort, labels)
+	defer kubeClient.Services(functionNs).Delete(functionSvc.ObjectMeta.Name, nil)
+
+	// the main test: get a service for a given function
+	t1 := time.Now()
+	svc, err := poolmgrClient.GetServiceForFunction(&f.Metadata)
+	if err != nil {
+		log.Panicf("failed to get func svc: %v", err)
+	}
+	log.Printf("svc for function created at: %v (in %v)", svc, time.Now().Sub(t1))
+
+	// ensure that a pod with the label functionName=f.Metadata.Name exists
+	podCount := countPods(kubeClient, functionNs, map[string]string{"functionName": f.Metadata.Name})
+	if podCount != 1 {
+		log.Panicf("expected 1 function pod, found %v", podCount)
+	}
+
+	// call the service to ensure it works
+
+	// wait for a bit
+
+	// tap service to simulate calling it again
+
+	// make sure the same pod is still there
+
+	// wait for idleTimeout to ensure the pod is removed
+
+	// remove env
+
+	// wait for pool to be destroyed
+
+	// that's it
+}

executor/fscache/functionServiceCache.go

@@ -0,0 +1,277 @@
+/*
+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 fscache
+
+import (
+	"log"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/pkg/api"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/cache"
+	"github.com/fission/fission/crd"
+)
+
+type fscRequestType int
+type backendType int
+
+const (
+	TOUCH fscRequestType = iota
+	LISTOLD
+	LOG
+	DELETE_BY_OBJECT
+)
+
+const (
+	POOLMGR backendType = iota
+	NEWDEPLOY
+)
+
+type (
+	FuncSvc struct {
+		Function         *metav1.ObjectMeta  // function this pod/service is for
+		Environment      *crd.Environment    // function's environment
+		Address          string              // Host:Port or IP:Port that the function's service can be reached at.
+		KubernetesObject api.ObjectReference // Kubernetes Object (within the function namespace)
+		Backend          backendType
+
+		Ctime time.Time
+		Atime time.Time
+	}
+
+	FunctionServiceCache struct {
+		byFunction   *cache.Cache // function-key -> funcSvc  : map[string]*funcSvc
+		byAddress    *cache.Cache // address      -> function : map[string]metav1.ObjectMeta
+		byKubeObject *cache.Cache // obj          -> function : map[api.ObjectReference]metav1.ObjectMeta
+
+		requestChannel chan *fscRequest
+	}
+	fscRequest struct {
+		requestType      fscRequestType
+		address          string
+		kubernetesObject api.ObjectReference
+		age              time.Duration
+		env              *metav1.ObjectMeta // used for ListOld
+		responseChannel  chan *fscResponse
+	}
+	fscResponse struct {
+		objects []api.ObjectReference
+		deleted bool
+		error
+	}
+)
+
+func MakeFunctionServiceCache() *FunctionServiceCache {
+	fsc := &FunctionServiceCache{
+		byFunction:     cache.MakeCache(0, 0),
+		byAddress:      cache.MakeCache(0, 0),
+		byKubeObject:   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
+			byKubeObjectCopy := fsc.byKubeObject.Copy()
+			kubeObjects := make([]api.ObjectReference, 0)
+			for objI, mI := range byKubeObjectCopy {
+				m := mI.(metav1.ObjectMeta)
+				fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
+				if err != nil {
+					resp.error = err
+				} else {
+					fsvc := fsvcI.(*FuncSvc)
+					if fsvc.Environment.Metadata.UID == req.env.UID &&
+						time.Now().Sub(fsvc.Atime) > req.age {
+
+						obj := objI.(api.ObjectReference)
+						kubeObjects = append(kubeObjects, obj)
+					}
+				}
+			}
+			resp.objects = kubeObjects
+		case LOG:
+			funcCopy := fsc.byFunction.Copy()
+			log.Printf("Cache has %v entries", len(funcCopy))
+			for key, fsvcI := range funcCopy {
+				fsvc := fsvcI.(*FuncSvc)
+				log.Printf("%v\t%v\t%v", key, fsvc.KubernetesObject.Kind, fsvc.KubernetesObject.Name)
+			}
+		case DELETE_BY_OBJECT:
+			resp.deleted, resp.error = fsc._deleteByKubeObject(req.kubernetesObject, req.age)
+		}
+		req.responseChannel <- resp
+	}
+}
+
+func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) {
+	key := crd.CacheKey(m)
+
+	fsvcI, err := fsc.byFunction.Get(key)
+	if err != nil {
+		return nil, err
+	}
+
+	// update atime
+	fsvc := fsvcI.(*FuncSvc)
+	fsvc.Atime = time.Now()
+
+	fsvcCopy := *fsvc
+	return &fsvcCopy, nil
+}
+
+// TODO: error should be second return
+func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (error, *FuncSvc) {
+	err, existing := fsc.byFunction.Set(crd.CacheKey(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
+
+	// Add to byAddress and byKubernetesObject caches. Ignore NameExists errors
+	// because of multiple-specialization. See issue #331.
+	err, _ = fsc.byAddress.Set(fsvc.Address, *fsvc.Function)
+	if err != nil {
+		if fe, ok := err.(fission.Error); ok {
+			if fe.Code == fission.ErrorNameExists {
+				err = nil
+			}
+		}
+		log.Printf("error caching fsvc: %v", err)
+		return err, nil
+	}
+	err, _ = fsc.byKubeObject.Set(fsvc.KubernetesObject, *fsvc.Function)
+	if err != nil {
+		if fe, ok := err.(fission.Error); ok {
+			if fe.Code == fission.ErrorNameExists {
+				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.(metav1.ObjectMeta)
+	fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
+	if err != nil {
+		return err
+	}
+	fsvc := fsvcI.(*FuncSvc)
+	fsvc.Atime = time.Now()
+	return nil
+}
+
+func (fsc *FunctionServiceCache) DeleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) {
+	responseChannel := make(chan *fscResponse)
+	fsc.requestChannel <- &fscRequest{
+		requestType:      DELETE_BY_OBJECT,
+		kubernetesObject: obj,
+		age:              minAge,
+		responseChannel:  responseChannel,
+	}
+	resp := <-responseChannel
+	return resp.deleted, resp.error
+}
+
+// _deleteByKubeObject deletes the entry keyed by Kubernetes Object, but only if it is
+// at least minAge old.
+func (fsc *FunctionServiceCache) _deleteByKubeObject(obj api.ObjectReference, minAge time.Duration) (bool, error) {
+	mI, err := fsc.byKubeObject.Get(obj)
+	if err != nil {
+		return false, err
+	}
+	m := mI.(metav1.ObjectMeta)
+	fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
+	if err != nil {
+		return false, err
+	}
+	fsvc := fsvcI.(*FuncSvc)
+
+	if time.Now().Sub(fsvc.Atime) < minAge {
+		return false, nil
+	}
+
+	fsc.byFunction.Delete(crd.CacheKey(&m))
+	fsc.byAddress.Delete(fsvc.Address)
+	fsc.byKubeObject.Delete(obj)
+
+	return true, nil
+}
+
+func (fsc *FunctionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]api.ObjectReference, error) {
+	responseChannel := make(chan *fscResponse)
+	fsc.requestChannel <- &fscRequest{
+		requestType:     LISTOLD,
+		age:             age,
+		env:             env,
+		responseChannel: responseChannel,
+	}
+	resp := <-responseChannel
+	return resp.objects, 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")
+}

executor/fscache/functionServiceCache_test.go

@@ -0,0 +1,91 @@
+package fscache
+
+import (
+	"log"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/pkg/api"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
+)
+
+func TestFunctionServiceCache(t *testing.T) {
+	fsc := MakeFunctionServiceCache()
+	if fsc == nil {
+		log.Panicf("error creating cache")
+	}
+
+	var fsvc *FuncSvc
+	now := time.Now()
+
+	fsvc = &FuncSvc{
+		Function: &metav1.ObjectMeta{
+			Name: "foo",
+			UID:  "1212",
+		},
+		Environment: &crd.Environment{
+			Metadata: metav1.ObjectMeta{
+				Name: "foo-env",
+				UID:  "2323",
+			},
+			Spec: fission.EnvironmentSpec{
+				Version: 1,
+				Runtime: fission.Runtime{
+					Image: "fission/foo-env",
+				},
+				Builder: fission.Builder{},
+			},
+		},
+		Address: "xxx",
+		KubernetesObject: api.ObjectReference{
+			Kind:       "pod",
+			Name:       "xxx",
+			APIVersion: "v1",
+			Namespace:  "fission-function",
+		},
+		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)
+	}
+
+	deleted, err := fsc.DeleteByKubeObject(fsvc.KubernetesObject, 0)
+	if err != nil {
+		fsc.Log()
+		log.Panicf("Failed to delete fsvc: %v", err)
+	}
+	if !deleted {
+		fsc.Log()
+		log.Panicf("Did not delete fsvc")
+	}
+
+	_, err = fsc.GetByFunction(fsvc.Function)
+	if err == nil {
+		fsc.Log()
+		log.Panicf("found fsvc while expecting empty cache: %v", err)
+	}
+}

executor/poolmgr/cleanup.go

@@ -0,0 +1,147 @@
+/*
+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"
+
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// cleanupOldPoolmgrResources looks for resources created by an old
+// poolmgr instance and cleans them up.
+func CleanupOldPoolmgrResources(client *kubernetes.Clientset, namespace string, instanceId string) {
+	go func() {
+		err := cleanup(client, namespace, instanceId)
+		if err != nil {
+			// TODO retry cleanup; logged and ignored for now
+			log.Printf("Failed to cleanup: %v", err)
+		}
+	}()
+}
+
+func cleanup(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	// Deployments are used for idle pools and can be cleaned up
+	// immediately.  (We should "adopt" these instead of creating
+	// a new pool.)
+	err := cleanupDeployments(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+	// See K8s #33845 and related bugs: deleting a deployment
+	// through the API doesn't cause the associated ReplicaSet to
+	// be deleted.  (Fixed recently, but we may be running a
+	// version before the fix.)
+	err = cleanupReplicaSets(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	// Pods might still be running user functions, so we give them
+	// a few minutes before terminating them.  This time is the
+	// maximum function runtime, plus the time a router might
+	// still route to an old instance, i.e. router cache expiry
+	// time.
+	time.Sleep(6 * time.Minute)
+
+	err = cleanupPods(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	err = cleanupServices(client, namespace, instanceId)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func cleanupDeployments(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	deploymentList, err := client.ExtensionsV1beta1().Deployments(namespace).List(meta_v1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, dep := range deploymentList.Items {
+		id, ok := dep.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up deployment %v", dep.ObjectMeta.Name)
+			err := client.ExtensionsV1beta1().Deployments(namespace).Delete(dep.ObjectMeta.Name, nil)
+			logErr("cleaning up deployment", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func cleanupReplicaSets(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	rsList, err := client.ExtensionsV1beta1().ReplicaSets(namespace).List(meta_v1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, rs := range rsList.Items {
+		id, ok := rs.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up replicaset %v", rs.ObjectMeta.Name)
+			err := client.ExtensionsV1beta1().ReplicaSets(namespace).Delete(rs.ObjectMeta.Name, nil)
+			logErr("cleaning up replicaset", err)
+		}
+	}
+	return nil
+}
+
+func cleanupPods(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	podList, err := client.CoreV1().Pods(namespace).List(meta_v1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, pod := range podList.Items {
+		id, ok := pod.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up pod %v", pod.ObjectMeta.Name)
+			err := client.CoreV1().Pods(namespace).Delete(pod.ObjectMeta.Name, nil)
+			logErr("cleaning up pod", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func cleanupServices(client *kubernetes.Clientset, namespace string, instanceId string) error {
+	svcList, err := client.CoreV1().Services(namespace).List(meta_v1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, svc := range svcList.Items {
+		id, ok := svc.ObjectMeta.Labels[POOLMGR_INSTANCEID_LABEL]
+		if ok && id != instanceId {
+			log.Printf("Cleaning up svc %v", svc.ObjectMeta.Name)
+			err := client.CoreV1().Services(namespace).Delete(svc.ObjectMeta.Name, nil)
+			logErr("cleaning up svc", err)
+			// ignore err
+		}
+	}
+	return nil
+}
+
+func logErr(msg string, err error) {
+	if err != nil {
+		log.Printf("Error %v: %v", msg, err)
+	}
+}

executor/poolmgr/gp.go

@@ -0,0 +1,681 @@
+/*
+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 (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"math/rand"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/dchest/uniuri"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/pkg/api"
+	apiv1 "k8s.io/client-go/pkg/api/v1"
+	"k8s.io/client-go/pkg/apis/extensions/v1beta1"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
+	"github.com/fission/fission/environments/fetcher"
+	fetcherClient "github.com/fission/fission/environments/fetcher/client"
+	"github.com/fission/fission/executor/fscache"
+)
+
+const POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId"
+const POD_PHASE_RUNNING string = "Running"
+
+type (
+	GenericPool struct {
+		env                    *crd.Environment
+		replicas               int32                         // num idle pods
+		deployment             *v1beta1.Deployment           // kubernetes deployment
+		namespace              string                        // namespace to keep our resources
+		podReadyTimeout        time.Duration                 // timeout for generic pods to become ready
+		idlePodReapTime        time.Duration                 // pods unused for idlePodReapTime are deleted
+		fsCache                *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname
+		useSvc                 bool                          // create k8s service for specialized pods
+		poolInstanceId         string                        // small random string to uniquify pod names
+		fetcherImage           string
+		fetcherImagePullPolicy apiv1.PullPolicy
+		runtimeImagePullPolicy apiv1.PullPolicy // pull policy for generic pool to created env deployment
+		kubernetesClient       *kubernetes.Clientset
+		fissionClient          *crd.FissionClient
+		instanceId             string // poolmgr instance id
+		labelsForPool          map[string]string
+		requestChannel         chan *choosePodRequest
+		sharedMountPath        string // used by generic pool when creating env deployment to specify the share volume path for fetcher & env
+	}
+
+	// serialize the choosing of pods so that choices don't conflict
+	choosePodRequest struct {
+		newLabels       map[string]string
+		responseChannel chan *choosePodResponse
+	}
+	choosePodResponse struct {
+		pod *apiv1.Pod
+		error
+	}
+)
+
+func getImagePullPolicy(policy string) apiv1.PullPolicy {
+	switch policy {
+	case "Always":
+		return apiv1.PullAlways
+	case "Never":
+		return apiv1.PullNever
+	default:
+		return apiv1.PullIfNotPresent
+	}
+}
+
+func MakeGenericPool(
+	fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset,
+	env *crd.Environment,
+	initialReplicas int32,
+	namespace string,
+	fsCache *fscache.FunctionServiceCache,
+	instanceId string) (*GenericPool, error) {
+
+	log.Printf("Creating pool for environment %v", env.Metadata)
+
+	fetcherImage := os.Getenv("FETCHER_IMAGE")
+	if len(fetcherImage) == 0 {
+		fetcherImage = "fission/fetcher"
+	}
+	fetcherImagePullPolicy := os.Getenv("FETCHER_IMAGE_PULL_POLICY")
+	if len(fetcherImagePullPolicy) == 0 {
+		fetcherImagePullPolicy = "IfNotPresent"
+	}
+	runtimeImagePullPolicy := os.Getenv("RUNTIME_IMAGE_PULL_POLICY")
+	if len(runtimeImagePullPolicy) == 0 {
+		runtimeImagePullPolicy = "IfNotPresent"
+	}
+
+	// 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, // TODO make this an env param instead?
+		requestChannel:   make(chan *choosePodRequest),
+		fissionClient:    fissionClient,
+		kubernetesClient: kubernetesClient,
+		namespace:        namespace,
+		podReadyTimeout:  5 * time.Minute, // TODO make this an env param?
+		idlePodReapTime:  3 * time.Minute, // TODO make this configurable
+		fsCache:          fsCache,
+		poolInstanceId:   uniuri.NewLen(8),
+		instanceId:       instanceId,
+		fetcherImage:     fetcherImage,
+		useSvc:           false,       // defaults off -- svc takes a second or more to become routable, slowing cold start
+		sharedMountPath:  "/userfunc", // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path
+	}
+
+	gp.runtimeImagePullPolicy = getImagePullPolicy(runtimeImagePullPolicy)
+
+	gp.fetcherImagePullPolicy = getImagePullPolicy(fetcherImagePullPolicy)
+	log.Printf("fetcher image: %v, pull policy: %v", gp.fetcherImage, gp.fetcherImagePullPolicy)
+
+	// Labels for generic deployment/RS/pods.
+	gp.labelsForPool = map[string]string{
+		"environmentName":        gp.env.Metadata.Name,
+		"environmentUid":         string(gp.env.Metadata.UID),
+		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
+	}
+
+	// create the pool
+	err := gp.createPool()
+	if err != nil {
+		return nil, err
+	}
+	log.Printf("[%v] Deployment created", env.Metadata)
+
+	go gp.choosePodService()
+
+	// Unless specified otherwise, periodically cleanup inactive pods.
+	if env.Spec.AllowedFunctionsPerContainer != fission.AllowedFunctionsPerContainerInfinite {
+		go gp.idlePodReaper()
+	}
+
+	return gp, nil
+}
+
+// choosePodService serializes the choosing of pods
+func (gp *GenericPool) choosePodService() {
+	for {
+		select {
+		case req := <-gp.requestChannel:
+			pod, err := gp._choosePod(req.newLabels)
+			if err != nil {
+				req.responseChannel <- &choosePodResponse{error: err}
+				continue
+			}
+			req.responseChannel <- &choosePodResponse{pod: pod}
+		}
+	}
+}
+
+// choosePod picks a ready pod from the pool and relabels it, waiting if necessary.
+// returns the pod API object.
+func (gp *GenericPool) choosePod(newLabels map[string]string) (*apiv1.Pod, error) {
+	req := &choosePodRequest{
+		newLabels:       newLabels,
+		responseChannel: make(chan *choosePodResponse),
+	}
+	gp.requestChannel <- req
+	resp := <-req.responseChannel
+	return resp.pod, resp.error
+}
+
+// _choosePod is called serially by choosePodService
+func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, error) {
+	startTime := time.Now()
+	for {
+		// Retries took too long, error out.
+		if time.Now().Sub(startTime) > gp.podReadyTimeout {
+			log.Printf("[%v] Erroring out, timed out", newLabels)
+			return nil, errors.New("timeout: waited too long to get a ready pod")
+		}
+
+		// Get pods; filter the ones that are ready
+		podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(
+			metav1.ListOptions{
+				LabelSelector: labels.Set(
+					gp.deployment.Spec.Selector.MatchLabels).AsSelector().String(),
+			})
+		if err != nil {
+			return nil, err
+		}
+		readyPods := make([]*apiv1.Pod, 0, len(podList.Items))
+		for i := range podList.Items {
+			pod := podList.Items[i]
+
+			// If a pod has no IP it's not ready
+			if len(pod.Status.PodIP) == 0 || string(pod.Status.Phase) != POD_PHASE_RUNNING {
+				continue
+			}
+
+			// Wait for all containers in the pod to be ready
+			podReady := true
+			for _, cs := range pod.Status.ContainerStatuses {
+				podReady = podReady && cs.Ready
+			}
+
+			// add it to the list of ready pods
+			if podReady {
+				readyPods = append(readyPods, &pod)
+			}
+		}
+		log.Printf("[%v] found %v ready pods of %v total", newLabels, len(readyPods), len(podList.Items))
+
+		// If there are no ready pods, wait and retry.
+		if len(readyPods) == 0 {
+			err = gp.waitForReadyPod()
+			if err != nil {
+				return nil, err
+			}
+			continue
+		}
+
+		// Pick a ready pod.  For now just choose randomly;
+		// ideally we'd care about which node it's running on,
+		// and make a good scheduling decision.
+		chosenPod := readyPods[rand.Intn(len(readyPods))]
+
+		if gp.env.Spec.AllowedFunctionsPerContainer != fission.AllowedFunctionsPerContainerInfinite {
+			// Relabel.  If the pod already got picked and
+			// modified, this should fail; in that case just
+			// retry.
+			chosenPod.ObjectMeta.Labels = newLabels
+			log.Printf("relabeling pod: [%v]", chosenPod.ObjectMeta.Name)
+			_, err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Update(chosenPod)
+			if err != nil {
+				log.Printf("failed to relabel pod [%v]: %v", chosenPod.ObjectMeta.Name, err)
+				continue
+			}
+		}
+		log.Printf("Chosen pod: %v (in %v)", chosenPod.ObjectMeta.Name, time.Now().Sub(startTime))
+		return chosenPod, nil
+	}
+}
+
+func (gp *GenericPool) labelsForFunction(metadata *metav1.ObjectMeta) map[string]string {
+	return map[string]string{
+		"functionName":           metadata.Name,
+		"functionUid":            string(metadata.UID),
+		"unmanaged":              "true", // this allows us to easily find pods not managed by the deployment
+		POOLMGR_INSTANCEID_LABEL: gp.instanceId,
+	}
+}
+
+func (gp *GenericPool) scheduleDeletePod(name string) {
+	go func() {
+		// The sleep allows debugging or collecting logs from the pod before it's
+		// cleaned up.  (We need a better solutions for both those things; log
+		// aggregation and storage will help.)
+		log.Printf("Error in pod '%v', scheduling cleanup", name)
+		time.Sleep(5 * time.Minute)
+		gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil)
+	}()
+}
+
+func (gp *GenericPool) getFetcherUrl(podIP string) string {
+	testUrl := os.Getenv("TEST_FETCHER_URL")
+	if len(testUrl) != 0 {
+		// it takes a second or so for the test service to
+		// become routable once a pod is relabeled. This is
+		// super hacky, but only runs in unit tests.
+		time.Sleep(5 * time.Second)
+		return testUrl
+	}
+	return fmt.Sprintf("http://%v:8000/", podIP)
+}
+
+func (gp *GenericPool) getSpecializeUrl(podIP string, version int) string {
+	u := os.Getenv("TEST_SPECIALIZE_URL")
+	if len(u) != 0 {
+		return u
+	}
+	if version == 1 {
+		return fmt.Sprintf("http://%v:8888/specialize", podIP)
+	}
+	return fmt.Sprintf("http://%v:8888/v%v/specialize", podIP, version)
+}
+
+// specializePod chooses a pod, copies the required user-defined function to that pod
+// (via fetcher), and calls the function-run container to load it, resulting in a
+// specialized pod.
+func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta) error {
+	// for fetcher we don't need to create a service, just talk to the pod directly
+	podIP := pod.Status.PodIP
+	if len(podIP) == 0 {
+		return errors.New("Pod has no IP")
+	}
+
+	// tell fetcher to get the function.
+	fetcherUrl := gp.getFetcherUrl(podIP)
+	log.Printf("[%v] calling fetcher to copy function", metadata.Name)
+
+	fn, err := gp.fissionClient.
+		Functions(metadata.Namespace).
+		Get(metadata.Name)
+	if err != nil {
+		return err
+	}
+
+	// for backward compatibility, since most v1 env
+	// still try to load user function from hard coded
+	// path /userfunc/user
+	targetFilename := "user"
+	if gp.env.Spec.Version == 2 {
+		targetFilename = string(fn.Metadata.UID)
+	}
+
+	err = fetcherClient.MakeClient(fetcherUrl).Fetch(&fetcher.FetchRequest{
+		FetchType: fetcher.FETCH_DEPLOYMENT,
+		Package: metav1.ObjectMeta{
+			Namespace: fn.Spec.Package.PackageRef.Namespace,
+			Name:      fn.Spec.Package.PackageRef.Name,
+		},
+		Filename: targetFilename,
+	})
+	if err != nil {
+		return err
+	}
+
+	// get function run container to specialize
+	log.Printf("[%v] specializing pod", metadata.Name)
+
+	// retry the specialize call a few times in case the env server hasn't come up yet
+	maxRetries := 20
+
+	loadReq := fission.FunctionLoadRequest{
+		FilePath:         filepath.Join(gp.sharedMountPath, targetFilename),
+		FunctionName:     fn.Spec.Package.FunctionName,
+		FunctionMetadata: &fn.Metadata,
+	}
+
+	body, err := json.Marshal(loadReq)
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < maxRetries; i++ {
+		var resp2 *http.Response
+		if gp.env.Spec.Version == 2 {
+			specializeUrl := gp.getSpecializeUrl(podIP, 2)
+			resp2, err = http.Post(specializeUrl, "application/json", bytes.NewReader(body))
+		} else {
+			specializeUrl := gp.getSpecializeUrl(podIP, 1)
+			resp2, err = http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{}))
+		}
+		if err == nil && resp2.StatusCode < 300 {
+			// Success
+			resp2.Body.Close()
+			return nil
+		}
+
+		// Only retry for the specific case of a connection error.
+		if urlErr, ok := err.(*url.Error); ok {
+			if netErr, ok := urlErr.Err.(*net.OpError); ok {
+				if netErr.Op == "dial" {
+					if i < maxRetries-1 {
+						time.Sleep(500 * time.Duration(2*i) * time.Millisecond)
+						log.Printf("Error connecting to pod (%v), retrying", netErr)
+						continue
+					}
+				}
+			}
+		}
+
+		if err == nil {
+			err = fission.MakeErrorFromHTTP(resp2)
+		}
+		log.Printf("Failed to specialize pod: %v", err)
+		return err
+	}
+
+	return nil
+}
+
+// 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-%v",
+		gp.env.Metadata.Name, gp.env.Metadata.UID, strings.ToLower(gp.poolInstanceId))
+
+	deployment := &v1beta1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   poolDeploymentName,
+			Labels: gp.labelsForPool,
+		},
+		Spec: v1beta1.DeploymentSpec{
+			Replicas: &gp.replicas,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: gp.labelsForPool,
+			},
+			Template: apiv1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: gp.labelsForPool,
+				},
+				Spec: apiv1.PodSpec{
+					Volumes: []apiv1.Volume{
+						{
+							Name: "userfunc",
+							VolumeSource: apiv1.VolumeSource{
+								EmptyDir: &apiv1.EmptyDirVolumeSource{},
+							},
+						},
+					},
+					Containers: []apiv1.Container{
+						{
+							Name:                   gp.env.Metadata.Name,
+							Image:                  gp.env.Spec.Runtime.Image,
+							ImagePullPolicy:        gp.runtimeImagePullPolicy,
+							TerminationMessagePath: "/dev/termination-log",
+							VolumeMounts: []apiv1.VolumeMount{
+								{
+									Name:      "userfunc",
+									MountPath: gp.sharedMountPath,
+								},
+							},
+						},
+						{
+							Name:                   "fetcher",
+							Image:                  gp.fetcherImage,
+							ImagePullPolicy:        gp.fetcherImagePullPolicy,
+							TerminationMessagePath: "/dev/termination-log",
+							VolumeMounts: []apiv1.VolumeMount{
+								{
+									Name:      "userfunc",
+									MountPath: gp.sharedMountPath,
+								},
+							},
+							Command: []string{"/fetcher", gp.sharedMountPath},
+						},
+					},
+					ServiceAccountName: "fission-fetcher",
+				},
+			},
+		},
+	}
+	depl, err := gp.kubernetesClient.ExtensionsV1beta1().Deployments(gp.namespace).Create(deployment)
+	if err != nil {
+		return err
+	}
+	gp.deployment = depl
+	return nil
+}
+
+func (gp *GenericPool) waitForReadyPod() error {
+	startTime := time.Now()
+	for {
+		// TODO: for now we just poll; use a watch instead
+		depl, err := gp.kubernetesClient.ExtensionsV1beta1().Deployments(gp.namespace).Get(
+			gp.deployment.ObjectMeta.Name, metav1.GetOptions{})
+		if err != nil {
+			log.Printf("err: %v", err)
+			return err
+		}
+		gp.deployment = depl
+		if gp.deployment.Status.AvailableReplicas > 0 {
+			return nil
+		}
+
+		if time.Now().Sub(startTime) > gp.podReadyTimeout {
+			return errors.New("timeout: waited too long for pod to be ready")
+		}
+		time.Sleep(1000 * time.Millisecond)
+	}
+}
+
+func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.Service, error) {
+	service := apiv1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: apiv1.ServiceSpec{
+			Type: apiv1.ServiceTypeClusterIP,
+			Ports: []apiv1.ServicePort{
+				{
+					Protocol:   apiv1.ProtocolTCP,
+					Port:       80,
+					TargetPort: intstr.FromInt(8888),
+				},
+			},
+			Selector: labels,
+		},
+	}
+	svc, err := gp.kubernetesClient.CoreV1().Services(gp.namespace).Create(&service)
+	return svc, err
+}
+
+func (gp *GenericPool) GetFuncSvc(m *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
+
+	log.Printf("[%v] Choosing pod from pool", m.Name)
+	newLabels := gp.labelsForFunction(m)
+	pod, err := gp.choosePod(newLabels)
+	if err != nil {
+		return nil, err
+	}
+
+	err = gp.specializePod(pod, m)
+	if err != nil {
+		gp.scheduleDeletePod(pod.ObjectMeta.Name)
+		return nil, err
+	}
+	log.Printf("Specialized pod: %v", pod.ObjectMeta.Name)
+
+	var svcHost string
+	if gp.useSvc {
+		svcName := fmt.Sprintf("svc-%v", m.Name)
+		if len(m.UID) > 0 {
+			svcName = fmt.Sprintf("%s-%v", svcName, m.UID)
+		}
+
+		labels := gp.labelsForFunction(m)
+		svc, err := gp.createSvc(svcName, labels)
+		if err != nil {
+			gp.scheduleDeletePod(pod.ObjectMeta.Name)
+			return nil, err
+		}
+		if svc.ObjectMeta.Name != svcName {
+			gp.scheduleDeletePod(pod.ObjectMeta.Name)
+			return nil, errors.New(fmt.Sprintf("sanity check failed for svc %v", svc.ObjectMeta.Name))
+		}
+
+		// the fission router isn't in the same namespace, so return a
+		// namespace-qualified hostname
+		svcHost = fmt.Sprintf("%v.%v", svcName, gp.namespace)
+	} else {
+		log.Printf("Using pod IP for specialized pod")
+		svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP)
+	}
+
+	kubeObjRef := api.ObjectReference{
+		Kind:            pod.TypeMeta.Kind,
+		Name:            pod.ObjectMeta.Name,
+		APIVersion:      pod.TypeMeta.APIVersion,
+		Namespace:       pod.ObjectMeta.Namespace,
+		ResourceVersion: pod.ObjectMeta.ResourceVersion,
+		UID:             pod.ObjectMeta.UID,
+	}
+
+	fsvc := &fscache.FuncSvc{
+		Function:         m,
+		Environment:      gp.env,
+		Address:          svcHost,
+		KubernetesObject: kubeObjRef,
+		Backend:          fscache.POOLMGR,
+		Ctime:            time.Now(),
+		Atime:            time.Now(),
+	}
+
+	err, _ = gp.fsCache.Add(*fsvc)
+	if err != nil {
+		return nil, err
+	}
+	return fsvc, nil
+}
+
+func (gp *GenericPool) CleanupFunctionService(obj api.ObjectReference) error {
+	// remove ourselves from fsCache (only if we're still old)
+	deleted, err := gp.fsCache.DeleteByKubeObject(obj, gp.idlePodReapTime)
+	if err != nil {
+		return err
+	}
+
+	if !deleted {
+		log.Printf("Not deleting %v, in use", obj.Name)
+		return nil
+	}
+
+	pod, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Get(obj.Name, metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
+	loggerUrl := fmt.Sprintf("http://%s:1234/v1/log/%s", pod.Spec.NodeName, pod.Name)
+	req, err := http.NewRequest("DELETE", loggerUrl, nil)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		log.Printf("Error from %s daemonset logger: %v", pod.Spec.NodeName, err)
+	} else {
+		if resp.StatusCode != 200 {
+			log.Printf("Received not http 200(OK) status from %s daemonset logger: %s", pod.Spec.NodeName, resp.Status)
+		}
+		resp.Body.Close()
+	}
+
+	// delete pod
+	err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(obj.Name, nil)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (gp *GenericPool) idlePodReaper() {
+	for {
+		time.Sleep(time.Minute)
+		objects, err := gp.fsCache.ListOld(&gp.env.Metadata, gp.idlePodReapTime)
+		if err != nil {
+			log.Printf("Error reaping idle pods: %v", err)
+			continue
+		}
+		for _, obj := range objects {
+			log.Printf("Reaping idle pod '%v'", obj.Name)
+			err := gp.CleanupFunctionService(obj)
+			if err != nil {
+				log.Printf("Error deleting idle pod '%v': %v", obj.Name, err)
+			}
+		}
+	}
+}
+
+// destroys the pool -- the deployment, replicaset and pods
+func (gp *GenericPool) destroy() error {
+	// Destroy deployment
+	err := gp.kubernetesClient.ExtensionsV1beta1().Deployments(gp.namespace).Delete(gp.deployment.ObjectMeta.Name, nil)
+	if err != nil {
+		log.Printf("Error destroying deployment: %v", err)
+		return err
+	}
+
+	// Destroy ReplicaSet.  Pre-1.6 K8s versions don't do this
+	// automatically but post-1.6 K8s will, and may beat us to it,
+	// so don't error out if we fail.
+	rsList, err := gp.kubernetesClient.ExtensionsV1beta1().ReplicaSets(gp.namespace).List(metav1.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector().String(),
+	})
+	if len(rsList.Items) >= 0 {
+		for _, rs := range rsList.Items {
+			err = gp.kubernetesClient.ExtensionsV1beta1().ReplicaSets(gp.namespace).Delete(rs.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting replicaset, ignoring: %v", err)
+			}
+		}
+	}
+
+	// Destroy Pods.  See note above.
+	podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(metav1.ListOptions{
+		LabelSelector: labels.Set(gp.labelsForPool).AsSelector().String(),
+	})
+	if len(podList.Items) >= 0 {
+		for _, pod := range podList.Items {
+			err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(pod.ObjectMeta.Name, nil)
+			if err != nil {
+				log.Printf("Error deleting pod, ignoring: %v", err)
+			}
+		}
+	}
+
+	return nil
+}

executor/poolmgr/gpm.go

@@ -0,0 +1,172 @@
+/*
+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"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
+	"github.com/fission/fission/executor/fscache"
+)
+
+type requestType int
+
+const (
+	GET_POOL requestType = iota
+	CLEANUP_POOLS
+)
+
+type (
+	GenericPoolManager struct {
+		pools            map[string]*GenericPool
+		kubernetesClient *kubernetes.Clientset
+		namespace        string
+
+		fissionClient  *crd.FissionClient
+		fsCache        *fscache.FunctionServiceCache
+		instanceId     string
+		requestChannel chan *request
+	}
+	request struct {
+		requestType
+		env             *crd.Environment
+		envList         []crd.Environment
+		responseChannel chan *response
+	}
+	response struct {
+		error
+		pool *GenericPool
+	}
+)
+
+func MakeGenericPoolManager(
+	fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset,
+	fissionNamespace string,
+	functionNamespace string,
+	fsCache *fscache.FunctionServiceCache,
+	instanceId string) *GenericPoolManager {
+
+	gpm := &GenericPoolManager{
+		pools:            make(map[string]*GenericPool),
+		kubernetesClient: kubernetesClient,
+		namespace:        functionNamespace,
+		fissionClient:    fissionClient,
+		fsCache:          fsCache,
+		instanceId:       instanceId,
+		requestChannel:   make(chan *request),
+	}
+	go gpm.service()
+	go gpm.eagerPoolCreator()
+
+	return gpm
+}
+
+func (gpm *GenericPoolManager) service() {
+	for {
+		req := <-gpm.requestChannel
+		switch req.requestType {
+		case GET_POOL:
+			var err error
+			pool, ok := gpm.pools[crd.CacheKey(&req.env.Metadata)]
+			if !ok {
+				var poolSize int32 = 3 // TODO configurable/autoscalable
+				switch req.env.Spec.AllowedFunctionsPerContainer {
+				case fission.AllowedFunctionsPerContainerInfinite:
+					poolSize = 1
+				}
+
+				pool, err = MakeGenericPool(
+					gpm.fissionClient, gpm.kubernetesClient, req.env, poolSize,
+					gpm.namespace, gpm.fsCache, gpm.instanceId)
+				if err != nil {
+					req.responseChannel <- &response{error: err}
+					continue
+				}
+				gpm.pools[crd.CacheKey(&req.env.Metadata)] = pool
+			}
+			req.responseChannel <- &response{pool: pool}
+		case CLEANUP_POOLS:
+			latestEnvSet := make(map[string]bool)
+			for _, env := range req.envList {
+				latestEnvSet[crd.CacheKey(&env.Metadata)] = true
+			}
+			for key, pool := range gpm.pools {
+				_, ok := latestEnvSet[key]
+				if !ok {
+					// Env no longer exists -- remove our cache
+					log.Printf("Destroying generic pool for environment [%v]", key)
+					delete(gpm.pools, key)
+
+					// and delete the pool asynchronously.
+					go pool.destroy()
+				}
+			}
+			// no response, caller doesn't wait
+		}
+	}
+}
+
+func (gpm *GenericPoolManager) GetPool(env *crd.Environment) (*GenericPool, error) {
+	c := make(chan *response)
+	gpm.requestChannel <- &request{
+		requestType:     GET_POOL,
+		env:             env,
+		responseChannel: c,
+	}
+	resp := <-c
+	return resp.pool, resp.error
+}
+
+func (gpm *GenericPoolManager) CleanupPools(envs []crd.Environment) {
+	gpm.requestChannel <- &request{
+		requestType: CLEANUP_POOLS,
+		envList:     envs,
+	}
+}
+
+func (gpm *GenericPoolManager) eagerPoolCreator() {
+	pollSleep := time.Duration(2 * time.Second)
+	for {
+		time.Sleep(pollSleep)
+
+		// get list of envs from controller
+		envs, err := gpm.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{})
+		if err != nil {
+			log.Fatalf("Failed to get environment list: %v", err)
+		}
+
+		// Create pools for all envs.  TODO: we should make this a bit less eager, only
+		// creating pools for envs that are actually used by functions.  Also we might want
+		// to keep these eagerly created pools smaller than the ones created when there are
+		// actual function calls.
+		for i := range envs.Items {
+			_, err := gpm.GetPool(&envs.Items[i])
+			if err != nil {
+				log.Printf("eager-create pool failed: %v", err)
+			}
+		}
+
+		// Clean up pools whose env was deleted
+		gpm.CleanupPools(envs.Items)
+	}
+}