Fission workflow env integration (#336)

Add a flag to the environment to control multiple specialization -- the max number of functions per container. This can be set to 1 or infinity. 

Add an api proxy to workflow apiserver from the controller.

Add function metadata to FunctionLoadRequest; every v2 environment now knows which function it's loading (but can ignore that information if it wants to).

Add function identity headers to router. This is useful for multiple specialization, so the router can disambiguate between different function calls. (If this turns out to be a non-trivial perf overhead, we could add these headers conditionally, but for now they are always added.)

controller/api.go

@@ -38,6 +38,7 @@ type (
 		fissionClient     *tpr.FissionClient
 		storageServiceUrl string
 		builderManagerUrl string
+		workflowApiUrl    string
 	}
 
 	logDBConfig struct {
@@ -64,6 +65,13 @@ func MakeAPI() (*API, error) {
 		api.builderManagerUrl = "http://buildermgr"
 	}
 
+	wfEnv := os.Getenv("WORKFLOW_API_URL")
+	if len(u) > 0 {
+		api.workflowApiUrl = strings.TrimSuffix(wfEnv, "/")
+	} else {
+		api.workflowApiUrl = "http://workflow-apiserver"
+	}
+
 	return api, err
 }
 
@@ -170,6 +178,7 @@ func (api *API) Serve(port int) {
 	r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy)
 	r.HandleFunc("/proxy/buildermgr/v1/build", api.BuilderManagerBuildProxy)
 	r.HandleFunc("/proxy/buildermgr/v1/builder", api.BuilderManagerEnvBuilderProxy)
+	r.HandleFunc("/proxy/workflow", api.WorkflowApiProxy)
 
 	address := fmt.Sprintf(":%v", port)
 

controller/workflowApiProxy.go

@@ -0,0 +1,29 @@
+package controller
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"strings"
+)
+
+func (api *API) WorkflowApiProxy(w http.ResponseWriter, r *http.Request) {
+	u := api.storageServiceUrl
+	ssUrl, err := url.Parse(u)
+	if err != nil {
+		msg := fmt.Sprintf("Error parsing url %v: %v", u, err)
+		http.Error(w, msg, 500)
+		return
+	}
+	director := func(req *http.Request) {
+		req.URL.Scheme = ssUrl.Scheme
+		req.URL.Host = ssUrl.Host
+		req.URL.Path = strings.TrimPrefix(ssUrl.Path, "/proxy/workflow")
+		req.URL.RawQuery = ssUrl.RawQuery
+	}
+	proxy := &httputil.ReverseProxy{
+		Director: director,
+	}
+	proxy.ServeHTTP(w, r)
+}

fission-bundle/build.sh

@@ -1,2 +1,2 @@
 #!/bin/sh
-CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build 
+CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build

poolmgr/gp.go

@@ -157,7 +157,10 @@ func MakeGenericPool(
 
 	go gp.choosePodService()
 
-	go gp.idlePodReaper()
+	// Unless specified otherwise, periodically cleanup inactive pods.
+	if env.Spec.AllowedFunctionsPerContainer != fission.AllowedFunctionsPerContainerInfinite {
+		go gp.idlePodReaper()
+	}
 
 	return gp, nil
 }
@@ -202,8 +205,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
 		// Get pods; filter the ones that are ready
 		podList, err := gp.kubernetesClient.Core().Pods(gp.namespace).List(
 			api.ListOptions{
-				LabelSelector: labels.Set(
+				LabelSelector: labels.Set(gp.deployment.Spec.Selector.MatchLabels).AsSelector(),
-					gp.deployment.Spec.Selector.MatchLabels).AsSelector(),
 			})
 		if err != nil {
 			return nil, err
@@ -228,8 +230,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
 				readyPods = append(readyPods, &pod)
 			}
 		}
-		log.Printf("[%v] found %v ready pods of %v total",
+		log.Printf("[%v] found %v ready pods of %v total", newLabels, len(readyPods), len(podList.Items))
-			newLabels, len(readyPods), len(podList.Items))
 
 		// If there are no ready pods, wait and retry.
 		if len(readyPods) == 0 {
@@ -245,15 +246,17 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*v1.Pod, error)
 		// and make a good scheduling decision.
 		chosenPod := readyPods[rand.Intn(len(readyPods))]
 
