diff --git a/buildermgr/envwatcher.go b/buildermgr/envwatcher.go index 4172a13a..eaa9c129 100644 --- a/buildermgr/envwatcher.go +++ b/buildermgr/envwatcher.go @@ -269,22 +269,32 @@ func (envw *environmentWatcher) createBuilder(env *crd.Environment) (*builderInf if err != nil { return nil, err } + // there should be only one service in svcList if len(svcList) == 0 { svc, err = envw.createBuilderService(env) if err != nil { return nil, fmt.Errorf("Error creating builder service: %v", err) } + } else if len(svcList) == 1 { + svc = &svcList[0] + } else { + return nil, fmt.Errorf("Found more than one builder service for environment %v", env.Metadata.Name) } deployList, err := envw.getBuilderDeploymentList(sel) if err != nil { return nil, err } + // there should be only one deploy in deployList if len(deployList) == 0 { deploy, err = envw.createBuilderDeployment(env) if err != nil { return nil, fmt.Errorf("Error creating builder deployment: %v", err) } + } else if len(deployList) == 1 { + deploy = &deployList[0] + } else { + return nil, fmt.Errorf("Found more than one builder deployment for environment %v", env.Metadata.Name) } return &builderInfo{ diff --git a/controller/api.go b/controller/api.go index 1b1afc37..41f500f9 100644 --- a/controller/api.go +++ b/controller/api.go @@ -181,7 +181,6 @@ 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) r.HandleFunc("/proxy/logs/{function}", api.FunctionPodLogs).Methods("POST") r.HandleFunc("/proxy/workflows-apiserver/{path:.*}", api.WorkflowApiserverProxy) diff --git a/controller/buildermgr.go b/controller/buildermgr.go index 4e955267..bd74428f 100644 --- a/controller/buildermgr.go +++ b/controller/buildermgr.go @@ -26,7 +26,7 @@ import ( func (api *API) BuilderManagerBuildProxy(w http.ResponseWriter, r *http.Request) { u := api.builderManagerUrl + "/v1/build" - proxy, err := api._getBuilderManagerProxy(u) + proxy, err := api.getBuilderManagerProxy(u) if err != nil { msg := fmt.Sprintf("Failed to establish proxy server: %v", err) log.Println(msg) @@ -36,19 +36,7 @@ func (api *API) BuilderManagerBuildProxy(w http.ResponseWriter, r *http.Request) 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) { +func (api *API) getBuilderManagerProxy(targetUrl string) (*httputil.ReverseProxy, error) { svcUrl, err := url.Parse(targetUrl) if err != nil { return nil, err diff --git a/environments/fetcher/client/client.go b/environments/fetcher/client/client.go index 1713ec79..c0dfa213 100644 --- a/environments/fetcher/client/client.go +++ b/environments/fetcher/client/client.go @@ -4,7 +4,6 @@ import ( "bytes" "encoding/json" "io/ioutil" - "log" "net/http" //"time" @@ -68,13 +67,11 @@ func (c *Client) Upload(fr *fetcher.UploadRequest) (*fetcher.UploadResponse, err return nil, err } - log.Printf("Received upload response: %v", string(rBody)) - - uploadReq := fetcher.UploadResponse{} - err = json.Unmarshal([]byte(rBody), &uploadReq) + uploadResp := fetcher.UploadResponse{} + err = json.Unmarshal([]byte(rBody), &uploadResp) if err != nil { return nil, err } - return &uploadReq, nil + return &uploadResp, nil } diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index 3d6f6260..80fecd01 100644 --- a/environments/fetcher/fetcher.go +++ b/environments/fetcher/fetcher.go @@ -152,13 +152,11 @@ func (fetcher *Fetcher) FetchHandler(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), 400) return } - log.Printf("fetcher received fetch request: %v", req) + log.Printf("fetcher received fetch request and started downloading: %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) @@ -283,7 +281,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) { return } - log.Println("Start uploading...") + log.Println("Starting upload...") ssClient := storageSvcClient.MakeClient(req.StorageSvcUrl) fileID, err := ssClient.Upload(dstFilepath, nil) diff --git a/environments/python3/server.py b/environments/python3/server.py index a20b19ae..d025277b 100644 --- a/environments/python3/server.py +++ b/environments/python3/server.py @@ -24,12 +24,41 @@ 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) + handler = body['functionName'] + + # The value of "functionName" is consist of `.`. + moduleName, funcName = handler.split(".") + + # check whether the destination is a directory or a file + if os.path.isdir(filepath): + # add package directory path into module search path + sys.path.append(filepath) + + # find module from package path we append previously. + # Python will try to find module from the same name file under + # the package directory. If search is successful, the return + # value is a 3-element tuple; otherwise, an exception "ImportError" + # is raised. + # Second parameter of find_module enforces python to find same + # name module from the given list of directories to prevent name + # confliction with built-in modules. + f, path, desc = imp.find_module(moduleName, [filepath]) + + # load module + # Return module object is the load is successful; otherwise, + # an exception is raised. + try: + mod = imp.load_module(moduleName, f, path, desc) + finally: + if f: + f.close() + else: + # load source from destination python file + mod = imp.load_source(moduleName, filepath) + + # load user function from module + userfunc = getattr(mod, funcName) + return "" @app.route('/', methods=['GET', 'POST', 'PUT', 'HEAD', 'OPTIONS', 'DELETE']) diff --git a/fission/environment.go b/fission/environment.go index 4f796449..4495c305 100644 --- a/fission/environment.go +++ b/fission/environment.go @@ -55,7 +55,6 @@ func envCreate(c *cli.Context) error { // 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 } @@ -132,7 +131,7 @@ func envUpdate(c *cli.Context) error { } if env.Spec.Version == 1 && (len(envBuilderImg) > 0 || len(envBuildCmd) > 0) { - fatal("Environment v1 API interface doesn't supported environment builder.") + fatal("Version 1 Environments do not support builders. Must specify --version=2.") } if len(envBuilderImg) > 0 { diff --git a/fission/function.go b/fission/function.go index 208bdbbd..7ed69544 100644 --- a/fission/function.go +++ b/fission/function.go @@ -20,6 +20,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "fmt" "io" @@ -31,6 +32,7 @@ import ( "text/tabwriter" "time" + "github.com/dchest/uniuri" "github.com/satori/go.uuid" "github.com/urfave/cli" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -87,36 +89,48 @@ func createArchive(client *client.Client, fileName string) *fission.Archive { return &archive } -func createPackage(client *client.Client, envName, srcArchiveName, deployArchiveName, buildcmd string) *metav1.ObjectMeta { +func createPackage(client *client.Client, fnName, envName, srcArchiveName, deployArchiveName, buildcmd string) *metav1.ObjectMeta { pkgSpec := fission.PackageSpec{ Environment: fission.EnvironmentReference{ Namespace: metav1.NamespaceDefault, Name: envName, }, } - var pkgStatus fission.BuildStatus = fission.BuildStatusSucceeded + + var pkgStatus fission.BuildStatus 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 + } + + // start a build only when a package has no deploy archive + if len(srcArchiveName) > 0 && len(deployArchiveName) == 0 { pkgStatus = fission.BuildStatusPending + } else { + pkgStatus = fission.BuildStatusNone } if len(buildcmd) > 0 { pkgSpec.BuildCommand = buildcmd } - pkgName := strings.ToLower(uuid.NewV4().String()) + fnList, err := json.Marshal([]string{fnName}) + checkErr(err, "encode json") + + annotation := map[string]string{ + "createdForFunction": fnName, + "usedByFunctions": string(fnList), + } + + pkgName := strings.ToLower(fmt.Sprintf("%v-%v", fnName, uniuri.NewLen(6))) pkg := &crd.Package{ Metadata: metav1.ObjectMeta{ - Name: pkgName, - Namespace: metav1.NamespaceDefault, + Name: pkgName, + Namespace: metav1.NamespaceDefault, + Annotations: annotation, }, Spec: pkgSpec, Status: fission.PackageStatus{ @@ -194,11 +208,8 @@ func fnCreate(c *cli.Context) error { entrypoint := c.String("entrypoint") buildcmd := c.String("buildcmd") - if len(buildcmd) == 0 { - buildcmd = "/builder" - } - pkgMetadata := createPackage(client, envName, srcArchiveName, deployArchiveName, buildcmd) + pkgMetadata := createPackage(client, fnName, envName, srcArchiveName, deployArchiveName, buildcmd) function := &crd.Function{ Metadata: metav1.ObjectMeta{ @@ -357,7 +368,7 @@ func fnUpdate(c *cli.Context) error { if len(deployArchiveName) > 0 || len(srcArchiveName) > 0 { // create a new package for function - pkgMetadata := createPackage(client, + pkgMetadata := createPackage(client, function.Metadata.Name, function.Spec.Environment.Name, srcArchiveName, deployArchiveName, buildcmd) // update function spec with resource version diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 61926bd4..47ec8e8b 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -68,7 +68,7 @@ type ( instanceId string // poolmgr instance id labelsForPool map[string]string requestChannel chan *choosePodRequest - sharedMountPath string + sharedMountPath string // used by generic pool when creating env deployment to specify the share volume path for fetcher & env } // serialize the choosing of pods so that choices don't conflict @@ -133,7 +133,7 @@ func MakeGenericPool( instanceId: instanceId, fetcherImage: fetcherImage, 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 + sharedMountPath: "/userfunc", // change this may break v1 compatibility, since most of the v1 environments have hard-coded "/userfunc" in loading path } gp.runtimeImagePullPolicy = getImagePullPolicy(runtimeImagePullPolicy) @@ -328,7 +328,13 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta return err } + // for backward compatibility, since most v1 env + // still try to load user function from hard coded + // path /userfunc/user targetFilename := "user" + if gp.env.Spec.Version == 2 { + targetFilename = string(fn.Metadata.UID) + } err = fetcherClient.MakeClient(fetcherUrl).Fetch(&fetcher.FetchRequest{ FetchType: fetcher.FETCH_DEPLOYMENT, @@ -336,7 +342,7 @@ func (gp *GenericPool) specializePod(pod *apiv1.Pod, metadata *metav1.ObjectMeta Namespace: fn.Spec.Package.PackageRef.Namespace, Name: fn.Spec.Package.PackageRef.Name, }, - Filename: targetFilename, // XXX use function id instead + Filename: targetFilename, }) if err != nil { return err @@ -433,7 +439,7 @@ func (gp *GenericPool) createPool() error { { Name: gp.env.Metadata.Name, Image: gp.env.Spec.Runtime.Image, - ImagePullPolicy: apiv1.PullIfNotPresent, + ImagePullPolicy: gp.runtimeImagePullPolicy, TerminationMessagePath: "/dev/termination-log", VolumeMounts: []apiv1.VolumeMount{ { diff --git a/test/build_and_test.sh b/test/build_and_test.sh index b118f11e..ff55f5cc 100755 --- a/test/build_and_test.sh +++ b/test/build_and_test.sh @@ -10,19 +10,22 @@ fi 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 +REPO=gcr.io/fission-ci +IMAGE=$REPO/fission-bundle +FETCHER_IMAGE=$REPO/fetcher 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_builder -build_and_push_python_env_builder $PYTHON_BUILDER_IMAGE:$TAG +ENV='python3' + +build_and_push_env_runtime $ENV $REPO/$ENV-env:$TAG + +build_and_push_env_builder $ENV $REPO/$ENV-env-builder:$TAG build_fission_cli diff --git a/test/test_utils.sh b/test/test_utils.sh index fc964adf..f17f3be4 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -50,10 +50,17 @@ build_and_push_fetcher() { popd } -build_and_push_python_env_runtime() { - image_tag=$1 +build_builder() { + pushd $ROOT/builder/cmd + ./build.sh + popd +} - pushd $ROOT/environments/python3/ +build_and_push_env_runtime() { + env=$1 + image_tag=$2 + + pushd $ROOT/environments/$env/ docker build -t $image_tag . gcloud_login @@ -62,13 +69,11 @@ build_and_push_python_env_runtime() { popd } -build_and_push_python_env_builder() { - image_tag=$1 +build_and_push_env_builder() { + env=$1 + image_tag=$2 - pushd $ROOT/builder/cmd - ./build.sh - popd - pushd $ROOT/environments/python3/builder + pushd $ROOT/environments/$env/builder builderDir=${GOPATH}/src/github.com/fission/fission/builder/cmd cp ${builderDir}/builder . @@ -80,7 +85,6 @@ build_and_push_python_env_builder() { popd } - build_fission_cli() { pushd $ROOT/fission go build . @@ -240,6 +244,7 @@ dump_logs() { dump_fission_logs $ns $fns controller dump_fission_logs $ns $fns router dump_fission_logs $ns $fns poolmgr + dump_fission_logs $ns $fns buildermgr dump_function_pod_logs $ns $fns dump_fission_crds } diff --git a/test/tests/test_buildermgr.sh b/test/tests/test_buildermgr.sh index ea6213ea..b59b7a5a 100755 --- a/test/tests/test_buildermgr.sh +++ b/test/tests/test_buildermgr.sh @@ -9,8 +9,8 @@ set -euo pipefail # 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 +PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python3-env:test +PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python3-env-builder:test fn=python-srcbuild-$(date +%s) @@ -23,21 +23,47 @@ checkFunctionResponse() { echo $response | grep -i "a: 1 b: {c: 3, d: 4}" } +waitBuild() { + echo "Waiting for builder manager to finish the build" + + while true; do + kubectl --namespace default get packages $1 -o jsonpath='{.status.buildstatus}'|grep succeeded + if [[ $? -eq 0 ]]; then + break + fi + done +} +export -f waitBuild + +waitEnvBuilder() { + echo "Waiting for env builder to catch up" + + while true; do + kubectl --namespace fission-builder get pod|grep python|grep Running + if [[ $? -eq 0 ]]; then + break + fi + done + + sleep 10 +} +export -f waitEnvBuilder + echo "Pre-test cleanup" fission env delete --name python || true +kubectl --namespace default get packages|grep -v NAME|awk '{print $1}'|xargs -I@ bash -c 'kubectl --namespace default delete packages @' || 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 +timeout 180s bash -c waitEnvBuilder 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" +fission fn create --name $fn --env python --src demo-src-pkg.zip --entrypoint "user.main" --buildcmd "./build.sh" trap "fission fn delete --name $fn" EXIT echo "Creating route" @@ -46,15 +72,10 @@ 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 +# wait for build to finish at most 60s +timeout 60s bash -c "waitBuild $pkg" checkFunctionResponse $fn @@ -62,8 +83,10 @@ 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 +pkg=$(kubectl --namespace default get functions $fn -o jsonpath='{.spec.package.packageref.name}') + +# wait for build to finish at most 60s +timeout 60s bash -c "waitBuild $pkg" checkFunctionResponse $fn diff --git a/types.go b/types.go index af957529..aafbfc2d 100644 --- a/types.go +++ b/types.go @@ -264,6 +264,7 @@ const ( BuildStatusRunning = "running" BuildStatusSucceeded = "succeeded" BuildStatusFailed = "failed" + BuildStatusNone = "none" ) const (