Split out the Package type into a first class Kubernetes resource (#295)

Split out the Package type into a first class Kubernetes resource. Before this change, packages were implicitly tied to functions.

This wasn't ideal because:
 * Functions will need to share packages
 * A package storage system may be more generally useful than just
   functions (for example, for storing static assets)

This change does the following:
* Updates the fission and tpr types to add a new Package and PackageSpec.  It also creates a PackageRef type, and a FunctionPackageRef type. The PackageRef simply references a package, but the FunctionPackageRef includes the name of a function within the package. This allows us to share packages between different functions.
* Updates fetcher and other components for first-class packages
* Allows customization of fetcher image pull policy in the helm charts

charts/fission-all/templates/deployment.yaml

@@ -238,6 +238,8 @@ spec:
         env:
         - name: FETCHER_IMAGE
           value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
+        - name: FETCHER_IMAGE_PULL_POLICY
+          value: "{{ .Values.pullPolicy }}"
       serviceAccount: fission-svc
 
 ---

charts/fission-core/templates/deployment.yaml

@@ -238,6 +238,8 @@ spec:
         env:
         - name: FETCHER_IMAGE
           value: "{{ .Values.fetcherImage }}:{{ .Values.fetcherImageTag }}"
+        - name: FETCHER_IMAGE_PULL_POLICY
+          value: "{{ .Values.pullPolicy }}"
       serviceAccount: fission-svc
 
 ---

charts/fission-core/values.yaml

@@ -13,7 +13,7 @@ image: fission/fission-bundle
 imageTag: v0.2.0-20170822
 
 ## Image pull policy
-pullPolicy: ifNotPresent
+pullPolicy: IfNotPresent
 
 ## Fission fetcher repository
 fetcherImage: fission/fetcher

controller/api.go

@@ -106,6 +106,12 @@ func (api *API) Serve(port int) {
 	r.HandleFunc(`/v1/{rest:[a-zA-Z0-9=\-\/]+}`, api.ApiVersionMismatchHandler)
 	r.HandleFunc("/", api.HomeHandler)
 
+	r.HandleFunc("/v2/packages", api.PackageApiList).Methods("GET")
+	r.HandleFunc("/v2/packages", api.PackageApiCreate).Methods("POST")
+	r.HandleFunc("/v2/packages/{package}", api.PackageApiGet).Methods("GET")
+	r.HandleFunc("/v2/packages/{package}", api.PackageApiUpdate).Methods("PUT")
+	r.HandleFunc("/v2/packages/{package}", api.PackageApiDelete).Methods("DELETE")
+
 	r.HandleFunc("/v2/functions", api.FunctionApiList).Methods("GET")
 	r.HandleFunc("/v2/functions", api.FunctionApiCreate).Methods("POST")
 	r.HandleFunc("/v2/functions/{function}", api.FunctionApiGet).Methods("GET")

controller/api_test.go

@@ -27,7 +27,6 @@ import (
 
 	"k8s.io/client-go/1.5/pkg/api"
 
-	"bytes"
 	"github.com/fission/fission"
 	"github.com/fission/fission/controller/client"
 	"github.com/fission/fission/tpr"
@@ -83,8 +82,8 @@ func TestFunctionApi(t *testing.T) {
 		},
 		Spec: fission.FunctionSpec{
 			EnvironmentName: "nodejs",
-			Deployment: fission.Package{
+			Deployment: fission.FunctionPackageRef{
-				Literal: []byte("code1"),
+				FunctionName: "xxx",
 			},
 		},
 	}
@@ -94,22 +93,18 @@ func TestFunctionApi(t *testing.T) {
 	})
 	assertNotFoundFailure(err, "function")
 
-	m, err := g.client.FunctionCreate(testFunc)
+	_, err = g.client.FunctionCreate(testFunc)
 	panicIf(err)
 
 	_, err = g.client.FunctionCreate(testFunc)
 	assertNameReuseFailure(err, "function")
 
-	code, err := g.client.FunctionGetRawDeployment(m)
+	testFunc.Spec.Deployment.FunctionName = "yyy"
-	panicIf(err)
-	assert(bytes.Compare(code, testFunc.Spec.Deployment.Literal) == 0, "code from FunctionGetRawDeployment must match created function")
-
-	testFunc.Spec.Deployment.Literal = []byte("code2")
 	_, err = g.client.FunctionUpdate(testFunc)
 	panicIf(err)
 
 	testFunc.Metadata.Name = name2
-	m, err = g.client.FunctionCreate(testFunc)
+	_, err = g.client.FunctionCreate(testFunc)
 	panicIf(err)
 
 	funcs, err := g.client.FunctionList()
@@ -327,7 +322,7 @@ func TestMain(m *testing.M) {
 
 	go Start(8888)
 
-	time.Sleep(500 * time.Millisecond)
+	time.Sleep(time.Second)
 	g.client = client.MakeClient("http://localhost:8888")
 
 	resp, err := http.Get("http://localhost:8888/")

controller/client/function.go

@@ -24,20 +24,11 @@ import (
 
 	"k8s.io/client-go/1.5/pkg/api"
 
-	"github.com/fission/fission"
 	"github.com/fission/fission/tpr"
 )
 
 func (c *Client) FunctionCreate(f *tpr.Function) (*api.ObjectMeta, error) {
 
-	// ensure size limits
-	if len(f.Spec.Source.Literal) > 256*1024 {
-		return nil, fission.MakeError(fission.ErrorSizeLimitExceeded, "Source package literal larger than 256k")
-	}
-	if len(f.Spec.Deployment.Literal) > 256*1024 {
-		return nil, fission.MakeError(fission.ErrorSizeLimitExceeded, "Deployment package literal larger than 256k")
-	}
-
 	reqbody, err := json.Marshal(f)
 	if err != nil {
 		return nil, err

controller/client/package.go

@@ -0,0 +1,132 @@
+/*
+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 client
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"k8s.io/client-go/1.5/pkg/api"
+
+	"github.com/fission/fission/tpr"
+)
+
+func (c *Client) PackageCreate(f *tpr.Package) (*api.ObjectMeta, error) {
+
+	reqbody, err := json.Marshal(f)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.Post(c.url("packages"), "application/json", bytes.NewReader(reqbody))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleCreateResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m api.ObjectMeta
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+
+	return &m, nil
+}
+
+func (c *Client) PackageGet(m *api.ObjectMeta) (*tpr.Package, error) {
+	relativeUrl := fmt.Sprintf("packages/%v", m.Name)
+	relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace)
+
+	resp, err := http.Get(c.url(relativeUrl))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var f tpr.Package
+	err = json.Unmarshal(body, &f)
+	if err != nil {
+		return nil, err
+	}
+
+	return &f, nil
+}
+
+func (c *Client) PackageUpdate(f *tpr.Package) (*api.ObjectMeta, error) {
+	reqbody, err := json.Marshal(f)
+	if err != nil {
+		return nil, err
+	}
+	relativeUrl := fmt.Sprintf("packages/%v", f.Metadata.Name)
+
+	resp, err := c.put(relativeUrl, "application/json", reqbody)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m api.ObjectMeta
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+	return &m, nil
+}
+
+func (c *Client) PackageDelete(m *api.ObjectMeta) error {
+	relativeUrl := fmt.Sprintf("packages/%v", m.Name)
+	relativeUrl += fmt.Sprintf("?namespace=%v", m.Namespace)
+	return c.delete(relativeUrl)
+}
+
+func (c *Client) PackageList() ([]tpr.Package, error) {
+	resp, err := http.Get(c.url("packages"))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	funcs := make([]tpr.Package, 0)
+	err = json.Unmarshal(body, &funcs)
+	if err != nil {
+		return nil, err
+	}
+
+	return funcs, nil
+}

controller/functionApi.go

@@ -67,18 +67,6 @@ func (a *API) FunctionApiCreate(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Ensure size limits
-	if len(f.Spec.Source.Literal) > 256*1024 {
-		err := fission.MakeError(fission.ErrorInvalidArgument, "Source package literal larger than 256K")
-		a.respondWithError(w, err)
-		return
-	}
-	if len(f.Spec.Deployment.Literal) > 256*1024 {
-		err := fission.MakeError(fission.ErrorInvalidArgument, "Deployment package literal larger than 256K")
-		a.respondWithError(w, err)
-		return
-	}
-
 	fnew, err := a.fissionClient.Functions(f.Metadata.Namespace).Create(&f)
 	if err != nil {
 		a.respondWithError(w, err)
@@ -102,7 +90,6 @@ func (a *API) FunctionApiGet(w http.ResponseWriter, r *http.Request) {
 	if len(ns) == 0 {
 		ns = api.NamespaceDefault
 	}
-	raw := r.FormValue("deploymentraw") // just the deployment pkg
 
 	f, err := a.fissionClient.Functions(ns).Get(name)
 	if err != nil {
@@ -110,15 +97,10 @@ func (a *API) FunctionApiGet(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var resp []byte
+	resp, err := json.Marshal(f)
-	if raw != "" {
+	if err != nil {
-		resp = []byte(f.Spec.Deployment.Literal)
+		a.respondWithError(w, err)
-	} else {
+		return
-		resp, err = json.Marshal(f)
-		if err != nil {
-			a.respondWithError(w, err)
-			return
-		}
 	}
 	a.respondWithSuccess(w, resp)
 }

controller/packageApi.go

@@ -0,0 +1,170 @@
+/*
+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 controller
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"k8s.io/client-go/1.5/pkg/api"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/tpr"
+)
+
+func (a *API) PackageApiList(w http.ResponseWriter, r *http.Request) {
+	funcs, err := a.fissionClient.Packages(api.NamespaceAll).List(api.ListOptions{})
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(funcs.Items)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	a.respondWithSuccess(w, resp)
+}
+
+func (a *API) PackageApiCreate(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	var f tpr.Package
+	err = json.Unmarshal(body, &f)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	err = validateResourceName(f.Metadata.Name)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	// Ensure size limits
+	if len(f.Spec.Literal) > 256*1024 {
+		err := fission.MakeError(fission.ErrorInvalidArgument, "Package literal larger than 256K")
+		a.respondWithError(w, err)
+		return
+	}
+
+	fnew, err := a.fissionClient.Packages(f.Metadata.Namespace).Create(&f)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(fnew.Metadata)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	w.WriteHeader(http.StatusCreated)
+	a.respondWithSuccess(w, resp)
+}
+
+func (a *API) PackageApiGet(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	name := vars["package"]
+	ns := vars["namespace"]
+	if len(ns) == 0 {
+		ns = api.NamespaceDefault
+	}
+	raw := r.FormValue("raw") // just the deployment pkg
+
+	f, err := a.fissionClient.Packages(ns).Get(name)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	var resp []byte
+	if raw != "" {
+		resp = []byte(f.Spec.Literal)
+	} else {
+		resp, err = json.Marshal(f)
+		if err != nil {
+			a.respondWithError(w, err)
+			return
+		}
+	}
+	a.respondWithSuccess(w, resp)
+}
+
+func (a *API) PackageApiUpdate(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	name := vars["package"]
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	var f tpr.Package
+	err = json.Unmarshal(body, &f)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	if name != f.Metadata.Name {
+		err = fission.MakeError(fission.ErrorInvalidArgument, "Package name doesn't match URL")
+		a.respondWithError(w, err)
+		return
+	}
+
+	fnew, err := a.fissionClient.Packages(f.Metadata.Namespace).Update(&f)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(fnew.Metadata)
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+	a.respondWithSuccess(w, resp)
+}
+
+func (a *API) PackageApiDelete(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	name := vars["package"]
+	ns := vars["namespace"]
+	if len(ns) == 0 {
+		ns = api.NamespaceDefault
+	}
+
+	err := a.fissionClient.Packages(ns).Delete(name, &api.DeleteOptions{})
+	if err != nil {
+		a.respondWithError(w, err)
+		return
+	}
+
+	a.respondWithSuccess(w, []byte(""))
+}

environments/fetcher/fetcher.go

@@ -151,17 +151,23 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) {
 		}
 
 		// get pkg
-		var pkg *fission.Package
+		var pkg *tpr.Package
 		if req.FetchType == FETCH_SOURCE {
-			pkg = &fn.Spec.Source
+			pkg, err = fetcher.fissionClient.Packages(fn.Spec.Source.PackageRef.Namespace).Get(fn.Spec.Source.PackageRef.Name)
 		} else if req.FetchType == FETCH_DEPLOYMENT {
-			pkg = &fn.Spec.Deployment
+			pkg, err = fetcher.fissionClient.Packages(fn.Spec.Deployment.PackageRef.Namespace).Get(fn.Spec.Deployment.PackageRef.Name)
+		}
+		if err != nil {
+			e := fmt.Sprintf("Failed to get package: %v", err)
+			log.Printf(e)
+			http.Error(w, e, 500)
+			return
 		}
 
 		// get package data as literal or by url
-		if len(pkg.Literal) > 0 {
+		if len(pkg.Spec.Literal) > 0 {
 			// write pkg.Literal into tmpPath
-			err = ioutil.WriteFile(tmpPath, pkg.Literal, 0600)
+			err = ioutil.WriteFile(tmpPath, pkg.Spec.Literal, 0600)
 			if err != nil {
 				e := fmt.Sprintf("Failed to write file %v: %v", tmpPath, err)
 				log.Printf(e)
@@ -171,7 +177,7 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) {
 		} else {
 			// download and verify
 
-			err = downloadUrl(pkg.URL, tmpPath)
+			err = downloadUrl(pkg.Spec.URL, tmpPath)
 			if err != nil {
 				e := fmt.Sprintf("Failed to download url %v: %v", req.Url, err)
 				log.Printf(e)
@@ -179,7 +185,7 @@ func (fetcher *Fetcher) Handler(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			err = verifyChecksum(tmpPath, &pkg.Checksum)
+			err = verifyChecksum(tmpPath, &pkg.Spec.Checksum)
 			if err != nil {
 				e := fmt.Sprintf("Failed to verify checksum: %v", err)
 				log.Printf(e)

fission/function.go

@@ -20,11 +20,12 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
-	//"net/http"
 	"os"
+	"strings"
 	"text/tabwriter"
 	"time"
 
+	"github.com/dchest/uniuri"
 	"github.com/satori/go.uuid"
 	"github.com/urfave/cli"
 	"k8s.io/client-go/1.5/pkg/api"
@@ -34,27 +35,6 @@ import (
 	"github.com/fission/fission/tpr"
 )
 
-func downloadUrl(url string) {
-	// var resp *http.Response
-	// resp, err := http.Get(url)
-	// if err != nil {
-	// 	checkErr(err, fmt.Sprintf("download function"))
-	// }
-
-	// defer resp.Body.Close()
-	// if resp.StatusCode != http.StatusOK {
-	// 	err = fmt.Errorf("%v - HTTP response returned non 200 status", resp.StatusCode)
-	// 	checkErr(err, fmt.Sprintf("download function"))
-	// }
-
-	// contents, err := ioutil.ReadAll(resp.Body)
-	// if err != nil {
-	// 	checkErr(err, fmt.Sprintf("download function body %v", url))
-	// }
-
-	// return
-}
-
 func fileSize(filePath string) int64 {
 	info, err := os.Stat(filePath)
 	checkErr(err, fmt.Sprintf("stat %v", filePath))
@@ -98,6 +78,19 @@ func fnCreate(c *cli.Context) error {
 	}
 
 	pkgContents := getPackageContents(fileName)
+	pkgName := fmt.Sprintf("%v-%v", fnName, strings.ToLower(uniuri.NewLen(6)))
+	pkg := &tpr.Package{
+		Metadata: api.ObjectMeta{
+			Name:      pkgName,
+			Namespace: api.NamespaceDefault,
+		},
+		Spec: fission.PackageSpec{
+			Type:    fission.PackageTypeLiteral,
+			Literal: pkgContents,
+		},
+	}
+	_, err := client.PackageCreate(pkg)
+	checkErr(err, "upload package")
 
 	function := &tpr.Function{
 		Metadata: api.ObjectMeta{
@@ -106,13 +99,16 @@ func fnCreate(c *cli.Context) error {
 		},
 		Spec: fission.FunctionSpec{
 			EnvironmentName: envName,
-			Deployment: fission.Package{
+			Deployment: fission.FunctionPackageRef{
-				Literal: pkgContents,
+				PackageRef: fission.PackageRef{
+					Name:      pkgName,
+					Namespace: api.NamespaceDefault,
+				},
 			},
 		},
 	}
 
-	_, err := client.FunctionCreate(function)
+	_, err = client.FunctionCreate(function)
 	checkErr(err, "create function")
 
 	fmt.Printf("function '%v' created\n", fnName)
@@ -159,11 +155,16 @@ func fnGet(c *cli.Context) error {
 		Name:      fnName,
 		Namespace: api.NamespaceDefault,
 	}
-
+	fn, err := client.FunctionGet(m)
-	code, err := client.FunctionGetRawDeployment(m)
 	checkErr(err, "get function")
 
-	os.Stdout.Write(code)
+	pkg, err := client.PackageGet(&api.ObjectMeta{
+		Name:      fn.Spec.Deployment.PackageRef.Name,
+		Namespace: fn.Spec.Deployment.PackageRef.Namespace,
+	})
+	checkErr(err, "get package")
+
+	os.Stdout.Write(pkg.Spec.Literal)
 	return err
 }
 
@@ -205,6 +206,13 @@ func fnUpdate(c *cli.Context) error {
 	})
 	checkErr(err, fmt.Sprintf("read function '%v'", fnName))
 
+	pkgName := function.Spec.Deployment.PackageRef.Name
+	pkg, err := client.PackageGet(&api.ObjectMeta{
+		Name:      pkgName,
+		Namespace: api.NamespaceDefault,
+	})
+	checkErr(err, fmt.Sprintf("read package '%v'", pkgName))
+
 	envName := c.String("env")
 	fileName := c.String("code")
 	if len(fileName) == 0 {
@@ -222,14 +230,18 @@ func fnUpdate(c *cli.Context) error {
 			os.Exit(1)
 		}
 
-		function.Spec.Deployment.Literal = getPackageContents(fileName)
+		pkg.Spec.Literal = getPackageContents(fileName)
-	}
+
-	if len(envName) > 0 {
+		_, err = client.PackageUpdate(pkg)
-		function.Spec.EnvironmentName = envName
+		checkErr(err, "update package")
 	}
 
-	_, err = client.FunctionUpdate(function)
+	if len(envName) > 0 {
-	checkErr(err, "update function")
+		function.Spec.EnvironmentName = envName
+
+		_, err = client.FunctionUpdate(function)
+		checkErr(err, "update function")
+	}
 
 	fmt.Printf("function '%v' updated\n", fnName)
 	return err

poolmgr/gp.go

@@ -50,20 +50,21 @@ const POD_PHASE_RUNNING string = "Running"
 
 type (
 	GenericPool struct {
-		env              *tpr.Environment
+		env                    *tpr.Environment
-		replicas         int32                 // num idle pods
+		replicas               int32                 // num idle pods
-		deployment       *v1beta1.Deployment   // kubernetes deployment
+		deployment             *v1beta1.Deployment   // kubernetes deployment
-		namespace        string                // namespace to keep our resources
+		namespace              string                // namespace to keep our resources
-		podReadyTimeout  time.Duration         // timeout for generic pods to become ready
+		podReadyTimeout        time.Duration         // timeout for generic pods to become ready
-		idlePodReapTime  time.Duration         // pods unused for idlePodReapTime are deleted
+		idlePodReapTime        time.Duration         // pods unused for idlePodReapTime are deleted
-		fsCache          *functionServiceCache // cache funcSvc's by function, address and podname
+		fsCache                *functionServiceCache // cache funcSvc's by function, address and podname
-		useSvc           bool                  // create k8s service for specialized pods
+		useSvc                 bool                  // create k8s service for specialized pods
-		poolInstanceId   string                // small random string to uniquify pod names
+		poolInstanceId         string                // small random string to uniquify pod names
-		fetcherImage     string
+		fetcherImage           string
-		kubernetesClient *kubernetes.Clientset
+		fetcherImagePullPolicy v1.PullPolicy
-		instanceId       string // poolmgr instance id
+		kubernetesClient       *kubernetes.Clientset
-		labelsForPool    map[string]string
+		instanceId             string // poolmgr instance id
-		requestChannel   chan *choosePodRequest
+		labelsForPool          map[string]string
+		requestChannel         chan *choosePodRequest
 	}
 
 	// serialize the choosing of pods so that choices don't conflict
@@ -91,6 +92,10 @@ func MakeGenericPool(
 	if len(fetcherImage) == 0 {
 		fetcherImage = "fission/fetcher"
 	}
+	fetcherImagePullPolicyS := os.Getenv("FETCHER_IMAGE_PULL_POLICY")
+	if len(fetcherImagePullPolicyS) == 0 {
+		fetcherImagePullPolicyS = "IfNotPresent"
+	}
 
 	// TODO: in general we need to provide the user a way to configure pools.  Initial
 	// replicas, autoscaling params, various timeouts, etc.
@@ -109,6 +114,17 @@ func MakeGenericPool(
 		useSvc:           false, // defaults off -- svc takes a second or more to become routable, slowing cold start
 	}
 
+	switch fetcherImagePullPolicyS {
+	case "Always":
+		gp.fetcherImagePullPolicy = v1.PullAlways
+	case "Never":
+		gp.fetcherImagePullPolicy = v1.PullNever
+	default:
+		gp.fetcherImagePullPolicy = v1.PullIfNotPresent
+	}
+
+	log.Printf("fetcher image: %v, pull policy: %v", gp.fetcherImage, gp.fetcherImagePullPolicy)
+
 	// Labels for generic deployment/RS/pods.
 	gp.labelsForPool = map[string]string{
 		"environmentName":        gp.env.Metadata.Name,
@@ -376,7 +392,7 @@ func (gp *GenericPool) createPool() error {
 						{
 							Name:                   "fetcher",
 							Image:                  gp.fetcherImage,
-							ImagePullPolicy:        v1.PullIfNotPresent,
+							ImagePullPolicy:        gp.fetcherImagePullPolicy,
 							TerminationMessagePath: "/dev/termination-log",
 							VolumeMounts: []v1.VolumeMount{
 								{

poolmgr/poolmgr_test.go

@@ -63,6 +63,7 @@ func createTestNamespace(kubeClient *kubernetes.Clientset, ns string) {
 	if err != nil {
 		log.Panicf("failed to create ns %v: %v", ns, err)
 	}
+	log.Printf("Created namespace %v", ns)
 }
 
 // create a nodeport service
@@ -172,6 +173,22 @@ func TestPoolmgr(t *testing.T) {
 	// waitForPool(functionNs, "nodejs")
 	time.Sleep(6 * time.Second)
 
+	// create a package
+	p := &tpr.Package{
+		Metadata: api.ObjectMeta{
+			Name:      "hello",
+			Namespace: fissionNs,
+		},
+		Spec: fission.PackageSpec{
+			Type:    fission.PackageTypeLiteral,
+			Literal: []byte(`module.exports = async function(context) { return { status: 200, body: "Hello, world!\n" }; }`),
+		},
+	}
+	_, err = fissionClient.Packages(fissionNs).Create(p)
+	if err != nil {
+		log.Panicf("failed to create package: %v", err)
+	}
+
 	// create a function
 	f := &tpr.Function{
 		Metadata: api.ObjectMeta{
@@ -179,10 +196,12 @@ func TestPoolmgr(t *testing.T) {
 			Namespace: fissionNs,
 		},
 		Spec: fission.FunctionSpec{
-			Source: fission.Package{},
+			Source: fission.FunctionPackageRef{},
-			Deployment: fission.Package{
+			Deployment: fission.FunctionPackageRef{
-				Type:    fission.PackageTypeLiteral,
+				PackageRef: fission.PackageRef{
-				Literal: []byte(`module.exports = async function(context) { return { status: 200, body: "Hello, world!\n" }; }`),
+					Name:      p.Metadata.Name,
+					Namespace: p.Metadata.Namespace,
+				},
 			},
 			EnvironmentName: env.Metadata.Name,
 		},

test/test_utils.sh

@@ -173,17 +173,24 @@ dump_fission_resources() {
     dump_fission_resource environment.fission.io    
 }
 
+dump_env_pods() {
+    fns=$1
+
+    echo --- All environment pods ---
+    kubectl -n $fns get pod -o yaml
+    echo --- End environment pods ---
+}
+
 dump_logs() {
     id=$1
 
     ns=f-$id
     fns=f-func-$id
 
+    dump_env_pods $fns
     dump_fission_logs $ns $fns router
     dump_fission_logs $ns $fns poolmgr
-
     dump_function_pod_logs $ns $fns
-
     dump_fission_resources
 }
 

tpr/client.go

@@ -138,6 +138,13 @@ func configureClient(config *rest.Config) {
 				&api.ListOptions{},
 				&api.DeleteOptions{},
 			)
+			scheme.AddKnownTypes(
+				groupversion,
+				&Package{},
+				&PackageList{},
+				&api.ListOptions{},
+				&api.DeleteOptions{},
+			)
 			return nil
 		})
 	schemeBuilder.AddToScheme(api.Scheme)
@@ -193,6 +200,10 @@ func (fc *FissionClient) Timetriggers(ns string) TimetriggerInterface {
 func (fc *FissionClient) Messagequeuetriggers(ns string) MessagequeuetriggerInterface {
 	return MakeMessagequeuetriggerInterface(fc.tprClient, ns)
 }
+func (fc *FissionClient) Packages(ns string) PackageInterface {
+	return MakePackageInterface(fc.tprClient, ns)
+}
+
 func (fc *FissionClient) WaitForTPRs() {
 	waitForTPRs(fc.tprClient)
 }

tpr/package.go

@@ -0,0 +1,119 @@
+/*
+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 tpr
+
+import (
+	"k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/watch"
+	"k8s.io/client-go/1.5/rest"
+)
+
+type (
+	PackageInterface interface {
+		Create(*Package) (*Package, error)
+		Get(name string) (*Package, error)
+		Update(*Package) (*Package, error)
+		Delete(name string, options *api.DeleteOptions) error
+		List(opts api.ListOptions) (*PackageList, error)
+		Watch(opts api.ListOptions) (watch.Interface, error)
+	}
+
+	packageClient struct {
+		client    *rest.RESTClient
+		namespace string
+	}
+)
+
+func MakePackageInterface(tprClient *rest.RESTClient, namespace string) PackageInterface {
+	return &packageClient{
+		client:    tprClient,
+		namespace: namespace,
+	}
+}
+
+func (c *packageClient) Create(f *Package) (*Package, error) {
+	var result Package
+	err := c.client.Post().
+		Resource("packages").
+		Namespace(c.namespace).
+		Body(f).
+		Do().Into(&result)
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+func (c *packageClient) Get(name string) (*Package, error) {
+	var result Package
+	err := c.client.Get().
+		Resource("packages").
+		Namespace(c.namespace).
+		Name(name).
+		Do().Into(&result)
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+func (c *packageClient) Update(f *Package) (*Package, error) {
+	var result Package
+	err := c.client.Put().
+		Resource("packages").
+		Namespace(c.namespace).
+		Name(f.Metadata.Name).
+		Body(f).
+		Do().Into(&result)
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+func (c *packageClient) Delete(name string, opts *api.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.namespace).
+		Resource("packages").
+		Name(name).
+		Body(opts).
+		Do().
+		Error()
+}
+
+func (c *packageClient) List(opts api.ListOptions) (*PackageList, error) {
+	var result PackageList
+	err := c.client.Get().
+		Namespace(c.namespace).
+		Resource("packages").
+		VersionedParams(&opts, api.ParameterCodec).
+		Do().
+		Into(&result)
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+func (c *packageClient) Watch(opts api.ListOptions) (watch.Interface, error) {
+	return c.client.Get().
+		Prefix("watch").
+		Namespace(c.namespace).
+		Resource("packages").
+		VersionedParams(&opts, api.ParameterCodec).
+		Watch()
+}

tpr/tpr.go

@@ -95,6 +95,15 @@ func EnsureFissionTPRs(clientset *kubernetes.Clientset) error {
 			},
 			Description: "Message queue triggers for functions",
 		},
+		{
+			ObjectMeta: v1.ObjectMeta{
+				Name: "package.fission.io",
+			},
+			Versions: []v1beta1.APIVersion{
+				{Name: "v1"},
+			},
+			Description: "Packages: archives containing source or binaries for one or more functions",
+		},
 	}
 	for _, tpr := range tprs {
 		err := ensureTPR(clientset, &tpr)

tpr/tpr_test.go

@@ -46,9 +46,13 @@ func functionTests(tprClient *rest.RESTClient) {
 			Name: "hello",
 		},
 		Spec: fission.FunctionSpec{
-			Source: fission.Package{},
+			Source: fission.FunctionPackageRef{},
-			Deployment: fission.Package{
+			Deployment: fission.FunctionPackageRef{
-				Literal: []byte("hi"),
+				PackageRef: fission.PackageRef{
+					Name:      "foo",
+					Namespace: "bar",
+				},
+				FunctionName: "hello",
 			},
 			EnvironmentName: "xxx",
 		},
@@ -70,7 +74,7 @@ func functionTests(tprClient *rest.RESTClient) {
 	// read
 	f, err = fi.Get(function.Metadata.Name)
 	panicIf(err)
-	if len(f.Spec.Deployment.Literal) != len(function.Spec.Deployment.Literal) {
+	if f.Spec.Deployment.FunctionName != function.Spec.Deployment.FunctionName {
 		log.Panicf("Bad result from Get: %v", f)
 	}
 

tpr/types.go

@@ -33,9 +33,24 @@ import (
 //   5. Add the type to configureClient in client.go
 //   6. Add the type to EnsureFissionTPRs in tpr.go
 //   7. Add tests to tpr_test.go
+//   8. Add a CRUD Interface type (analogous to FunctionInterface in function.go)
+//   9. Add a getter method for your interface type to FissionClient in client.go
 //
 
 type (
+	// Packages. Think of these as function-level images.
+	Package struct {
+		unversioned.TypeMeta `json:",inline"`
+		Metadata             api.ObjectMeta      `json:"metadata"`
+		Spec                 fission.PackageSpec `json:"spec"`
+	}
+	PackageList struct {
+		unversioned.TypeMeta `json:",inline"`
+		Metadata             unversioned.ListMeta `json:"metadata"`
+
+		Items []Package `json:"items"`
+	}
+
 	// Functions.
 	Function struct {
 		unversioned.TypeMeta `json:",inline"`
@@ -144,6 +159,9 @@ func (w *Timetrigger) GetObjectKind() unversioned.ObjectKind {
 func (w *Messagequeuetrigger) GetObjectKind() unversioned.ObjectKind {
 	return &w.TypeMeta
 }
+func (w *Package) GetObjectKind() unversioned.ObjectKind {
+	return &w.TypeMeta
+}
 
 func (f *Function) GetObjectMeta() meta.Object {
 	return &f.Metadata
@@ -163,6 +181,9 @@ func (w *Timetrigger) GetObjectMeta() meta.Object {
 func (w *Messagequeuetrigger) GetObjectMeta() meta.Object {
 	return &w.Metadata
 }
+func (w *Package) GetObjectMeta() meta.Object {
+	return &w.Metadata
+}
 
 func (fl *FunctionList) GetObjectKind() unversioned.ObjectKind {
 	return &fl.TypeMeta
@@ -182,6 +203,9 @@ func (wl *TimetriggerList) GetObjectKind() unversioned.ObjectKind {
 func (wl *MessagequeuetriggerList) GetObjectKind() unversioned.ObjectKind {
 	return &wl.TypeMeta
 }
+func (wl *PackageList) GetObjectKind() unversioned.ObjectKind {
+	return &wl.TypeMeta
+}
 
 func (fl *FunctionList) GetListMeta() unversioned.List {
 	return &fl.Metadata
@@ -201,8 +225,6 @@ func (wl *TimetriggerList) GetListMeta() unversioned.List {
 func (wl *MessagequeuetriggerList) GetListMeta() unversioned.List {
 	return &wl.Metadata
 }
-
+func (wl *PackageList) GetListMeta() unversioned.List {
-// In the client-go TPR example, UnmarshalJSON is defined here for the
+	return &wl.Metadata
-// singular and list types.  That's supposed to be a workaround for
+}
-// some ugorji bug.  But we don't seem to need it, and all our tests
-// pass without it, so we don't define any UnmarshalJSON methods.

types.go

@@ -41,7 +41,7 @@ type (
 
 	// Package contains or references a collection of source or
 	// binary files.
-	Package struct {
+	PackageSpec struct {
 		// Type defines how the package is specified: literal or URL.
 		Type PackageType `json:"type"`
 
@@ -55,11 +55,24 @@ type (
 		// Checksum ensures the integrity of packages
 		// refereced by URL. Ignored for literals.
 		Checksum Checksum `json:"checksum"`
+	}
 
-		// EntryPoint optionally specifies an entry point in
+	PackageRef struct {
-		// the package. Each environment defines a default
+		Name      string
-		// entry point, but that can be overridden here.
+		Namespace string
-		EntryPoint string `json:"entrypoint"`
+	}
+	FunctionPackageRef struct {
+		PackageRef PackageRef
+
+		// FunctionName specifies a specific function within the package. This allows
+		// functions to share packages, by having different functions within the same
+		// package.
+		//
+		// Fission itself does not interpret this path. It is passed verbatim to
+		// build and runtime environments.
+		//
+		// This is optional: if unspecified, the environment has a default name.
+		FunctionName string `json:"functionName"`
 	}
 
 	// FunctionSpec describes the contents of the function.
@@ -71,11 +84,11 @@ type (
 
 		// Source is an source package for this function; it's used for the build step if
 		// the environment defines a build container.
-		Source Package `json:"source"`
+		Source FunctionPackageRef `json:"source"`
 
 		// Deployment is a deployable package for this function. This is the package that's
 		// loaded into the environment's runtime container.
-		Deployment Package `json:"deployment"`
+		Deployment FunctionPackageRef `json:"deployment"`
 	}
 
 	FunctionReferenceType string