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 "/".