From 6f017cf40029d55fa2d84d1e5306c42a326a4843 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 25 Aug 2017 13:40:45 -0700 Subject: [PATCH] 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 | 2 + charts/fission-core/templates/deployment.yaml | 2 + charts/fission-core/values.yaml | 2 +- controller/api.go | 6 + controller/api_test.go | 17 +- controller/client/function.go | 9 - controller/client/package.go | 132 ++++++++++++++ controller/functionApi.go | 26 +-- controller/packageApi.go | 170 ++++++++++++++++++ environments/fetcher/fetcher.go | 20 ++- fission/function.go | 80 +++++---- poolmgr/gp.go | 46 +++-- poolmgr/poolmgr_test.go | 27 ++- test/test_utils.sh | 11 +- tpr/client.go | 11 ++ tpr/package.go | 119 ++++++++++++ tpr/tpr.go | 9 + tpr/tpr_test.go | 12 +- tpr/types.go | 32 +++- types.go | 27 ++- 20 files changed, 639 insertions(+), 121 deletions(-) create mode 100644 controller/client/package.go create mode 100644 controller/packageApi.go create mode 100644 tpr/package.go diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 2c9212bc..9943af44 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/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 --- diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 13f49a1e..6ef68a82 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/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 --- diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index b7ba41c7..5690cb7c 100644 --- a/charts/fission-core/values.yaml +++ b/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 diff --git a/controller/api.go b/controller/api.go index fa960cdb..30e60194 100644 --- a/controller/api.go +++ b/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") diff --git a/controller/api_test.go b/controller/api_test.go index 8ff71270..8600d432 100644 --- a/controller/api_test.go +++ b/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{ - Literal: []byte("code1"), + Deployment: fission.FunctionPackageRef{ + 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) - panicIf(err) - assert(bytes.Compare(code, testFunc.Spec.Deployment.Literal) == 0, "code from FunctionGetRawDeployment must match created function") - - testFunc.Spec.Deployment.Literal = []byte("code2") + testFunc.Spec.Deployment.FunctionName = "yyy" _, 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/") diff --git a/controller/client/function.go b/controller/client/function.go index 14d17431..73c14bec 100644 --- a/controller/client/function.go +++ b/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 diff --git a/controller/client/package.go b/controller/client/package.go new file mode 100644 index 00000000..1c25ef18 --- /dev/null +++ b/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 +} diff --git a/controller/functionApi.go b/controller/functionApi.go index d1e28b63..eaf86428 100644 --- a/controller/functionApi.go +++ b/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 - if raw != "" { - resp = []byte(f.Spec.Deployment.Literal) - } else { - resp, err = json.Marshal(f) - if err != nil { - a.respondWithError(w, err) - return - } + resp, err := json.Marshal(f) + if err != nil { + a.respondWithError(w, err) + return } a.respondWithSuccess(w, resp) } diff --git a/controller/packageApi.go b/controller/packageApi.go new file mode 100644 index 00000000..63601df1 --- /dev/null +++ b/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("")) +} diff --git a/environments/fetcher/fetcher.go b/environments/fetcher/fetcher.go index c389cbfe..23d5ed7d 100644 --- a/environments/fetcher/fetcher.go +++ b/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) diff --git a/fission/function.go b/fission/function.go index 8b6dc6c5..2760d186 100644 --- a/fission/function.go +++ b/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{ - Literal: pkgContents, + Deployment: fission.FunctionPackageRef{ + 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, } - - code, err := client.FunctionGetRawDeployment(m) + fn, err := client.FunctionGet(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) - } - if len(envName) > 0 { - function.Spec.EnvironmentName = envName + pkg.Spec.Literal = getPackageContents(fileName) + + _, err = client.PackageUpdate(pkg) + checkErr(err, "update package") } - _, err = client.FunctionUpdate(function) - checkErr(err, "update function") + if len(envName) > 0 { + function.Spec.EnvironmentName = envName + + _, err = client.FunctionUpdate(function) + checkErr(err, "update function") + } fmt.Printf("function '%v' updated\n", fnName) return err diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 99dab819..117e8879 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -50,20 +50,21 @@ const POD_PHASE_RUNNING string = "Running" type ( GenericPool struct { - env *tpr.Environment - replicas int32 // num idle pods - deployment *v1beta1.Deployment // kubernetes deployment - namespace string // namespace to keep our resources - podReadyTimeout time.Duration // timeout for generic pods to become ready - idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted - fsCache *functionServiceCache // cache funcSvc's by function, address and podname - useSvc bool // create k8s service for specialized pods - poolInstanceId string // small random string to uniquify pod names - fetcherImage string - kubernetesClient *kubernetes.Clientset - instanceId string // poolmgr instance id - labelsForPool map[string]string - requestChannel chan *choosePodRequest + env *tpr.Environment + replicas int32 // num idle pods + deployment *v1beta1.Deployment // kubernetes deployment + namespace string // namespace to keep our resources + podReadyTimeout time.Duration // timeout for generic pods to become ready + idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted + fsCache *functionServiceCache // cache funcSvc's by function, address and podname + useSvc bool // create k8s service for specialized pods + poolInstanceId string // small random string to uniquify pod names + fetcherImage string + fetcherImagePullPolicy v1.PullPolicy + kubernetesClient *kubernetes.Clientset + instanceId string // poolmgr instance id + 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{ { diff --git a/poolmgr/poolmgr_test.go b/poolmgr/poolmgr_test.go index 98e3427d..53137799 100644 --- a/poolmgr/poolmgr_test.go +++ b/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{}, - Deployment: fission.Package{ - Type: fission.PackageTypeLiteral, - Literal: []byte(`module.exports = async function(context) { return { status: 200, body: "Hello, world!\n" }; }`), + Source: fission.FunctionPackageRef{}, + Deployment: fission.FunctionPackageRef{ + PackageRef: fission.PackageRef{ + Name: p.Metadata.Name, + Namespace: p.Metadata.Namespace, + }, }, EnvironmentName: env.Metadata.Name, }, diff --git a/test/test_utils.sh b/test/test_utils.sh index 2146d4e7..219e154a 100755 --- a/test/test_utils.sh +++ b/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 } diff --git a/tpr/client.go b/tpr/client.go index f5f41066..a2752b92 100644 --- a/tpr/client.go +++ b/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) } diff --git a/tpr/package.go b/tpr/package.go new file mode 100644 index 00000000..47540937 --- /dev/null +++ b/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() +} diff --git a/tpr/tpr.go b/tpr/tpr.go index 08d4d362..108bedb8 100644 --- a/tpr/tpr.go +++ b/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) diff --git a/tpr/tpr_test.go b/tpr/tpr_test.go index 6e6b8026..d4e5a81f 100644 --- a/tpr/tpr_test.go +++ b/tpr/tpr_test.go @@ -46,9 +46,13 @@ func functionTests(tprClient *rest.RESTClient) { Name: "hello", }, Spec: fission.FunctionSpec{ - Source: fission.Package{}, - Deployment: fission.Package{ - Literal: []byte("hi"), + Source: fission.FunctionPackageRef{}, + Deployment: fission.FunctionPackageRef{ + 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) } diff --git a/tpr/types.go b/tpr/types.go index 84c04b2e..0941d419 100644 --- a/tpr/types.go +++ b/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 } - -// In the client-go TPR example, UnmarshalJSON is defined here for the -// 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. +func (wl *PackageList) GetListMeta() unversioned.List { + return &wl.Metadata +} diff --git a/types.go b/types.go index a1b644e9..e0cab61f 100644 --- a/types.go +++ b/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 - // the package. Each environment defines a default - // entry point, but that can be overridden here. - EntryPoint string `json:"entrypoint"` + PackageRef struct { + Name string + Namespace string + } + 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