Add builder manager support (#308)

This change orchestrates function builds.

Environments (in v2) define a builder image, just like they do a runtime image. The builder image contains a build script that's invoked with source and deployment paths (as env vars).

The buildermgr watches for environments with build images defined, and creates build deployments and services.

Functions can define source and deployment. Buildermgr watches for functions with source code (and build status == pending) and invokes the environment's builder when appropriate. It captures logs from the build and sets the build status (success/failure) and build lots into the PackageStatus.

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",

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 {