From 8d103fa35bf3bb458c8f7767015a51293e9b7581 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 31 Aug 2017 15:42:59 -0700 Subject: [PATCH] Large functions: API proxy for storage svc, upload support in the CLI (#304) This change contains CLI support for uploading large functions to the storage service. It also adds a reverse proxy into the storage service to the fission API. The helm chart is not yet updated to actually run the storage service -- that will come in the next change. --- controller/api.go | 15 ++++++++++-- controller/storagesvc.go | 38 +++++++++++++++++++++++++++++++ fission-bundle/main.go | 49 ++++++++++++++++++++++++++++------------ fission/function.go | 48 ++++++++++++++++++++++++--------------- 4 files changed, 115 insertions(+), 35 deletions(-) create mode 100644 controller/storagesvc.go diff --git a/controller/api.go b/controller/api.go index 30e60194..1e3a518d 100644 --- a/controller/api.go +++ b/controller/api.go @@ -35,7 +35,8 @@ import ( type ( API struct { - fissionClient *tpr.FissionClient + fissionClient *tpr.FissionClient + storageServiceUrl string } logDBConfig struct { @@ -46,7 +47,16 @@ type ( ) func MakeAPI() (*API, error) { - return makeTPRBackedAPI() + api, err := makeTPRBackedAPI() + + u := os.Getenv("STORAGE_SERVICE_URL") + if len(u) > 0 { + api.storageServiceUrl = strings.TrimSuffix(u, "/") + } else { + api.storageServiceUrl = "http://storagesvc" + } + + return api, err } func (api *API) respondWithSuccess(w http.ResponseWriter, resp []byte) { @@ -149,6 +159,7 @@ func (api *API) Serve(port int) { r.HandleFunc("/v2/triggers/messagequeue/{mqTrigger}", api.MessageQueueTriggerApiDelete).Methods("DELETE") r.HandleFunc("/proxy/{dbType}", api.FunctionLogsApiPost).Methods("POST") + r.HandleFunc("/proxy/storage/v1/archive", api.StorageServiceProxy) address := fmt.Sprintf(":%v", port) diff --git a/controller/storagesvc.go b/controller/storagesvc.go new file mode 100644 index 00000000..7cd965d4 --- /dev/null +++ b/controller/storagesvc.go @@ -0,0 +1,38 @@ +/* +Copyright 2017 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 ( + "fmt" + "log" + "net/http" + "net/http/httputil" + "net/url" +) + +func (api *API) StorageServiceProxy(w http.ResponseWriter, r *http.Request) { + u := api.storageServiceUrl + "/v1/archive" + ssUrl, err := url.Parse(u) + if err != nil { + msg := fmt.Sprintf("Error parsing url %v: %v", u, err) + log.Println(msg) + http.Error(w, msg, 500) + return + } + proxy := httputil.NewSingleHostReverseProxy(ssUrl) + proxy.ServeHTTP(w, r) +} diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 45a9efcc..659d7576 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -90,33 +90,46 @@ func main() { Use it to start one or more of the fission servers: - Controller keeps track of functions, triggers, environments. + Controller is a stateless API frontend for fission resources. - Pool manager maintains a pool of generalized function containers, and specializes them on-demand. Poolmgr must be run from a pod in a Kubernetes cluster. + Pool manager maintains a pool of generalized function containers, and + specializes them on-demand. Poolmgr must be run from a pod in a + Kubernetes cluster. - Router implements HTTP triggers: it routes to running instances, working with the controller and poolmgr. + Router implements HTTP triggers: it routes to running instances, + working with the controller and poolmgr. + + Kubewatcher implements Kubernetes Watch triggers: it watches + Kubernetes resources and invokes functions described in the + KubernetesWatchTrigger. + + The storage service implements storage for functions too large to fit + in the Kubernetes API resource object. It supports various storage + backends. Usage: fission-bundle --controllerPort= fission-bundle --routerPort= [--poolmgrUrl=] fission-bundle --poolmgrPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] + fission-bundle --storageServicePort= --filePath= fission-bundle --logger fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] Options: - --controllerPort= Port that the controller should listen on. - --routerPort= Port that the router should listen on. - --poolmgrPort= Port that the poolmgr should listen on. - --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. - --routerUrl= Router URL. - --etcdUrl= Etcd URL. - --filepath= Directory to store functions in. - --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. - --kubewatcher Start Kubernetes events watcher. - --logger Start logger. - --timer Start Timer. - --mqt Start message queue trigger. + --controllerPort= Port that the controller should listen on. + --routerPort= Port that the router should listen on. + --poolmgrPort= Port that the poolmgr should listen on. + --storageServicePort= Port that the storage service should listen on. + --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. + --routerUrl= Router URL. + --etcdUrl= Etcd URL. + --filePath= Directory to store functions in. + --namespace= Kubernetes namespace in which to run function containers. Defaults to 'fission-function'. + --kubewatcher Start Kubernetes events watcher. + --logger Start logger. + --timer Start Timer. + --mqt Start message queue trigger. ` arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false) if err != nil { @@ -160,5 +173,11 @@ Options: runMessageQueueMgr(routerUrl) } + if arguments["--storageServicePort"] == true { + port := getPort(arguments["--storageServicePort"]) + filePath := arguments["--filePath"].(string) + runStorageSvc(port, filePath) + } + select {} } diff --git a/fission/function.go b/fission/function.go index bf66cb37..4f5add11 100644 --- a/fission/function.go +++ b/fission/function.go @@ -34,6 +34,7 @@ import ( "github.com/fission/fission" "github.com/fission/fission/controller/client" "github.com/fission/fission/fission/logdb" + storageSvcClient "github.com/fission/fission/storagesvc/client" "github.com/fission/fission/tpr" ) @@ -43,24 +44,43 @@ func fileSize(filePath string) int64 { return info.Size() } +// updatePackageSpecWithFile uploads or serializes the file into the +// provided package object +func updatePackageSpecWithFile(client *client.Client, pkgSpec *fission.PackageSpec, fileName string) { + if fileSize(fileName) < fission.PackageLiteralSizeLimit { + pkgContents := getPackageContents(fileName) + + pkgSpec.Type = fission.PackageTypeLiteral + pkgSpec.Literal = pkgContents + } else { + u := strings.TrimSuffix(client.Url, "/") + "/proxy/storage" + ssClient := storageSvcClient.MakeClient(u) + + // TODO add a progress bar + id, err := ssClient.Upload(fileName, nil) + checkErr(err, fmt.Sprintf("upload file %v", fileName)) + + archiveUrl := ssClient.GetUrl(id) + + pkgSpec.Type = fission.PackageTypeUrl + pkgSpec.URL = archiveUrl + } +} + // createPackageFromFile is a function that helps to upload the content // of given file to controller to create a TPR package resource, and then // return a function package reference for further usage. func createPackageFromFile(client *client.Client, fnName string, fileName string) fission.FunctionPackageRef { - // TODO fallback to uploading + setting a Package URL - checkFileSize(fileName) - 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, - }, } + + updatePackageSpecWithFile(client, &pkg.Spec, fileName) + _, err := client.PackageCreate(pkg) checkErr(err, "upload package") @@ -75,8 +95,6 @@ func createPackageFromFile(client *client.Client, fnName string, fileName string // updatePackageContents is a function that reads content from given file // and updates the package content of TPR package resource. func updatePackageContents(client *client.Client, pkgName string, fileName string) error { - // TODO fallback to uploading + setting a Package URL - checkFileSize(fileName) pkg, err := client.PackageGet(&api.ObjectMeta{ Name: pkgName, Namespace: api.NamespaceDefault, @@ -84,19 +102,13 @@ func updatePackageContents(client *client.Client, pkgName string, fileName strin if err != nil { return errors.New(fmt.Sprintf("read package '%v'", pkgName)) } - pkg.Spec.Literal = getPackageContents(fileName) + + updatePackageSpecWithFile(client, &pkg.Spec, fileName) + _, err = client.PackageUpdate(pkg) return err } -func checkFileSize(fileName string) { - if fileSize(fileName) > fission.PackageLiteralSizeLimit { - // TODO fallback to uploading + setting a Package URL - fmt.Printf("File size >256k not supported yet") - os.Exit(1) - } -} - func getPackageContents(filePath string) []byte { var code []byte var err error