From e587eca08f8961a6860b48fb39d6690bda0ac2c9 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Mon, 25 Sep 2017 09:53:09 -0500 Subject: [PATCH] Add builder manager support (#308) This change orchestrates function builds. Environments (in v2) define a builder image, just like they do a runtime image. The builder image contains a build script that's invoked with source and deployment paths (as env vars). The buildermgr watches for environments with build images defined, and creates build deployments and services. Functions can define source and deployment. Buildermgr watches for functions with source code (and build status == pending) and invokes the environment's builder when appropriate. It captures logs from the build and sets the build status (success/failure) and build lots into the PackageStatus. --- builder/builder.go | 7 +- builder/client/client.go | 27 +- builder/cmd/Dockerfile | 5 - builder/cmd/build.sh | 5 +- builder/cmd/main.go | 2 +- buildermgr/api.go | 108 ++++ buildermgr/buildermgr.go | 39 ++ buildermgr/client/client.go | 61 +++ buildermgr/common.go | 211 ++++++++ buildermgr/envwatcher.go | 477 ++++++++++++++++++ buildermgr/pkgwatcher.go | 91 ++++ charts/README.md | 1 + charts/fission-all/templates/deployment.yaml | 75 +++ charts/fission-all/values.yaml | 4 + charts/fission-core/templates/deployment.yaml | 73 +++ charts/fission-core/values.yaml | 4 + controller/api.go | 10 + controller/buildermgr.go | 67 +++ controller/environmentApi.go | 1 + environments/fetcher/client/client.go | 49 +- environments/fetcher/cmd/main.go | 3 +- environments/fetcher/fetcher.go | 161 +++++- environments/python3/builder/Dockerfile | 4 +- environments/python3/builder/build.sh | 11 - environments/python3/server.py | 18 +- examples/python/sourcepkg/__init__.py | 0 examples/python/sourcepkg/build.sh | 2 + examples/python/sourcepkg/requirements.txt | 1 + examples/python/sourcepkg/user.py | 12 + fission-bundle/main.go | 44 +- fission/environment.go | 26 +- fission/function.go | 170 ++++--- fission/main.go | 14 +- poolmgr/gp.go | 96 +++- poolmgr/gpm.go | 2 +- test/build_and_test.sh | 6 + test/test_utils.sh | 30 ++ test/tests/test_buildermgr.sh | 73 +++ tpr/types.go | 2 + types.go | 11 +- 40 files changed, 1827 insertions(+), 176 deletions(-) delete mode 100644 builder/cmd/Dockerfile create mode 100644 buildermgr/api.go create mode 100644 buildermgr/buildermgr.go create mode 100644 buildermgr/client/client.go create mode 100644 buildermgr/common.go create mode 100644 buildermgr/envwatcher.go create mode 100644 buildermgr/pkgwatcher.go create mode 100644 controller/buildermgr.go delete mode 100755 environments/python3/builder/build.sh create mode 100644 examples/python/sourcepkg/__init__.py create mode 100755 examples/python/sourcepkg/build.sh create mode 100644 examples/python/sourcepkg/requirements.txt create mode 100644 examples/python/sourcepkg/user.py create mode 100755 test/tests/test_buildermgr.sh diff --git a/builder/builder.go b/builder/builder.go index afd7e560..7981a68f 100644 --- a/builder/builder.go +++ b/builder/builder.go @@ -98,7 +98,12 @@ func (builder *Builder) Handler(w http.ResponseWriter, r *http.Request) { srcPkgPath := filepath.Join(builder.sharedVolumePath, req.SrcPkgFilename) deployPkgFilename := fmt.Sprintf("%v-%v", req.SrcPkgFilename, strings.ToLower(uniuri.NewLen(6))) deployPkgPath := filepath.Join(builder.sharedVolumePath, deployPkgFilename) - buildLogs, err := builder.build(req.BuildCommand, srcPkgPath, deployPkgPath) + buildCmd := req.BuildCommand + if len(buildCmd) == 0 { + // use default build command + buildCmd = "/build" + } + buildLogs, err := builder.build(buildCmd, srcPkgPath, deployPkgPath) if err != nil { e := errors.New(fmt.Sprintf("Error building source package: %v", err)) http.Error(w, e.Error(), 500) diff --git a/builder/client/client.go b/builder/client/client.go index 0fe806c4..dd28d276 100644 --- a/builder/client/client.go +++ b/builder/client/client.go @@ -19,6 +19,7 @@ package client import ( "bytes" "encoding/json" + "io/ioutil" "net/http" "strings" @@ -32,25 +33,37 @@ type ( } ) -func MakeClient(serverUrl string) *Client { +func MakeClient(builderUrl string) *Client { return &Client{ - url: strings.TrimSuffix(serverUrl, "/"), + url: strings.TrimSuffix(builderUrl, "/"), } } -func (c *Client) Build(req *builder.PackageBuildRequest) error { +func (c *Client) Build(req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) { body, err := json.Marshal(req) if err != nil { - return err + return nil, err } resp, err := http.Post(c.url, "application/json", bytes.NewReader(body)) if err != nil { - return err + return nil, err } defer resp.Body.Close() if resp.StatusCode != 200 { - return fission.MakeErrorFromHTTP(resp) + return nil, fission.MakeErrorFromHTTP(resp) } - return nil + + rBody, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + pkgBuildResp := builder.PackageBuildResponse{} + err = json.Unmarshal([]byte(rBody), &pkgBuildResp) + if err != nil { + return nil, err + } + + return &pkgBuildResp, nil } diff --git a/builder/cmd/Dockerfile b/builder/cmd/Dockerfile deleted file mode 100644 index 11afba20..00000000 --- a/builder/cmd/Dockerfile +++ /dev/null @@ -1,5 +0,0 @@ -FROM alpine:3.4 - -ADD builder / - -EXPOSE 8000 diff --git a/builder/cmd/build.sh b/builder/cmd/build.sh index 69ba9c97..e6e3cfb0 100755 --- a/builder/cmd/build.sh +++ b/builder/cmd/build.sh @@ -1,2 +1,3 @@ -#!/bin/sh -GOOS=linux GOARCH=386 go build -o builder . +#!/bin/bash +CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o builder . + diff --git a/builder/cmd/main.go b/builder/cmd/main.go index b1619f85..9b6a4137 100644 --- a/builder/cmd/main.go +++ b/builder/cmd/main.go @@ -38,5 +38,5 @@ func main() { builder := builder.MakeBuilder(dir) mux := http.NewServeMux() mux.HandleFunc("/", builder.Handler) - http.ListenAndServe(":8000", mux) + http.ListenAndServe(":8001", mux) } diff --git a/buildermgr/api.go b/buildermgr/api.go new file mode 100644 index 00000000..c6e88d75 --- /dev/null +++ b/buildermgr/api.go @@ -0,0 +1,108 @@ +/* +Copyright 2017 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 buildermgr + +import ( + "encoding/json" + "fmt" + "io/ioutil" + "log" + "net/http" + "os" + + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/pkg/api" + + "github.com/fission/fission" + "github.com/fission/fission/tpr" +) + +type ( + BuildRequest struct { + Package api.ObjectMeta `json:"package"` + } + + BuilderMgr struct { + fissionClient *tpr.FissionClient + kubernetesClient *kubernetes.Clientset + storageSvcUrl string + namespace string + } +) + +func MakeBuilderMgr(fissionClient *tpr.FissionClient, + kubernetesClient *kubernetes.Clientset, storageSvcUrl string, + envBuilderNamespace string) *BuilderMgr { + + envWatcher := makeEnvironmentWatcher(fissionClient, kubernetesClient, envBuilderNamespace) + go envWatcher.watchEnvironments() + + pkgWatcher := makePackageWatcher(fissionClient, kubernetesClient, envBuilderNamespace, storageSvcUrl) + go pkgWatcher.watchPackages() + + return &BuilderMgr{ + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + storageSvcUrl: storageSvcUrl, + namespace: envBuilderNamespace, + } +} + +func (builderMgr *BuilderMgr) build(w http.ResponseWriter, r *http.Request) { + body, err := ioutil.ReadAll(r.Body) + if err != nil { + e := fmt.Sprintf("Failed to read request: %v", err) + log.Println(e) + http.Error(w, e, 500) + return + } + + buildReq := BuildRequest{} + err = json.Unmarshal([]byte(body), &buildReq) + if err != nil { + e := fmt.Sprintf("invalid request body: %v", err) + log.Println(e) + http.Error(w, e, 400) + return + } + + buildLogs, err := buildPackage(builderMgr.fissionClient, builderMgr.kubernetesClient, + builderMgr.namespace, builderMgr.storageSvcUrl, buildReq) + if err != nil { + code, e := fission.GetHTTPError(err) + http.Error(w, e, code) + return + } + + w.Header().Set("Content-Type", "text/plain") + _, err = w.Write([]byte(buildLogs)) + if err != nil { + e := fmt.Sprintf("Failed to reply http request: %v", err) + log.Println(e) + http.Error(w, e, 500) + } +} + +func (builderMgr *BuilderMgr) Serve(port int) { + r := mux.NewRouter() + r.HandleFunc("/v1/build", builderMgr.build).Methods("POST") + address := fmt.Sprintf(":%v", port) + log.Printf("Start buildermgr at port %v", address) + log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) +} diff --git a/buildermgr/buildermgr.go b/buildermgr/buildermgr.go new file mode 100644 index 00000000..4ff64c2e --- /dev/null +++ b/buildermgr/buildermgr.go @@ -0,0 +1,39 @@ +/* +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 buildermgr + +import ( + "log" + + "github.com/fission/fission/tpr" +) + +// Start the buildermgr service. +func Start(port int, storageSvcUrl string, envBuilderNamespace string) error { + fissionClient, kubernetesClient, err := tpr.MakeFissionClient() + if err != nil { + log.Printf("Failed to get kubernetes client: %v", err) + return err + } + + api := MakeBuilderMgr(fissionClient, kubernetesClient, + storageSvcUrl, envBuilderNamespace) + + go api.Serve(port) + + return nil +} diff --git a/buildermgr/client/client.go b/buildermgr/client/client.go new file mode 100644 index 00000000..6bb9c581 --- /dev/null +++ b/buildermgr/client/client.go @@ -0,0 +1,61 @@ +/* +Copyright 2017 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" + "net/http" + "strings" + + "github.com/fission/fission" + "github.com/fission/fission/buildermgr" +) + +type ( + Client struct { + url string + } +) + +func MakeClient(builderUrl string) *Client { + return &Client{ + url: strings.TrimSuffix(builderUrl, "/") + "/v1", + } +} + +func (c *Client) PackageBuild(req *buildermgr.BuildRequest) ([]byte, error) { + body, err := json.Marshal(req) + if err != nil { + return nil, err + } + resp, err := http.Post(c.url+"/build", "application/json", bytes.NewReader(body)) + if err != nil { + return nil, err + } + return c.handleResponse(resp) +} + +func (c *Client) handleResponse(resp *http.Response) ([]byte, error) { + if resp.StatusCode != 200 { + return nil, fission.MakeErrorFromHTTP(resp) + } + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + return body, err +} diff --git a/buildermgr/common.go b/buildermgr/common.go new file mode 100644 index 00000000..8094b607 --- /dev/null +++ b/buildermgr/common.go @@ -0,0 +1,211 @@ +/* +Copyright 2017 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 buildermgr + +import ( + "fmt" + "log" + "strings" + + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/pkg/api" + + "github.com/dchest/uniuri" + "github.com/fission/fission" + "github.com/fission/fission/builder" + builderClient "github.com/fission/fission/builder/client" + "github.com/fission/fission/environments/fetcher" + fetcherClient "github.com/fission/fission/environments/fetcher/client" + "github.com/fission/fission/tpr" +) + +// buildPackage helps to build source package into deployment package. +// Following is steps buildPackage takes to complete the build process. +// 1. Check package status +// 2. Update package status to running state +// 3. Send fetch request to fetcher to fetch source package. +// 4. Send build request to builder to start a build. +// 5. Send upload request to fetcher to upload deployment package. +// 6. Update package status to succeed state +// 7. Update package resource in package ref of functions that share the same package +// *. Update package status to failed state,if any one of steps above failed +func buildPackage(fissionClient *tpr.FissionClient, kubernetesClient *kubernetes.Clientset, + builderNamespace string, storageSvcUrl string, buildReq BuildRequest) (buildLogs string, err error) { + + pkg, err := fissionClient.Packages( + buildReq.Package.Namespace).Get(buildReq.Package.Name) + if err != nil { + e := fmt.Sprintf("Error getting function TPR info: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + // Only do build for pending packages + if pkg.Status.BuildStatus != fission.BuildStatusPending { + e := "package is not in pending state" + log.Println(e) + return e, fission.MakeError(400, e) + } + + // update package status to running state, so that + // we can know what status a package is through cli. + _, err = updatePackage(fissionClient, pkg, fission.BuildStatusRunning, "", nil) + if err != nil { + e := fmt.Sprintf("Error setting package pending state: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + env, err := fissionClient.Environments(api.NamespaceDefault).Get(pkg.Spec.Environment.Name) + if err != nil { + e := fmt.Sprintf("Error getting environment TPR info: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + svcName := fmt.Sprintf("%v-%v.%v", env.Metadata.Name, env.Metadata.ResourceVersion, builderNamespace) + srcPkgFilename := fmt.Sprintf("%v-%v", pkg.Metadata.Name, strings.ToLower(uniuri.NewLen(6))) + fetcherC := fetcherClient.MakeClient(fmt.Sprintf("http://%v:8000", svcName)) + builderC := builderClient.MakeClient(fmt.Sprintf("http://%v:8001", svcName)) + + fetchReq := &fetcher.FetchRequest{ + FetchType: fetcher.FETCH_SOURCE, + Package: pkg.Metadata, + Filename: srcPkgFilename, + } + + // send fetch request to fetcher + err = fetcherC.Fetch(fetchReq) + if err != nil { + e := fmt.Sprintf("Error fetching source package: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + pkgBuildReq := &builder.PackageBuildRequest{ + SrcPkgFilename: srcPkgFilename, + BuildCommand: pkg.Spec.BuildCommand, + } + + log.Printf("Start building with source package: %v", srcPkgFilename) + // send build request to builder + buildResp, err := builderC.Build(pkgBuildReq) + if err != nil { + e := fmt.Sprintf("Error building deployment package: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + log.Printf("Build succeed, source package: %v, deployment package: %v", srcPkgFilename, buildResp.ArtifactFilename) + + uploadReq := &fetcher.UploadRequest{ + Filename: buildResp.ArtifactFilename, + StorageSvcUrl: storageSvcUrl, + } + + log.Printf("Start uploading deployment package: %v", buildResp.ArtifactFilename) + // ask fetcher to upload the deployment package + uploadResp, err := fetcherC.Upload(uploadReq) + if err != nil { + e := fmt.Sprintf("Error uploading deployment package: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + log.Printf("Start updating info of package: %v", pkg.Metadata.Name) + // update package status and also build logs + newPkgRV, err := updatePackage(fissionClient, pkg, + fission.BuildStatusSucceeded, buildResp.BuildLogs, uploadResp) + if err != nil { + e := fmt.Sprintf("Error creating deployment package TPR resource: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + fnList, err := fissionClient. + Functions(api.NamespaceDefault).List(api.ListOptions{}) + if err != nil { + e := fmt.Sprintf("Error getting function list: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + + // A package may be used by multiple functions. Update + // functions with old package resource version + for _, fn := range fnList.Items { + if fn.Spec.Package.PackageRef.Name == pkg.Metadata.Name && + fn.Spec.Package.PackageRef.Namespace == pkg.Metadata.Namespace && + fn.Spec.Package.PackageRef.ResourceVersion != pkg.Metadata.ResourceVersion { + fn.Spec.Package.PackageRef.ResourceVersion = newPkgRV + // update TPR + _, err = fissionClient.Functions(fn.Metadata.Namespace).Update(&fn) + if err != nil { + e := fmt.Sprintf("Error updating function package resource version: %v", err) + log.Println(e) + updatePackage(fissionClient, pkg, fission.BuildStatusFailed, e, nil) + return e, fission.MakeError(500, e) + } + } + } + + log.Printf("Completed build request for package: %v", pkg.Metadata.Name) + + return buildResp.BuildLogs, nil +} + +func updatePackage(fissionClient *tpr.FissionClient, + pkg *tpr.Package, status fission.BuildStatus, buildLogs string, + uploadResp *fetcher.UploadResponse) (string, error) { + + // Kubernetes checks resource version before applying + // new resource config. The update operation will be + // rejected if the resource version in metadata is lower + // than the latest version. Set resource version empty + // string to skip resource version check. + pkg.Metadata.ResourceVersion = "" + + pkg.Status = fission.PackageStatus{ + BuildStatus: status, + BuildLog: buildLogs, + } + + if uploadResp != nil { + pkg.Spec.Deployment = fission.Archive{ + Type: fission.ArchiveTypeUrl, + URL: uploadResp.ArchiveDownloadUrl, + Checksum: uploadResp.Checksum, + } + } + + // update package spec + pkg, err := fissionClient.Packages(api.NamespaceDefault).Update(pkg) + if err != nil { + log.Printf("Error updating package: %v", err) + return "", err + } + + // return resource version for function to update function package ref + return pkg.Metadata.ResourceVersion, nil +} diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go new file mode 100644 index 00000000..016e8ecc --- /dev/null +++ b/buildermgr/envwatcher.go @@ -0,0 +1,477 @@ +/* +Copyright 2017 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 buildermgr + +import ( + "fmt" + "log" + "os" + "time" + + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/pkg/api" + "k8s.io/client-go/1.5/pkg/api/v1" + "k8s.io/client-go/1.5/pkg/apis/extensions/v1beta1" + "k8s.io/client-go/1.5/pkg/labels" + "k8s.io/client-go/1.5/pkg/util/intstr" + "k8s.io/client-go/1.5/pkg/watch" + + "github.com/fission/fission/tpr" +) + +type requestType int + +const ( + GET_BUILDER requestType = iota + CLEANUP_BUILDERS + + LABEL_ENV_NAME = "envName" + LABEL_ENV_RESOURCEVERSION = "envResourceVersion" +) + +type ( + builderInfo struct { + envMetadata *api.ObjectMeta + deployment *v1beta1.Deployment + service *v1.Service + } + + envwRequest struct { + requestType + env *tpr.Environment + envList []tpr.Environment + respChan chan envwResponse + } + + envwResponse struct { + builderInfo *builderInfo + err error + } + + environmentWatcher struct { + cache map[string]*builderInfo + requestChan chan envwRequest + builderNamespace string + fissionClient *tpr.FissionClient + kubernetesClient *kubernetes.Clientset + fetcherImage string + fetcherImagePullPolicy v1.PullPolicy + } +) + +func makeEnvironmentWatcher(fissionClient *tpr.FissionClient, + kubernetesClient *kubernetes.Clientset, builderNamespace string) *environmentWatcher { + + 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" + } + + var pullPolicy v1.PullPolicy + switch fetcherImagePullPolicy { + case "Always": + pullPolicy = v1.PullAlways + case "Never": + pullPolicy = v1.PullNever + default: + pullPolicy = v1.PullIfNotPresent + } + + envWatcher := &environmentWatcher{ + cache: make(map[string]*builderInfo), + requestChan: make(chan envwRequest), + builderNamespace: builderNamespace, + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + fetcherImage: fetcherImage, + fetcherImagePullPolicy: pullPolicy, + } + + go envWatcher.service() + + return envWatcher +} + +func (envw *environmentWatcher) getCacheKey(envName string, envResourceVersion string) string { + return fmt.Sprintf("%v-%v", envName, envResourceVersion) +} + +func (envw *environmentWatcher) getLabels(envName string, envResourceVersion string) map[string]string { + return map[string]string{ + LABEL_ENV_NAME: envName, + LABEL_ENV_RESOURCEVERSION: envResourceVersion, + } +} + +func (envw *environmentWatcher) watchEnvironments() { + rv := "" + for { + wi, err := envw.fissionClient.Environments(api.NamespaceAll).Watch(api.ListOptions{ + ResourceVersion: rv, + }) + if err != nil { + log.Fatalf("Error watching environment list: %v", err) + } + + for { + ev, more := <-wi.ResultChan() + if !more { + // restart watch from last rv + break + } + if ev.Type == watch.Error { + // restart watch from the start + rv = "" + time.Sleep(time.Second) + break + } + env := ev.Object.(*tpr.Environment) + rv = env.Metadata.ResourceVersion + envw.sync() + } + } +} + +func (envw *environmentWatcher) sync() { + envList, err := envw.fissionClient.Environments(api.NamespaceAll).List(api.ListOptions{}) + if err != nil { + log.Fatalf("Error syncing environment TPR resources: %v", err) + } + + // Create environment builders for all environments + for i := range envList.Items { + env := envList.Items[i] + + if env.Spec.Version == 1 || // builder is not supported with v1 interface + len(env.Spec.Builder.Image) == 0 { // ignore env without builder image + continue + } + _, err := envw.getEnvBuilder(&env) + if err != nil { + log.Printf("Error creating builder for %v: %v", env.Metadata.Name, err) + } + } + envw.cleanupEnvBuilders(envList.Items) +} + +func (envw *environmentWatcher) service() { + for { + req := <-envw.requestChan + switch req.requestType { + case GET_BUILDER: + key := envw.getCacheKey(req.env.Metadata.Name, req.env.Metadata.ResourceVersion) + builderInfo, ok := envw.cache[key] + if !ok { + builderInfo, err := envw.createBuilder(req.env) + if err != nil { + req.respChan <- envwResponse{err: err} + continue + } + envw.cache[key] = builderInfo + } + req.respChan <- envwResponse{builderInfo: builderInfo} + + case CLEANUP_BUILDERS: + latestEnvList := make(map[string]*tpr.Environment) + for i := range req.envList { + env := req.envList[i] + key := envw.getCacheKey(env.Metadata.Name, env.Metadata.ResourceVersion) + latestEnvList[key] = &env + } + + // If an environment is deleted when builder manager down, + // the builder belongs to the environment will be out-of- + // control (an orphan builder) since there is no record in + // cache and TPR. We need to iterate over the services & + // deployments to remove both normal and orphan builders. + + svcList, err := envw.getBuilderServiceList(nil) + if err != nil { + log.Println(err.Error()) + } + for _, svc := range svcList { + envName := svc.ObjectMeta.Labels[LABEL_ENV_NAME] + envResourceVersion := svc.ObjectMeta.Labels[LABEL_ENV_RESOURCEVERSION] + key := envw.getCacheKey(envName, envResourceVersion) + if _, ok := latestEnvList[key]; !ok { + err := envw.deleteBuilderService(svc.ObjectMeta.Labels) + if err != nil { + log.Printf("Error removing builder service: %v", err) + } + } + delete(envw.cache, svc.ObjectMeta.Name) + } + + deployList, err := envw.getBuilderDeploymentList(nil) + if err != nil { + log.Printf(err.Error()) + } + for _, deploy := range deployList { + envName := deploy.ObjectMeta.Labels[LABEL_ENV_NAME] + envResourceVersion := deploy.ObjectMeta.Labels[LABEL_ENV_RESOURCEVERSION] + key := envw.getCacheKey(envName, envResourceVersion) + if _, ok := latestEnvList[key]; !ok { + err := envw.deleteBuilderDeployment(deploy.ObjectMeta.Labels) + if err != nil { + log.Printf("Error removing builder deployment: %v", err) + } + } + delete(envw.cache, deploy.ObjectMeta.Name) + } + } + } +} + +func (envw *environmentWatcher) getEnvBuilder(env *tpr.Environment) (*builderInfo, error) { + respChan := make(chan envwResponse) + envw.requestChan <- envwRequest{ + requestType: GET_BUILDER, + env: env, + respChan: respChan, + } + resp := <-respChan + return resp.builderInfo, resp.err +} + +func (envw *environmentWatcher) cleanupEnvBuilders(envs []tpr.Environment) { + envw.requestChan <- envwRequest{ + requestType: CLEANUP_BUILDERS, + envList: envs, + } +} + +func (envw *environmentWatcher) createBuilder(env *tpr.Environment) (*builderInfo, error) { + var svc *v1.Service + var deploy *v1beta1.Deployment + + sel := envw.getLabels(env.Metadata.Name, env.Metadata.ResourceVersion) + + svcList, err := envw.getBuilderServiceList(sel) + if err != nil { + return nil, err + } + if len(svcList) == 0 { + svc, err = envw.createBuilderService(env) + if err != nil { + return nil, fmt.Errorf("Error creating builder service: %v", err) + } + } + + deployList, err := envw.getBuilderDeploymentList(sel) + if err != nil { + return nil, err + } + if len(deployList) == 0 { + deploy, err = envw.createBuilderDeployment(env) + if err != nil { + return nil, fmt.Errorf("Error creating builder deployment: %v", err) + } + } + + return &builderInfo{ + envMetadata: &env.Metadata, + service: svc, + deployment: deploy, + }, nil +} + +func (envw *environmentWatcher) deleteBuilderService(sel map[string]string) error { + svcList, err := envw.getBuilderServiceList(sel) + if err != nil { + return err + } + for _, svc := range svcList { + log.Printf("Removing builder service: %v", svc.ObjectMeta.Name) + + // cascading deletion + // https://kubernetes.io/docs/concepts/workloads/controllers/garbage-collection/ + falseVal := false + delOpt := &api.DeleteOptions{ + OrphanDependents: &falseVal, + } + + err = envw.kubernetesClient. + Services(envw.builderNamespace). + Delete(svc.ObjectMeta.Name, delOpt) + if err != nil { + return fmt.Errorf("Error deleting builder service: %v", err) + } + } + return nil +} + +func (envw *environmentWatcher) deleteBuilderDeployment(sel map[string]string) error { + deployList, err := envw.getBuilderDeploymentList(sel) + if err != nil { + return err + } + for _, deploy := range deployList { + log.Printf("Removing builder deployment: %v", deploy.ObjectMeta.Name) + + falseVal := false + delOpt := &api.DeleteOptions{ + OrphanDependents: &falseVal, + } + + err = envw.kubernetesClient. + Deployments(envw.builderNamespace). + Delete(deploy.ObjectMeta.Name, delOpt) + if err != nil { + return fmt.Errorf("Error deleteing builder deployment: %v", err) + } + } + return nil +} + +func (envw *environmentWatcher) getBuilderServiceList(sel map[string]string) ([]v1.Service, error) { + svcList, err := envw.kubernetesClient.Services(envw.builderNamespace).List( + api.ListOptions{ + LabelSelector: labels.Set(sel).AsSelector(), + }) + if err != nil { + return nil, fmt.Errorf("Error getting builder service list: %v", err) + } + return svcList.Items, nil +} + +func (envw *environmentWatcher) createBuilderService(env *tpr.Environment) (*v1.Service, error) { + name := envw.getCacheKey(env.Metadata.Name, env.Metadata.ResourceVersion) + sel := envw.getLabels(env.Metadata.Name, env.Metadata.ResourceVersion) + service := v1.Service{ + ObjectMeta: v1.ObjectMeta{ + Namespace: envw.builderNamespace, + Name: name, + Labels: sel, + }, + Spec: v1.ServiceSpec{ + Selector: sel, + Type: v1.ServiceTypeClusterIP, + Ports: []v1.ServicePort{ + { + Name: "fetcher-port", + Protocol: v1.ProtocolTCP, + Port: 8000, + TargetPort: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8000, + }, + }, + { + Name: "builder-port", + Protocol: v1.ProtocolTCP, + Port: 8001, + TargetPort: intstr.IntOrString{ + Type: intstr.Int, + IntVal: 8001, + }, + }, + }, + }, + } + log.Printf("Creating builder service: %v", name) + _, err := envw.kubernetesClient.Services(envw.builderNamespace).Create(&service) + if err != nil { + return nil, err + } + return &service, nil +} + +func (envw *environmentWatcher) getBuilderDeploymentList(sel map[string]string) ([]v1beta1.Deployment, error) { + deployList, err := envw.kubernetesClient.Deployments(envw.builderNamespace).List( + api.ListOptions{ + LabelSelector: labels.Set(sel).AsSelector(), + }) + if err != nil { + return nil, fmt.Errorf("Error getting builder deployment list: %v", err) + } + return deployList.Items, nil +} + +func (envw *environmentWatcher) createBuilderDeployment(env *tpr.Environment) (*v1beta1.Deployment, error) { + sharedMountPath := "/package" + name := envw.getCacheKey(env.Metadata.Name, env.Metadata.ResourceVersion) + sel := envw.getLabels(env.Metadata.Name, env.Metadata.ResourceVersion) + var replicas int32 = 1 + deployment := &v1beta1.Deployment{ + ObjectMeta: v1.ObjectMeta{ + Namespace: envw.builderNamespace, + Name: name, + Labels: sel, + }, + Spec: v1beta1.DeploymentSpec{ + Replicas: &replicas, + Selector: &v1beta1.LabelSelector{ + MatchLabels: sel, + }, + Template: v1.PodTemplateSpec{ + ObjectMeta: v1.ObjectMeta{ + Labels: sel, + }, + Spec: v1.PodSpec{ + Volumes: []v1.Volume{ + { + Name: "package", + VolumeSource: v1.VolumeSource{ + EmptyDir: &v1.EmptyDirVolumeSource{}, + }, + }, + }, + Containers: []v1.Container{ + { + Name: "builder", + Image: env.Spec.Builder.Image, + ImagePullPolicy: v1.PullAlways, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []v1.VolumeMount{ + { + Name: "package", + MountPath: sharedMountPath, + }, + }, + Command: []string{"/builder", sharedMountPath}, + }, + { + Name: "fetcher", + Image: envw.fetcherImage, + ImagePullPolicy: envw.fetcherImagePullPolicy, + TerminationMessagePath: "/dev/termination-log", + VolumeMounts: []v1.VolumeMount{ + { + Name: "package", + MountPath: sharedMountPath, + }, + }, + Command: []string{"/fetcher", sharedMountPath}, + }, + }, + ServiceAccountName: "fission-builder", + }, + }, + }, + } + log.Printf("Creating builder deployment: %v", envw.getCacheKey(env.Metadata.Name, env.Metadata.ResourceVersion)) + _, err := envw.kubernetesClient.Deployments(envw.builderNamespace).Create(deployment) + if err != nil { + return nil, err + } + return deployment, nil +} diff --git a/buildermgr/pkgwatcher.go b/buildermgr/pkgwatcher.go new file mode 100644 index 00000000..a4819e6b --- /dev/null +++ b/buildermgr/pkgwatcher.go @@ -0,0 +1,91 @@ +/* +Copyright 2017 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 buildermgr + +import ( + "log" + "time" + + "k8s.io/client-go/1.5/kubernetes" + "k8s.io/client-go/1.5/pkg/api" + "k8s.io/client-go/1.5/pkg/watch" + + "github.com/fission/fission" + "github.com/fission/fission/tpr" +) + +type ( + packageWatcher struct { + fissionClient *tpr.FissionClient + kubernetesClient *kubernetes.Clientset + builderNamespace string + storageSvcUrl string + } +) + +func makePackageWatcher(fissionClient *tpr.FissionClient, + kubernetesClient *kubernetes.Clientset, builderNamespace string, storageSvcUrl string) *packageWatcher { + pkgw := &packageWatcher{ + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + builderNamespace: builderNamespace, + storageSvcUrl: storageSvcUrl, + } + return pkgw +} + +func (pkgw *packageWatcher) build(pkgMetadata api.ObjectMeta) { + buildReq := BuildRequest{ + Package: pkgMetadata, + } + _, err := buildPackage(pkgw.fissionClient, + pkgw.kubernetesClient, pkgw.builderNamespace, pkgw.storageSvcUrl, buildReq) + if err != nil { + log.Printf("Error building package %v: %v", buildReq.Package.Name, err) + } +} + +func (pkgw *packageWatcher) watchPackages() { + rv := "" + for { + wi, err := pkgw.fissionClient.Packages(api.NamespaceDefault).Watch(api.ListOptions{ + ResourceVersion: rv, + }) + if err != nil { + log.Fatalf("Error watching package TPR resources: %v", err) + } + + for { + ev, more := <-wi.ResultChan() + if !more { + break + } + if ev.Type == watch.Error { + rv = "" + time.Sleep(time.Second) + break + } + pkg := ev.Object.(*tpr.Package) + rv = pkg.Metadata.ResourceVersion + + // only do build for packages in pending state + if pkg.Status.BuildStatus == fission.BuildStatusPending { + go pkgw.build(pkg.Metadata) + } + } + } +} diff --git a/charts/README.md b/charts/README.md index e9cb6951..4075a4b5 100644 --- a/charts/README.md +++ b/charts/README.md @@ -47,6 +47,7 @@ The following table lists the configurable parameters of the Fission chart and t | `controllerPort` | Fission Controller Service Port | `31313` | | `routerPort` | Fission Router Service Port | `31314` | | `functionNamespace` | Namespace for Fission functions | `fission-function` | +| `builderNamespace` | Namespace for Fission environment builders | `fission-builder` | | `openshift` | RBAC for openshift | `false` | diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index f51c94e8..b6e9d00f 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -81,6 +81,15 @@ metadata: name: fission-function chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +--- +apiVersion: v1 +kind: Namespace +metadata: + name: {{ .Values.builderNamespace }} + labels: + name: fission-builder + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + --- apiVersion: v1 kind: ServiceAccount @@ -153,6 +162,27 @@ roleRef: name: cluster-admin apiGroup: rbac.authorization.k8s.io +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: fission-builder + namespace: {{ .Values.builderNamespace }} + +--- +kind: ClusterRoleBinding +apiVersion: rbac.authorization.k8s.io/v1beta1 +metadata: + name: fission-builder-tpr +subjects: +- kind: ServiceAccount + name: fission-builder + namespace: {{ .Values.builderNamespace }} +roleRef: + kind: ClusterRole + name: cluster-admin + apiGroup: rbac.authorization.k8s.io + {{ end }} --- @@ -240,6 +270,51 @@ spec: value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" - name: FETCHER_IMAGE_PULL_POLICY value: "{{ .Values.pullPolicy }}" + - name: RUNTIME_IMAGE_PULL_POLICY + value: "{{ .Values.pullPolicy }}" + serviceAccount: fission-svc + +--- +apiVersion: v1 +kind: Service +metadata: + name: buildermgr + labels: + svc: buildermgr + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + type: ClusterIP + ports: + - port: 80 + targetPort: 8889 + selector: + svc: buildermgr + +--- +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: buildermgr + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + replicas: 1 + template: + metadata: + labels: + svc: buildermgr + spec: + containers: + - name: buildermgr + image: "{{ .Values.image }}:{{ .Values.imageTag }}" + imagePullPolicy: {{ .Values.pullPolicy }} + command: ["/fission-bundle"] + args: ["--builderMgrPort", "8889", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] + env: + - name: FETCHER_IMAGE + value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" + - name: FETCHER_IMAGE_PULL_POLICY + value: "{{ .Values.pullPolicy }}" serviceAccount: fission-svc --- diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index 6ae5070e..5e8b23e0 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -34,6 +34,10 @@ natsStreamingPort: 31316 ## the release namespace) functionNamespace: fission-function +## Namespace in which to run fission builders (this is different from +## the release namespace) +builderNamespace: fission-builder + ## Set up openshift RBAC rule openshift: false diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 2589729b..635c965d 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -81,6 +81,15 @@ metadata: name: fission-function chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +--- +apiVersion: v1 +kind: Namespace +metadata: + name: {{ .Values.builderNamespace }} + labels: + name: fission-builder + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + --- apiVersion: v1 kind: ServiceAccount @@ -153,6 +162,27 @@ roleRef: name: cluster-admin apiGroup: rbac.authorization.k8s.io +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: fission-builder + namespace: {{ .Values.builderNamespace }} + +--- +kind: ClusterRoleBinding +apiVersion: rbac.authorization.k8s.io/v1beta1 +metadata: + name: fission-builder-tpr +subjects: +- kind: ServiceAccount + name: fission-builder + namespace: {{ .Values.builderNamespace }} +roleRef: + kind: ClusterRole + name: cluster-admin + apiGroup: rbac.authorization.k8s.io + {{ end }} --- @@ -242,6 +272,49 @@ spec: value: "{{ .Values.pullPolicy }}" serviceAccount: fission-svc +--- +apiVersion: v1 +kind: Service +metadata: + name: buildermgr + labels: + svc: buildermgr + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + type: ClusterIP + ports: + - port: 80 + targetPort: 8889 + selector: + svc: buildermgr + +--- +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: buildermgr + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" +spec: + replicas: 1 + template: + metadata: + labels: + svc: buildermgr + spec: + containers: + - name: buildermgr + image: "{{ .Values.image }}:{{ .Values.imageTag }}" + imagePullPolicy: {{ .Values.pullPolicy }} + command: ["/fission-bundle"] + args: ["--builderMgrPort", "8889", "--storageSvcUrl", "http://storagesvc.{{ .Release.Namespace }}", "--envbuilder-namespace", "{{ .Values.builderNamespace }}"] + env: + - name: FETCHER_IMAGE + value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}" + - name: FETCHER_IMAGE_PULL_POLICY + value: "{{ .Values.pullPolicy }}" + serviceAccount: fission-svc + --- apiVersion: extensions/v1beta1 kind: Deployment diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index e3d96bd6..3ab7a880 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -31,6 +31,10 @@ routerPort: 31314 ## the release namespace) functionNamespace: fission-function +## Namespace in which to run fission builders (this is different from +## the release namespace) +builderNamespace: fission-builder + ## Set up openshift RBAC rule openshift: false diff --git a/controller/api.go b/controller/api.go index 57fd86a2..4e256861 100644 --- a/controller/api.go +++ b/controller/api.go @@ -37,6 +37,7 @@ type ( API struct { fissionClient *tpr.FissionClient storageServiceUrl string + builderManagerUrl string } logDBConfig struct { @@ -56,6 +57,13 @@ func MakeAPI() (*API, error) { api.storageServiceUrl = "http://storagesvc" } + u = os.Getenv("BUILDER_MANAGER_URL") + if len(u) > 0 { + api.builderManagerUrl = strings.TrimSuffix(u, "/") + } else { + api.builderManagerUrl = "http://buildermgr" + } + return api, err } @@ -160,6 +168,8 @@ func (api *API) Serve(port int) { r.HandleFunc("/proxy/{dbType}", api.FunctionLogsApiPost).Methods("POST") r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy) + r.HandleFunc("/proxy/buildermgr/v1/build", api.BuilderManagerBuildProxy) + r.HandleFunc("/proxy/buildermgr/v1/builder", api.BuilderManagerEnvBuilderProxy) address := fmt.Sprintf(":%v", port) diff --git a/controller/buildermgr.go b/controller/buildermgr.go new file mode 100644 index 00000000..4e955267 --- /dev/null +++ b/controller/buildermgr.go @@ -0,0 +1,67 @@ +/* +Copyright 2017 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 controller + +import ( + "fmt" + "log" + "net/http" + "net/http/httputil" + "net/url" +) + +func (api *API) BuilderManagerBuildProxy(w http.ResponseWriter, r *http.Request) { + u := api.builderManagerUrl + "/v1/build" + proxy, err := api._getBuilderManagerProxy(u) + if err != nil { + msg := fmt.Sprintf("Failed to establish proxy server: %v", err) + log.Println(msg) + http.Error(w, msg, 500) + return + } + proxy.ServeHTTP(w, r) +} + +func (api *API) BuilderManagerEnvBuilderProxy(w http.ResponseWriter, r *http.Request) { + u := api.builderManagerUrl + "/v1/builder" + proxy, err := api._getBuilderManagerProxy(u) + if err != nil { + msg := fmt.Sprintf("Failed to establish proxy server: %v", err) + log.Println(msg) + http.Error(w, msg, 500) + return + } + proxy.ServeHTTP(w, r) +} + +func (api *API) _getBuilderManagerProxy(targetUrl string) (*httputil.ReverseProxy, error) { + svcUrl, err := url.Parse(targetUrl) + if err != nil { + return nil, err + } + // set up proxy server director + director := func(req *http.Request) { + // only replace url Scheme and Host to remote server + // and leave query string intact + req.URL.Scheme = svcUrl.Scheme + req.URL.Host = svcUrl.Host + req.URL.Path = svcUrl.Path + } + return &httputil.ReverseProxy{ + Director: director, + }, nil +} diff --git a/controller/environmentApi.go b/controller/environmentApi.go index 3840d6cf..e397a970 100644 --- a/controller/environmentApi.go +++ b/controller/environmentApi.go @@ -139,6 +139,7 @@ func (a *API) EnvironmentApiUpdate(w http.ResponseWriter, r *http.Request) { a.respondWithError(w, err) return } + a.respondWithSuccess(w, resp) } diff --git a/environments/fetcher/client/client.go b/environments/fetcher/client/client.go index e31897ca..1713ec79 100644 --- a/environments/fetcher/client/client.go +++ b/environments/fetcher/client/client.go @@ -3,6 +3,8 @@ package client import ( "bytes" "encoding/json" + "io/ioutil" + "log" "net/http" //"time" @@ -11,7 +13,19 @@ import ( //"github.com/fission/fission/router" ) -func DoFetchRequest(fetcherUrl string, fr *fetcher.FetchRequest) error { +type ( + Client struct { + url string + } +) + +func MakeClient(fetcherUrl string) *Client { + return &Client{ + url: fetcherUrl, + } +} + +func (c *Client) Fetch(fr *fetcher.FetchRequest) error { body, err := json.Marshal(fr) if err != nil { return err @@ -21,7 +35,7 @@ func DoFetchRequest(fetcherUrl string, fr *fetcher.FetchRequest) error { // Transport: router.MakeRetryingRoundTripper(10, 50*time.Millisecond), // } - resp, err := http.Post(fetcherUrl, "application/json", bytes.NewReader(body)) + resp, err := http.Post(c.url, "application/json", bytes.NewReader(body)) if err != nil { return err } @@ -33,3 +47,34 @@ func DoFetchRequest(fetcherUrl string, fr *fetcher.FetchRequest) error { return nil } + +func (c *Client) Upload(fr *fetcher.UploadRequest) (*fetcher.UploadResponse, error) { + body, err := json.Marshal(fr) + if err != nil { + return nil, err + } + resp, err := http.Post(c.url+"/upload", "application/json", bytes.NewReader(body)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return nil, fission.MakeErrorFromHTTP(resp) + } + + rBody, err := ioutil.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + log.Printf("Received upload response: %v", string(rBody)) + + uploadReq := fetcher.UploadResponse{} + err = json.Unmarshal([]byte(rBody), &uploadReq) + if err != nil { + return nil, err + } + + return &uploadReq, nil +} diff --git a/environments/fetcher/cmd/main.go b/environments/fetcher/cmd/main.go index 05ccaafd..638c9a6a 100644 --- a/environments/fetcher/cmd/main.go +++ b/environments/fetcher/cmd/main.go @@ -21,6 +21,7 @@ func main() { } fetcher := fetcher.MakeFetcher(dir) mux := http.NewServeMux() - mux.HandleFunc("/", fetcher.Handler) + mux.HandleFunc("/", fetcher.FetchHandler) + mux.HandleFunc("/upload", fetcher.UploadHandler) http.ListenAndServe(":8000", mux) } diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index 7f3c0426..8a9558a5 100644 --- a/environments/fetcher/fetcher.go +++ b/environments/fetcher/fetcher.go @@ -20,6 +20,7 @@ import ( "k8s.io/client-go/1.5/pkg/api" "github.com/fission/fission" + storageSvcClient "github.com/fission/fission/storagesvc/client" "github.com/fission/fission/tpr" ) @@ -28,12 +29,26 @@ type ( FetchRequest struct { FetchType FetchRequestType `json:"fetchType"` - Function api.ObjectMeta `json:"function"` + Package api.ObjectMeta `json:"package"` Url string `json:"url"` StorageSvcUrl string `json:"storagesvcurl"` Filename string `json:"filename"` } + // UploadRequest send from builder manager describes which + // deployment package should be upload to storage service. + UploadRequest struct { + Filename string `json:"filename"` + StorageSvcUrl string `json:"storagesvcurl"` + } + + // UploadResponse defines the download url of an archive and + // its checksum. + UploadResponse struct { + ArchiveDownloadUrl string `json:"archiveDownloadUrl"` + Checksum fission.Checksum `json:"checksum"` + } + Fetcher struct { sharedVolumePath string fissionClient *tpr.FissionClient @@ -79,33 +94,44 @@ func downloadUrl(url string, localPath string) error { return nil } -func verifyChecksum(path string, checksum *fission.Checksum) error { - if checksum.Type != fission.ChecksumTypeSHA256 { - return fission.MakeError(fission.ErrorInvalidArgument, "Unsupported checksum type") - } - +func getChecksum(path string) (*fission.Checksum, error) { f, err := os.Open(path) if err != nil { - return err + return nil, err } defer f.Close() hasher := sha256.New() _, err = io.Copy(hasher, f) if err != nil { - return err + return nil, err } c := hex.EncodeToString(hasher.Sum(nil)) - if c != checksum.Sum { + + return &fission.Checksum{ + Type: fission.ChecksumTypeSHA256, + Sum: c, + }, nil +} + +func verifyChecksum(path string, checksum *fission.Checksum) error { + if checksum.Type != fission.ChecksumTypeSHA256 { + return fission.MakeError(fission.ErrorInvalidArgument, "Unsupported checksum type") + } + c, err := getChecksum(path) + if err != nil { + return err + } + if c.Sum != checksum.Sum { return fission.MakeError(fission.ErrorChecksumFail, "Checksum validation failed") } return nil } -func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) { +func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { - http.Error(w, "", 405) + http.Error(w, "only POST is supported on this endpoint", 405) return } @@ -129,11 +155,13 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), 400) return } - log.Printf("fetcher received request: %v", req) + log.Printf("fetcher received fetch request: %v", req) tmpFile := req.Filename + ".tmp" tmpPath := filepath.Join(fetcher.sharedVolumePath, tmpFile) + log.Printf("Start downloading...") + if req.FetchType == FETCH_URL { // fetch the file and save it to the tmp path err := downloadUrl(req.Url, tmpPath) @@ -144,19 +172,8 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) { return } } else { - // get function object - fn, err := fetcher.fissionClient.Functions(req.Function.Namespace).Get(req.Function.Name) - if err != nil { - e := fmt.Sprintf("Failed to get function: %v", err) - log.Printf(e) - http.Error(w, e, 500) - return - } - // get pkg - var pkg *tpr.Package - pkg, err = fetcher.fissionClient. - Packages(fn.Spec.Package.PackageRef.Namespace).Get(fn.Spec.Package.PackageRef.Name) + pkg, err := fetcher.fissionClient.Packages(req.Package.Namespace).Get(req.Package.Name) if err != nil { e := fmt.Sprintf("Failed to get package: %v", err) log.Printf(e) @@ -200,7 +217,6 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) { return } } - } // check file type here, if the file is a zip file unarchive it. @@ -229,6 +245,85 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } +func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + http.Error(w, "only POST is supported on this endpoint", 405) + return + } + + startTime := time.Now() + defer func() { + elapsed := time.Now().Sub(startTime) + log.Printf("elapsed time in upload request = %v", elapsed) + }() + + // parse request + body, err := ioutil.ReadAll(r.Body) + if err != nil { + log.Printf("Error reading request body") + http.Error(w, err.Error(), 500) + return + } + + var req UploadRequest + err = json.Unmarshal(body, &req) + if err != nil { + log.Printf("Error reading request body: %v", err) + http.Error(w, err.Error(), 400) + return + } + log.Printf("fetcher received upload request: %v", req) + + zipFilename := req.Filename + ".zip" + srcFilepath := filepath.Join(fetcher.sharedVolumePath, req.Filename) + dstFilepath := filepath.Join(fetcher.sharedVolumePath, zipFilename) + + err = fetcher.archive(srcFilepath, dstFilepath) + if err != nil { + e := fmt.Sprintf("Error archiving zip file: %v", err) + log.Println(e) + http.Error(w, e, 500) + return + } + + log.Println("Start uploading...") + ssClient := storageSvcClient.MakeClient(req.StorageSvcUrl) + + fileID, err := ssClient.Upload(dstFilepath, nil) + if err != nil { + e := fmt.Sprintf("Error uploading zip file: %v", err) + log.Println(e) + http.Error(w, e, 500) + return + } + + sum, err := getChecksum(dstFilepath) + if err != nil { + e := fmt.Sprintf("Error calculating checksum of zip file: %v", err) + log.Println(e) + http.Error(w, e, 500) + return + } + + resp := UploadResponse{ + ArchiveDownloadUrl: ssClient.GetUrl(fileID), + Checksum: *sum, + } + + rBody, err := json.Marshal(resp) + if err != nil { + e := fmt.Sprintf("Error encoding upload response: %v", err) + log.Println(e) + http.Error(w, e, 500) + return + } + + log.Println("Completed upload request") + w.Header().Add("Content-Type", "application/json") + w.Write(rBody) + w.WriteHeader(http.StatusOK) +} + func (fetcher *Fetcher) rename(src string, dst string) error { err := os.Rename(src, dst) if err != nil { @@ -239,7 +334,21 @@ func (fetcher *Fetcher) rename(src string, dst string) error { // archive is a function that zips directory into a zip file func (fetcher *Fetcher) archive(src string, dst string) error { - return archiver.Zip.Make(dst, []string{src}) + var files []string + target, err := os.Stat(src) + if err != nil { + return errors.New(fmt.Sprintf("Failed to zip file: %v", err)) + } + if target.IsDir() { + // list all + fs, _ := ioutil.ReadDir(src) + for _, f := range fs { + files = append(files, filepath.Join(src, f.Name())) + } + } else { + files = append(files, src) + } + return archiver.Zip.Make(dst, files) } // unarchive is a function that unzips a zip file to destination diff --git a/environments/python3/builder/Dockerfile b/environments/python3/builder/Dockerfile index 2cb3c9cc..4aa32245 100644 --- a/environments/python3/builder/Dockerfile +++ b/environments/python3/builder/Dockerfile @@ -6,6 +6,6 @@ RUN pip3 install --upgrade pip RUN rm -r /root/.cache ADD defaultBuildCmd /usr/local/bin/build -ADD builder / +ADD builder /builder -EXPOSE 8000 +EXPOSE 8001 diff --git a/environments/python3/builder/build.sh b/environments/python3/builder/build.sh deleted file mode 100755 index b70d3580..00000000 --- a/environments/python3/builder/build.sh +++ /dev/null @@ -1,11 +0,0 @@ -#!/bin/sh -set -e -builderDir=${GOPATH}/src/github.com/fission/fission/builder/cmd -pushd ${builderDir} -GOOS=linux GOARCH=386 go build -o builder . -popd -cp ${builderDir}/builder . -docker build -t python-builder . -docker tag python-builder fission/python-builder:$tag -docker push fission/python-builder:$tag - diff --git a/environments/python3/server.py b/environments/python3/server.py index 91366509..a20b19ae 100644 --- a/environments/python3/server.py +++ b/environments/python3/server.py @@ -3,21 +3,35 @@ import logging import sys import imp +import os from flask import Flask, request, abort, g app = Flask(__name__) -codepath = '/userfunc/user' - userfunc = None @app.route('/specialize', methods=['POST']) def load(): global userfunc + # load user function from codepath + codepath = '/userfunc/user' userfunc = (imp.load_source('user', codepath)).main return "" +@app.route('/v2/specialize', methods=['POST']) +def loadv2(): + global userfunc + body = request.get_json() + filepath = body['filepath'] + functionName = body['functionName'] + # add filepath into syspath for module import + sys.path.append(filepath) + fn, path, desc = imp.find_module('user', [filepath]) + mod = imp.load_module('user', fn, path, desc) + userfunc = getattr(mod, functionName) + return "" + @app.route('/', methods=['GET', 'POST', 'PUT', 'HEAD', 'OPTIONS', 'DELETE']) def f(): if userfunc == None: diff --git a/examples/python/sourcepkg/__init__.py b/examples/python/sourcepkg/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/examples/python/sourcepkg/build.sh b/examples/python/sourcepkg/build.sh new file mode 100755 index 00000000..e9215130 --- /dev/null +++ b/examples/python/sourcepkg/build.sh @@ -0,0 +1,2 @@ +#!/bin/sh +pip3 install -r ${SRC_PKG}/requirements.txt -t ${SRC_PKG} && cp -r ${SRC_PKG} ${DEPLOY_PKG} \ No newline at end of file diff --git a/examples/python/sourcepkg/requirements.txt b/examples/python/sourcepkg/requirements.txt new file mode 100644 index 00000000..c3726e8b --- /dev/null +++ b/examples/python/sourcepkg/requirements.txt @@ -0,0 +1 @@ +pyyaml diff --git a/examples/python/sourcepkg/user.py b/examples/python/sourcepkg/user.py new file mode 100644 index 00000000..3c198aca --- /dev/null +++ b/examples/python/sourcepkg/user.py @@ -0,0 +1,12 @@ +import sys +import yaml + +document = """ + a: 1 + b: + c: 3 + d: 4 +""" + +def main(): + return yaml.dump(yaml.load(document)) diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 14639059..c652819b 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -6,6 +6,7 @@ import ( "strconv" "github.com/docopt/docopt-go" + "github.com/fission/fission/buildermgr" "github.com/fission/fission/controller" "github.com/fission/fission/kubewatcher" "github.com/fission/fission/logger" @@ -68,6 +69,13 @@ func runStorageSvc(port int, filePath string) { filePath, subdir, port) } +func runBuilderMgr(port int, storageSvcUrl string, envBuilderNamespace string) { + err := buildermgr.Start(port, storageSvcUrl, envBuilderNamespace) + if err != nil { + log.Fatalf("Error starting buildermgr: %v", err) + } +} + func getPort(portArg interface{}) int { portArgStr := portArg.(string) port, err := strconv.Atoi(portArgStr) @@ -113,23 +121,26 @@ Usage: fission-bundle --poolmgrPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] fission-bundle --storageServicePort= --filePath= + fission-bundle --builderMgrPort= [--storageSvcUrl=] [--envbuilder-namespace=] fission-bundle --logger fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] Options: - --controllerPort= Port that the controller should listen on. - --routerPort= Port that the router should listen on. - --poolmgrPort= Port that the poolmgr should listen on. - --storageServicePort= Port that the storage service should listen on. - --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. - --routerUrl= Router URL. - --etcdUrl= Etcd URL. - --filePath= Directory to store functions in. - --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. - --kubewatcher Start Kubernetes events watcher. - --logger Start logger. - --timer Start Timer. - --mqt Start message queue trigger. + --controllerPort= Port that the controller should listen on. + --routerPort= Port that the router should listen on. + --poolmgrPort= Port that the poolmgr should listen on. + --storageServicePort= Port that the storage service should listen on. + --builderMgrPort= Port that the buildermgr should listen on. + --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. + --routerUrl= Router URL. + --etcdUrl= Etcd URL. + --storageSvcUrl= StorageService URL. + --filePath= Directory to store functions in. + --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. + --kubewatcher Start Kubernetes events watcher. + --logger Start logger. + --timer Start Timer. + --mqt Start message queue trigger. ` arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false) if err != nil { @@ -138,9 +149,11 @@ Options: functionNs := getStringArgWithDefault(arguments["--namespace"], "fission-function") fissionNs := getStringArgWithDefault(arguments["--fission-namespace"], "fission") + envBuilderNs := getStringArgWithDefault(arguments["--envbuilder-namespace"], "fission-builder") poolmgrUrl := getStringArgWithDefault(arguments["--poolmgrUrl"], "http://poolmgr.fission") routerUrl := getStringArgWithDefault(arguments["--routerUrl"], "http://router.fission") + storageSvcUrl := getStringArgWithDefault(arguments["--storageSvcUrl"], "http://storagesvc.fission") if arguments["--controllerPort"] != nil { port := getPort(arguments["--controllerPort"]) @@ -179,5 +192,10 @@ Options: runStorageSvc(port, filePath) } + if arguments["--builderMgrPort"] != nil { + port := getPort(arguments["--builderMgrPort"]) + runBuilderMgr(port, storageSvcUrl, envBuilderNs) + } + select {} } diff --git a/fission/environment.go b/fission/environment.go index 422b95ad..14ee3424 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -18,7 +18,6 @@ package main import ( "fmt" - "log" "os" "text/tabwriter" @@ -42,12 +41,22 @@ func envCreate(c *cli.Context) error { fatal("Need an image, use --image.") } + envVersion := c.Int("version") envBuilderImg := c.String("builder") - envBuildCmd := c.String("buildcmd") - if len(envBuilderImg) > 0 && len(envBuildCmd) == 0 { - log.Printf("No build command is specified, use the default build command.") - envBuildCmd = "build" + + if len(envBuilderImg) > 0 { + envVersion = 2 + if len(envBuildCmd) == 0 { + envBuildCmd = "build" + } + } + + // Environment API interface version is not specified and + // builder image is empty, set default interface version + if envVersion == 0 { + fmt.Println("Use default environment v1 API interface") + envVersion = 1 } env := &tpr.Environment{ @@ -56,7 +65,7 @@ func envCreate(c *cli.Context) error { Namespace: api.NamespaceDefault, }, Spec: fission.EnvironmentSpec{ - Version: 1, + Version: envVersion, Runtime: fission.Runtime{ Image: envImg, }, @@ -121,6 +130,11 @@ func envUpdate(c *cli.Context) error { if len(envImg) > 0 { env.Spec.Runtime.Image = envImg } + + if env.Spec.Version == 1 && (len(envBuilderImg) > 0 || len(envBuildCmd) > 0) { + fatal("Environment v1 API interface doesn't supported environment builder.") + } + if len(envBuilderImg) > 0 { env.Spec.Builder.Image = envBuilderImg } diff --git a/fission/function.go b/fission/function.go index 34489c58..893e0708 100644 --- a/fission/function.go +++ b/fission/function.go @@ -28,7 +28,6 @@ import ( "text/tabwriter" "time" - "github.com/dchest/uniuri" "github.com/satori/go.uuid" "github.com/urfave/cli" "k8s.io/client-go/1.5/pkg/api" @@ -85,6 +84,47 @@ func createArchive(client *client.Client, fileName string) *fission.Archive { return &archive } +func createPackage(client *client.Client, envName, srcArchiveName, deployArchiveName, buildcmd string) *api.ObjectMeta { + pkgSpec := fission.PackageSpec{ + Environment: fission.EnvironmentReference{ + Namespace: api.NamespaceDefault, + Name: envName, + }, + } + var pkgStatus fission.BuildStatus = fission.BuildStatusSucceeded + + if len(deployArchiveName) > 0 { + pkgSpec.Deployment = *createArchive(client, deployArchiveName) + if len(srcArchiveName) > 0 { + fmt.Println("Deployment may be overwritten by builder manager after source package compilation") + } + } + if len(srcArchiveName) > 0 { + pkgSpec.Source = *createArchive(client, srcArchiveName) + // set pending status to package + pkgStatus = fission.BuildStatusPending + } + + if len(buildcmd) > 0 { + pkgSpec.BuildCommand = buildcmd + } + + pkgName := strings.ToLower(uuid.NewV4().String()) + pkg := &tpr.Package{ + Metadata: api.ObjectMeta{ + Name: pkgName, + Namespace: api.NamespaceDefault, + }, + Spec: pkgSpec, + Status: fission.PackageStatus{ + BuildStatus: pkgStatus, + }, + } + pkgMetadata, err := client.PackageCreate(pkg) + checkErr(err, "create package") + return pkgMetadata +} + func getContents(filePath string) []byte { var code []byte var err error @@ -97,6 +137,10 @@ func getContents(filePath string) []byte { func fnCreate(c *cli.Context) error { client := getClient(c.GlobalString("server")) + if len(c.String("package")) > 0 { + fatal("--package is deprecated, please use --deploy instead.") + } + fnName := c.String("name") if len(fnName) == 0 { fatal("Need --name argument.") @@ -107,17 +151,24 @@ func fnCreate(c *cli.Context) error { fatal("Need --env argument.") } - srcPkgName := c.String("srcpkg") - - deployPkgName := c.String("code") - if len(deployPkgName) == 0 { - deployPkgName = c.String("package") + srcArchiveName := c.String("src") + deployArchiveName := c.String("code") + if len(deployArchiveName) == 0 { + deployArchiveName = c.String("deploy") } - if len(srcPkgName) == 0 && len(deployPkgName) == 0 { - fatal("Need --code or --package to specify deployment package, or use --srcpkg to specify source package.") + if len(srcArchiveName) == 0 && len(deployArchiveName) == 0 { + fatal("Need --code or --deploy to specify deployment archive, or use --src to specify source archive.") } + entrypoint := c.String("entrypoint") + buildcmd := c.String("buildcmd") + if len(buildcmd) == 0 { + buildcmd = "/builder" + } + + pkgMetadata := createPackage(client, envName, srcArchiveName, deployArchiveName, buildcmd) + function := &tpr.Function{ Metadata: api.ObjectMeta{ Name: fnName, @@ -128,35 +179,18 @@ func fnCreate(c *cli.Context) error { Name: envName, Namespace: api.NamespaceDefault, }, + Package: fission.FunctionPackageRef{ + FunctionName: entrypoint, + PackageRef: fission.PackageRef{ + Namespace: pkgMetadata.Namespace, + Name: pkgMetadata.Name, + ResourceVersion: pkgMetadata.ResourceVersion, + }, + }, }, } - var pkgSpec fission.PackageSpec - if len(srcPkgName) > 0 { - pkgSpec.Source = *createArchive(client, srcPkgName) - } - if len(deployPkgName) > 0 { - pkgSpec.Deployment = *createArchive(client, deployPkgName) - } - pkgName := fmt.Sprintf("%v-%v", fnName, strings.ToLower(uniuri.NewLen(6))) - pkg := &tpr.Package{ - Metadata: api.ObjectMeta{ - Name: pkgName, - Namespace: api.NamespaceDefault, - }, - Spec: pkgSpec, - } - newpkg, err := client.PackageCreate(pkg) - checkErr(err, "create package") - - function.Spec.Package = fission.FunctionPackageRef{ - PackageRef: fission.PackageRef{ - Name: newpkg.Name, - Namespace: newpkg.Namespace, - ResourceVersion: newpkg.ResourceVersion, - }, - } - _, err = client.FunctionCreate(function) + _, err := client.FunctionCreate(function) checkErr(err, "create function") fmt.Printf("function '%v' created\n", fnName) @@ -248,6 +282,10 @@ func fnUpdate(c *cli.Context) error { fatal("Need name of function, use --name") } + if len(c.String("package")) > 0 { + fatal("--package is deprecated, please use --deploy instead.") + } + function, err := client.FunctionGet(&api.ObjectMeta{ Name: fnName, Namespace: api.NamespaceDefault, @@ -255,42 +293,50 @@ func fnUpdate(c *cli.Context) error { checkErr(err, fmt.Sprintf("read function '%v'", fnName)) envName := c.String("env") - deployPkgName := c.String("code") - if len(deployPkgName) == 0 { - deployPkgName = c.String("package") + deployArchiveName := c.String("code") + if len(deployArchiveName) == 0 { + deployArchiveName = c.String("deploy") } - srcPkgName := c.String("srcpkg") + srcArchiveName := c.String("src") - if len(envName) == 0 && len(deployPkgName) == 0 && len(srcPkgName) == 0 { - fatal("Need --env or --code or --package or --srcpkg argument.") - } - - if len(deployPkgName) != 0 || len(srcPkgName) != 0 { - // get existing package - pkg, err := client.PackageGet(&api.ObjectMeta{ - Name: function.Spec.Package.PackageRef.Name, - Namespace: function.Spec.Package.PackageRef.Namespace, - }) - // update package spec - if len(srcPkgName) > 0 { - archive := createArchive(client, srcPkgName) - pkg.Spec.Source = *archive - } - if len(deployPkgName) > 0 { - archive := createArchive(client, deployPkgName) - pkg.Spec.Deployment = *archive - } - // updage package object - newpkg, err := client.PackageUpdate(pkg) - checkErr(err, "update package") - // update function spec with resource version - function.Spec.Package.PackageRef.ResourceVersion = newpkg.ResourceVersion + if len(envName) == 0 && len(deployArchiveName) == 0 && len(srcArchiveName) == 0 { + fatal("Need --env or --code or --package or --deploy argument.") } if len(envName) > 0 { function.Spec.Environment.Name = envName } + entrypoint := c.String("entrypoint") + if len(entrypoint) > 0 { + function.Spec.Package.FunctionName = entrypoint + } + + pkg, err := client.PackageGet(&api.ObjectMeta{ + Name: function.Spec.Package.PackageRef.Name, + Namespace: function.Spec.Package.PackageRef.Namespace, + }) + checkErr(err, fmt.Sprintf("read package '%v'", function.Spec.Package.PackageRef.Name)) + + buildcmd := c.String("buildcmd") + if len(buildcmd) == 0 { + // use previous build command if not specified. + buildcmd = pkg.Spec.BuildCommand + } + + if len(deployArchiveName) > 0 || len(srcArchiveName) > 0 { + // create a new package for function + pkgMetadata := createPackage(client, + function.Spec.Environment.Name, srcArchiveName, deployArchiveName, buildcmd) + + // update function spec with resource version + function.Spec.Package.PackageRef = fission.PackageRef{ + Namespace: pkgMetadata.Namespace, + Name: pkgMetadata.Name, + ResourceVersion: pkgMetadata.ResourceVersion, + } + } + _, err = client.FunctionUpdate(function) checkErr(err, "update function") diff --git a/fission/main.go b/fission/main.go index a822c2dd..f64c45a6 100644 --- a/fission/main.go +++ b/fission/main.go @@ -40,17 +40,20 @@ func main() { fnNameFlag := cli.StringFlag{Name: "name", Usage: "function name"} fnEnvNameFlag := cli.StringFlag{Name: "env", Usage: "environment name for function"} fnCodeFlag := cli.StringFlag{Name: "code", Usage: "local path or URL for source code"} - fnPackageFlag := cli.StringFlag{Name: "package", Usage: "local path or URL for binary package"} - fnSrcPackageFlag := cli.StringFlag{Name: "srcpkg", Usage: "local path or URL for source package"} + fnPackageFlag := cli.StringFlag{Name: "package", Usage: "(Deprecated) local path or URL for binary package"} + fnDeployArchiveFlag := cli.StringFlag{Name: "deployarchive, deploy", Usage: "local path or URL for deployment archive"} + fnSrcArchiveFlag := cli.StringFlag{Name: "sourcearchive, src", Usage: "local path or URL for source archive"} fnPodFlag := cli.StringFlag{Name: "pod", Usage: "function pod name, optional (use latest if unspecified)"} fnFollowFlag := cli.BoolFlag{Name: "follow, f", Usage: "specify if the logs should be streamed"} fnDetailFlag := cli.BoolFlag{Name: "detail, d", Usage: "display detailed information"} fnLogDBTypeFlag := cli.StringFlag{Name: "dbtype", Usage: "log database type, e.g. influxdb (currently only influxdb is supported)"} + fnEntryPointFlag := cli.StringFlag{Name: "entrypoint", Usage: "entry point for environment v2 to load with"} + fnBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "build command for builder to run with"} fnSubcommands := []cli.Command{ - {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcPackageFlag, htUrlFlag, htMethodFlag}, Action: fnCreate}, + {Name: "create", Usage: "Create new function (and optionally, an HTTP route to it)", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag, htUrlFlag, htMethodFlag}, Action: fnCreate}, {Name: "get", Usage: "Get function source code", Flags: []cli.Flag{fnNameFlag}, Action: fnGet}, {Name: "getmeta", Usage: "Get function metadata", Flags: []cli.Flag{fnNameFlag}, Action: fnGetMeta}, - {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcPackageFlag}, Action: fnUpdate}, + {Name: "update", Usage: "Update function source code", Flags: []cli.Flag{fnNameFlag, fnEnvNameFlag, fnCodeFlag, fnPackageFlag, fnSrcArchiveFlag, fnDeployArchiveFlag, fnEntryPointFlag, fnBuildCmdFlag}, Action: fnUpdate}, {Name: "delete", Usage: "Delete function", Flags: []cli.Flag{fnNameFlag}, Action: fnDelete}, {Name: "list", Usage: "List all functions", Flags: []cli.Flag{}, Action: fnList}, {Name: "logs", Usage: "Display function logs", Flags: []cli.Flag{fnNameFlag, fnPodFlag, fnFollowFlag, fnDetailFlag, fnLogDBTypeFlag}, Action: fnLogs}, @@ -100,8 +103,9 @@ func main() { envImageFlag := cli.StringFlag{Name: "image", Usage: "Environment image URL"} envBuilderImageFlag := cli.StringFlag{Name: "builder", Usage: "Environment builder image URL (optional)"} envBuildCmdFlag := cli.StringFlag{Name: "buildcmd", Usage: "Build command for environment builder to build source package (optional)"} + envVersionFlag := cli.IntFlag{Name: "version", Usage: "Environment API version: defaults to 1 (means v1 interface)"} envSubcommands := []cli.Command{ - {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag}, Action: envCreate}, + {Name: "create", Aliases: []string{"add"}, Usage: "Add an environment", Flags: []cli.Flag{envNameFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag, envVersionFlag}, Action: envCreate}, {Name: "get", Usage: "Get environment details", Flags: []cli.Flag{envNameFlag}, Action: envGet}, {Name: "update", Usage: "Update environment", Flags: []cli.Flag{envNameFlag, envImageFlag, envBuilderImageFlag, envBuildCmdFlag}, Action: envUpdate}, {Name: "delete", Usage: "Delete environment", Flags: []cli.Flag{envNameFlag}, Action: envDelete}, diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 888d2b9a..0f78ce2f 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -27,6 +27,7 @@ import ( "net/http" "net/url" "os" + "path/filepath" "strings" "time" @@ -61,10 +62,13 @@ type ( poolInstanceId string // small random string to uniquify pod names fetcherImage string fetcherImagePullPolicy v1.PullPolicy + runtimeImagePullPolicy v1.PullPolicy // pull policy for generic pool to created env deployment kubernetesClient *kubernetes.Clientset + fissionClient *tpr.FissionClient instanceId string // poolmgr instance id labelsForPool map[string]string requestChannel chan *choosePodRequest + sharedMountPath string } // serialize the choosing of pods so that choices don't conflict @@ -78,7 +82,19 @@ type ( } ) +func getImagePullPolicy(policy string) v1.PullPolicy { + switch policy { + case "Always": + return v1.PullAlways + case "Never": + return v1.PullNever + default: + return v1.PullIfNotPresent + } +} + func MakeGenericPool( + fissionClient *tpr.FissionClient, kubernetesClient *kubernetes.Clientset, env *tpr.Environment, initialReplicas int32, @@ -92,9 +108,13 @@ func MakeGenericPool( if len(fetcherImage) == 0 { fetcherImage = "fission/fetcher" } - fetcherImagePullPolicyS := os.Getenv("FETCHER_IMAGE_PULL_POLICY") - if len(fetcherImagePullPolicyS) == 0 { - fetcherImagePullPolicyS = "IfNotPresent" + 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 @@ -103,6 +123,7 @@ func MakeGenericPool( 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? @@ -111,18 +132,13 @@ func MakeGenericPool( poolInstanceId: uniuri.NewLen(8), instanceId: instanceId, fetcherImage: fetcherImage, - useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start + useSvc: false, // defaults off -- svc takes a second or more to become routable, slowing cold start + sharedMountPath: "/userfunc", // used by generic pool when creating env deployment to specify the share volume path for fetcher & env } - switch fetcherImagePullPolicyS { - case "Always": - gp.fetcherImagePullPolicy = v1.PullAlways - case "Never": - gp.fetcherImagePullPolicy = v1.PullNever - default: - gp.fetcherImagePullPolicy = v1.PullIfNotPresent - } + 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. @@ -276,12 +292,15 @@ func (gp *GenericPool) getFetcherUrl(podIP string) string { return fmt.Sprintf("http://%v:8000/", podIP) } -func (gp *GenericPool) getSpecializeUrl(podIP string) string { +func (gp *GenericPool) getSpecializeUrl(podIP string, version int) string { u := os.Getenv("TEST_SPECIALIZE_URL") if len(u) != 0 { return u } - return fmt.Sprintf("http://%v:8888/specialize", podIP) + 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 @@ -297,10 +316,23 @@ func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *api.ObjectMeta) erro // tell fetcher to get the function. fetcherUrl := gp.getFetcherUrl(podIP) log.Printf("[%v] calling fetcher to copy function", metadata.Name) - err := fetcherClient.DoFetchRequest(fetcherUrl, &fetcher.FetchRequest{ + + fn, err := gp.fissionClient. + Functions(metadata.Namespace). + Get(metadata.Name) + if err != nil { + return err + } + + targetFilename := "user" + + err = fetcherClient.MakeClient(fetcherUrl).Fetch(&fetcher.FetchRequest{ FetchType: fetcher.FETCH_DEPLOYMENT, - Function: *metadata, - Filename: "user", // XXX use function id instead + Package: api.ObjectMeta{ + Namespace: fn.Spec.Package.PackageRef.Namespace, + Name: fn.Spec.Package.PackageRef.Name, + }, + Filename: targetFilename, // XXX use function id instead }) if err != nil { return err @@ -311,12 +343,29 @@ func (gp *GenericPool) specializePod(pod *v1.Pod, metadata *api.ObjectMeta) erro // get function run container to specialize log.Printf("[%v] specializing pod", metadata.Name) - specializeUrl := gp.getSpecializeUrl(podIP) // 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, + } + + body, err := json.Marshal(loadReq) + if err != nil { + return err + } + for i := 0; i < maxRetries; i++ { - resp2, err := http.Post(specializeUrl, "text/plain", bytes.NewReader([]byte{})) + 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() @@ -352,7 +401,6 @@ func (gp *GenericPool) createPool() error { poolDeploymentName := fmt.Sprintf("%v-%v-%v", gp.env.Metadata.Name, gp.env.Metadata.UID, strings.ToLower(gp.poolInstanceId)) - sharedMountPath := "/userfunc" deployment := &v1beta1.Deployment{ ObjectMeta: v1.ObjectMeta{ Name: poolDeploymentName, @@ -380,12 +428,12 @@ func (gp *GenericPool) createPool() error { { Name: gp.env.Metadata.Name, Image: gp.env.Spec.Runtime.Image, - ImagePullPolicy: v1.PullIfNotPresent, + ImagePullPolicy: gp.runtimeImagePullPolicy, TerminationMessagePath: "/dev/termination-log", VolumeMounts: []v1.VolumeMount{ { Name: "userfunc", - MountPath: sharedMountPath, + MountPath: gp.sharedMountPath, }, }, }, @@ -397,10 +445,10 @@ func (gp *GenericPool) createPool() error { VolumeMounts: []v1.VolumeMount{ { Name: "userfunc", - MountPath: sharedMountPath, + MountPath: gp.sharedMountPath, }, }, - Command: []string{"/fetcher", sharedMountPath}, + Command: []string{"/fetcher", gp.sharedMountPath}, }, }, ServiceAccountName: "fission-fetcher", diff --git a/poolmgr/gpm.go b/poolmgr/gpm.go index 793590ab..0be023f4 100644 --- a/poolmgr/gpm.go +++ b/poolmgr/gpm.go @@ -87,7 +87,7 @@ func (gpm *GenericPoolManager) service() { pool, ok := gpm.pools[tpr.CacheKey(&req.env.Metadata)] if !ok { pool, err = MakeGenericPool( - gpm.kubernetesClient, req.env, + gpm.fissionClient, gpm.kubernetesClient, req.env, 3, // TODO configurable/autoscalable gpm.namespace, gpm.fsCache, gpm.instanceId) if err != nil { diff --git a/test/build_and_test.sh b/test/build_and_test.sh index 39a99a07..b118f11e 100755 --- a/test/build_and_test.sh +++ b/test/build_and_test.sh @@ -12,12 +12,18 @@ source $(dirname $0)/test_utils.sh IMAGE=gcr.io/fission-ci/fission-bundle FETCHER_IMAGE=gcr.io/fission-ci/fetcher +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env +PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python-env-builder TAG=test build_and_push_fission_bundle $IMAGE:$TAG build_and_push_fetcher $FETCHER_IMAGE:$TAG +build_and_push_python_env_runtime $PYTHON_RUNTIME_IMAGE:$TAG + +build_and_push_python_env_builder $PYTHON_BUILDER_IMAGE:$TAG + build_fission_cli install_and_test $IMAGE $TAG $FETCHER_IMAGE $TAG diff --git a/test/test_utils.sh b/test/test_utils.sh index f3ac6044..d953b92e 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -50,6 +50,36 @@ build_and_push_fetcher() { popd } +build_and_push_python_env_runtime() { + image_tag=$1 + + pushd $ROOT/environments/python3/ + docker build -t $image_tag . + + gcloud_login + + gcloud docker -- push $image_tag + popd +} + +build_and_push_python_env_builder() { + image_tag=$1 + + pushd $ROOT/builder/cmd + ./build.sh + popd + pushd $ROOT/environments/python3/builder + builderDir=${GOPATH}/src/github.com/fission/fission/builder/cmd + cp ${builderDir}/builder . + + docker build -t $image_tag . + + gcloud_login + + gcloud docker -- push $image_tag + popd +} + build_fission_cli() { pushd $ROOT/fission diff --git a/test/tests/test_buildermgr.sh b/test/tests/test_buildermgr.sh new file mode 100755 index 00000000..ea6213ea --- /dev/null +++ b/test/tests/test_buildermgr.sh @@ -0,0 +1,73 @@ +#!/bin/bash + +set -euo pipefail + +# Create a function with source package in python +# to test builder manger functionality. +# There are two ways to trigger the build +# 1. manually trigger by http post +# 2. package watcher triggers the build if any changes to packages + +ROOT=$(dirname $0)/../.. +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env:test +PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python-env-builder:test + +fn=python-srcbuild-$(date +%s) + +checkFunctionResponse() { + echo "Doing an HTTP GET on the function's route" + response=$(curl http://$FISSION_ROUTER/$1) + + echo "Checking for valid response" + echo $response + echo $response | grep -i "a: 1 b: {c: 3, d: 4}" +} + +echo "Pre-test cleanup" +fission env delete --name python || true + +echo "Creating python env" +fission env create --name python --image $PYTHON_RUNTIME_IMAGE --builder $PYTHON_BUILDER_IMAGE +trap "fission env delete --name python" EXIT + +echo "Waiting for env builder to catch up" +sleep 30 + +echo "Creating source pacakage" +zip -jr demo-src-pkg.zip $ROOT/examples/python/sourcepkg/ + +echo "Creating function " $fn +fission fn create --name $fn --env python --src demo-src-pkg.zip --entrypoint "main" --buildcmd "./build.sh" +trap "fission fn delete --name $fn" EXIT + +echo "Creating route" +fission route create --function $fn --url /$fn --method GET + +echo "Waiting for router to catch up" +sleep 3 + +echo "Doing an HTTP POST on the builder manager's route to start a build" +pkg=$(kubectl --namespace default get functions $fn -o jsonpath='{.spec.package.packageref.name}') +echo $pkg +response=$(curl -X POST $FISSION_URL/proxy/buildermgr/v1/build \ + -H 'content-type: application/json' \ + -d "{\"package\": {\"namespace\": \"default\",\"name\": \"$pkg\"}}") + +echo "Waiting for builder manager to finish the build triggered by http request" +sleep 30 + +checkFunctionResponse $fn + +echo "Updating function " $fn +fission fn update --name $fn --src demo-src-pkg.zip +trap "fission fn delete --name $fn" EXIT + +echo "Waiting for builder manager to finish the build triggered by packageWatcher" +sleep 30 + +checkFunctionResponse $fn + +# crappy cleanup, improve this later +kubectl get httptrigger -o name | tail -1 | cut -f2 -d'/' | xargs kubectl delete httptrigger + +echo "All done." diff --git a/tpr/types.go b/tpr/types.go index 0941d419..77c073cf 100644 --- a/tpr/types.go +++ b/tpr/types.go @@ -43,6 +43,8 @@ type ( unversioned.TypeMeta `json:",inline"` Metadata api.ObjectMeta `json:"metadata"` Spec fission.PackageSpec `json:"spec"` + + Status fission.PackageStatus `json:"status"` } PackageList struct { unversioned.TypeMeta `json:",inline"` diff --git a/types.go b/types.go index e5966488..24ffef97 100644 --- a/types.go +++ b/types.go @@ -65,9 +65,10 @@ type ( BuildStatus string PackageSpec struct { - Environment EnvironmentReference `json:"environment"` - Source Archive `json:"source"` - Deployment Archive `json:"deployment"` + Environment EnvironmentReference `json:"environment"` + Source Archive `json:"source"` + Deployment Archive `json:"deployment"` + BuildCommand string `json:"buildcmd"` // In the future, we can have a debug build here too } PackageStatus struct { @@ -220,11 +221,11 @@ type ( // env-specific. Optional. FilePath string `json:"filepath"` - // Entrypoint has an environment-specific meaning; + // FunctionName has an environment-specific meaning; // usually, it defines a function within a module // containing multiple functions. Optional; default is // environment-specific. - EntryPoint string `json:"entrypoint"` + FunctionName string `json:"functionName"` // URL to expose this function at. Optional; defaults // to "/".