-		// Relabel.  If the pod already got picked and
+		if gp.env.Spec.AllowedFunctionsPerContainer != fission.AllowedFunctionsPerContainerInfinite {
-		// modified, this should fail; in that case just
+			// Relabel.  If the pod already got picked and
-		// retry.
+			// modified, this should fail; in that case just
-		chosenPod.ObjectMeta.Labels = newLabels
+			// retry.
-		log.Printf("relabeling pod: [%v]", chosenPod.ObjectMeta.Name)
+			chosenPod.ObjectMeta.Labels = newLabels
-		_, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(chosenPod)
+			log.Printf("relabeling pod: [%v]", chosenPod.ObjectMeta.Name)
-		if err != nil {
+			_, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(chosenPod)
-			log.Printf("failed to relabel pod [%v]: %v", chosenPod.ObjectMeta.Name, err)
+			if err != nil {
-			continue
+				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
@@ -348,8 +351,9 @@ func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *api.ObjectMeta) erro
 	maxRetries := 20
 
 	loadReq := fission.FunctionLoadRequest{
-		FilePath:     filepath.Join(gp.sharedMountPath, targetFilename),
+		FilePath:         filepath.Join(gp.sharedMountPath, targetFilename),
-		FunctionName: fn.Spec.Package.FunctionName,
+		FunctionName:     fn.Spec.Package.FunctionName,
+		FunctionMetadata: &fn.Metadata,
 	}
 
 	body, err := json.Marshal(loadReq)
@@ -526,7 +530,7 @@ func (gp *GenericPool) GetFuncSvc(m *api.ObjectMeta) (*funcSvc, error) {
 	if gp.useSvc {
 		svcName := fmt.Sprintf("svc-%v", m.Name)
 		if len(m.UID) > 0 {
-			svcName += ("-" + string(m.UID))
+			svcName = fmt.Sprintf("%s-%v", svcName, m.UID)
 		}
 
 		labels := gp.labelsForFunction(m)

poolmgr/gpm.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/client-go/1.5/kubernetes"
 	"k8s.io/client-go/1.5/pkg/api"
 
+	"github.com/fission/fission"
 	"github.com/fission/fission/tpr"
 )
 
@@ -86,9 +87,14 @@ func (gpm *GenericPoolManager) service() {
 			var err error
 			pool, ok := gpm.pools[tpr.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,
+					gpm.fissionClient, gpm.kubernetesClient, req.env, poolSize,
-					3, // TODO configurable/autoscalable
 					gpm.namespace, gpm.fsCache, gpm.instanceId)
 				if err != nil {
 					req.responseChannel <- &response{error: err}

router/functionHandler.go

@@ -98,6 +98,9 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
 		request.Header.Add(fmt.Sprintf("X-Fission-Params-%v", k), v)
 	}
 
+	// System Params
+	MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request)
+
 	// cache lookup
 	serviceUrl, err := fh.fmap.lookup(fh.function)
 	if err != nil {

router/util.go

@@ -0,0 +1,28 @@
+package router
+
+import (
+	"fmt"
+	"k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/types"
+	"net/http"
+)
+
+const (
+	HEADERS_FISSION_FUNCTION_PREFIX = "Fission-Function"
+)
+
+func MetadataToHeaders(prefix string, meta *api.ObjectMeta, request *http.Request) {
+	request.Header.Add(fmt.Sprintf("X-%s-Uid", prefix), string(meta.UID))
+	request.Header.Add(fmt.Sprintf("X-%s-Name", prefix), meta.Name)
+	request.Header.Add(fmt.Sprintf("X-%s-Namespace", prefix), meta.Namespace)
+	request.Header.Add(fmt.Sprintf("X-%s-ResourceVersion", prefix), meta.ResourceVersion)
+}
+
+func HeadersToMetadata(prefix string, headers http.Header) *api.ObjectMeta {
+	return &api.ObjectMeta{
+		Name:            headers.Get(fmt.Sprintf("X-%s-Name", prefix)),
+		UID:             types.UID(headers.Get(fmt.Sprintf("X-%s-Uid", prefix))),
+		Namespace:       headers.Get(fmt.Sprintf("X-%s-Namespace", prefix)),
+		ResourceVersion: headers.Get(fmt.Sprintf("X-%s-ResourceVersion", prefix)),
+	}
+}

types.go

@@ -16,6 +16,8 @@ limitations under the License.
 
 package fission
 
+import "k8s.io/client-go/1.5/pkg/api"
+
 type (
 	//
 	// Functions and packages
@@ -163,8 +165,14 @@ type (
 		// Optional, but strongly encouraged. Used to populate
 		// links from UI, CLI, etc.
 		DocumentationURL string `json:"documentationurl"`
+
+		// Optional
+		// Defaults to 'Single'
+		AllowedFunctionsPerContainer AllowedFunctionsPerContainer `json:"allowedFunctionsPerContainer"`
 	}
 
+	AllowedFunctionsPerContainer string
+
 	//
 	// Triggers
 	//
@@ -230,6 +238,9 @@ type (
 		// URL to expose this function at. Optional; defaults
 		// to "/".
 		URL string `json:"url"`
+
+		// Metatdata
+		FunctionMetadata *api.ObjectMeta
 	}
 )
 
@@ -253,6 +264,11 @@ const (
 	BuildStatusFailed    = "failed"
 )
 
+const (
+	AllowedFunctionsPerContainerSingle   = "single"
+	AllowedFunctionsPerContainerInfinite = "infinite"
+)
+
 const (
 	// FunctionReferenceFunctionName means that the function
 	// reference is simply by function name.