Builder manager bugfixes (#367)

This contains a few different builder manager fixes:

* Update the zip file structure to avoid an extra subdirectory
* Annotate packages with what functions are using them
* Only trigger builds when there is no deployment archive
* Python environment loadpath bugfixes
* Other bugfixes

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{

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)
 

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) {
+func (api *API) getBuilderManagerProxy(targetUrl string) (*httputil.ReverseProxy, error) {
-	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

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))
+	uploadResp := fetcher.UploadResponse{}
-
+	err = json.Unmarshal([]byte(rBody), &uploadResp)
-	uploadReq := fetcher.UploadResponse{}
-	err = json.Unmarshal([]byte(rBody), &uploadReq)
 	if err != nil {
 		return nil, err
 	}
 
-	return &uploadReq, nil
+	return &uploadResp, nil
 }

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)

environments/python3/server.py

@@ -24,12 +24,41 @@ def loadv2():
     global userfunc
     body = request.get_json()
     filepath = body['filepath']
-    functionName = body['functionName']
+    handler = body['functionName']
-    # add filepath into syspath for module import
+
-    sys.path.append(filepath)
+    # The value of "functionName" is consist of `<module-name>.<function-name>`.
-    fn, path, desc = imp.find_module('user', [filepath])
+    moduleName, funcName = handler.split(".")
-    mod = imp.load_module('user', fn, path, desc)
+
-    userfunc = getattr(mod, functionName)
+    # 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'])

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 {

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,
+			Name:        pkgName,
-			Namespace: metav1.NamespaceDefault,
+			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

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{
 								{

test/build_and_test.sh

@@ -10,19 +10,22 @@ fi
 
 source $(dirname $0)/test_utils.sh
 
-IMAGE=gcr.io/fission-ci/fission-bundle
+REPO=gcr.io/fission-ci
-FETCHER_IMAGE=gcr.io/fission-ci/fetcher
+IMAGE=$REPO/fission-bundle
-PYTHON_RUNTIME_IMAGE=gcr.io/fission-ci/python-env
+FETCHER_IMAGE=$REPO/fetcher
-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_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
 

test/test_utils.sh

@@ -50,10 +50,17 @@ build_and_push_fetcher() {
     popd
 }
 
-build_and_push_python_env_runtime() {
+build_builder() {
-    image_tag=$1
+    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() {
+build_and_push_env_builder() {
-    image_tag=$1
+    env=$1
+    image_tag=$2
 
-    pushd $ROOT/builder/cmd
+    pushd $ROOT/environments/$env/builder
-    ./build.sh
-    popd
-    pushd $ROOT/environments/python3/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
 }

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_RUNTIME_IMAGE=gcr.io/fission-ci/python3-env:test
-PYTHON_BUILDER_IMAGE=gcr.io/fission-ci/python-env-builder: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"
+timeout 180s bash -c waitEnvBuilder
-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"
+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"
+# wait for build to finish at most 60s
-sleep 30
+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"
+pkg=$(kubectl --namespace default get functions $fn -o jsonpath='{.spec.package.packageref.name}')
-sleep 30
+
+# wait for build to finish at most 60s
+timeout 60s bash -c "waitBuild $pkg"
 
 checkFunctionResponse $fn
 

types.go

@@ -264,6 +264,7 @@ const (
 	BuildStatusRunning   = "running"
 	BuildStatusSucceeded = "succeeded"
 	BuildStatusFailed    = "failed"
+	BuildStatusNone      = "none"
 )
 
 const (