From 33f967b61bb511cb452fdf6a92210b6cbb88d90f Mon Sep 17 00:00:00 2001 From: Erwin van Eyk Date: Tue, 26 Sep 2017 18:36:07 -0700 Subject: [PATCH] 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 | 9 ++++++++ controller/workflowApiProxy.go | 29 ++++++++++++++++++++++++++ fission-bundle/build.sh | 2 +- poolmgr/gp.go | 38 +++++++++++++++++++--------------- poolmgr/gpm.go | 10 +++++++-- router/functionHandler.go | 3 +++ router/util.go | 28 +++++++++++++++++++++++++ types.go | 16 ++++++++++++++ 8 files changed, 115 insertions(+), 20 deletions(-) create mode 100644 controller/workflowApiProxy.go create mode 100644 router/util.go diff --git a/controller/api.go b/controller/api.go index 4e256861..22404734 100644 --- a/controller/api.go +++ b/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) diff --git a/controller/workflowApiProxy.go b/controller/workflowApiProxy.go new file mode 100644 index 00000000..447c15f1 --- /dev/null +++ b/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) +} diff --git a/fission-bundle/build.sh b/fission-bundle/build.sh index f7378e10..eb2398a8 100755 --- a/fission-bundle/build.sh +++ b/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 diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 0f78ce2f..4281f6d1 100644 --- a/poolmgr/gp.go +++ b/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( - gp.deployment.Spec.Selector.MatchLabels).AsSelector(), + LabelSelector: labels.Set(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", - newLabels, len(readyPods), len(podList.Items)) + 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 { @@ -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 - // modified, this should fail; in that case just - // retry. - chosenPod.ObjectMeta.Labels = newLabels - log.Printf("relabeling pod: [%v]", chosenPod.ObjectMeta.Name) - _, err = gp.kubernetesClient.Core().Pods(gp.namespace).Update(chosenPod) - if err != nil { - log.Printf("failed to relabel pod [%v]: %v", chosenPod.ObjectMeta.Name, err) - continue + 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.Core().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 @@ -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), - FunctionName: fn.Spec.Package.FunctionName, + FilePath: filepath.Join(gp.sharedMountPath, targetFilename), + 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) diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 0be023f4..f8471a70 100644 --- a/poolmgr/gpm.go +++ b/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, - 3, // TODO configurable/autoscalable + gpm.fissionClient, gpm.kubernetesClient, req.env, poolSize, gpm.namespace, gpm.fsCache, gpm.instanceId) if err != nil { req.responseChannel <- &response{error: err} diff --git a/router/functionHandler.go b/router/functionHandler.go index b899c3d5..f4a1e6f2 100644 --- a/router/functionHandler.go +++ b/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 { diff --git a/router/util.go b/router/util.go new file mode 100644 index 00000000..08caa374 --- /dev/null +++ b/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)), + } +} diff --git a/types.go b/types.go index 24ffef97..37960efc 100644 --- a/types.go +++ b/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.