Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Large functions: API proxy for storage svc, upload support in the CLI (#304)

Browse Source 
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.
This commit is contained in:
Soam Vasani
2017-08-31 15:42:59 -07:00
committed by GitHub
parent 9d1b096bac
commit 8d103fa35b
4 changed files with 115 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/api.go
+13 -2
View File
@@ -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)
controller/storagesvc.go
+38
View File
@@ -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)
}
fission-bundle/main.go
+34 -15
View File
@@ -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=<port>
fission-bundle --routerPort=<port> [--poolmgrUrl=<url>]
fission-bundle --poolmgrPort=<port> [--namespace=<namespace>] [--fission-namespace=<namespace>]
fission-bundle --kubewatcher [--routerUrl=<url>]
fission-bundle --storageServicePort=<port> --filePath=<filePath>
fission-bundle --logger
fission-bundle --timer [--routerUrl=<url>]
fission-bundle --mqt [--routerUrl=<url>]
Options:
--controllerPort=<port> Port that the controller should listen on.
--routerPort=<port> Port that the router should listen on.
--poolmgrPort=<port> Port that the poolmgr should listen on.
--poolmgrUrl=<url> Poolmgr URL. Not required if --poolmgrPort is specified.
--routerUrl=<url> Router URL.
--etcdUrl=<etcdUrl> Etcd URL.
--filepath=<filepath> Directory to store functions in.
--namespace=<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> Port that the controller should listen on.
--routerPort=<port> Port that the router should listen on.
--poolmgrPort=<port> Port that the poolmgr should listen on.
--storageServicePort=<port> Port that the storage service should listen on.
--poolmgrUrl=<url> Poolmgr URL. Not required if --poolmgrPort is specified.
--routerUrl=<url> Router URL.
--etcdUrl=<etcdUrl> Etcd URL.
--filePath=<filePath> Directory to store functions in.
--namespace=<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 {}
}
fission/function.go
+30 -18
View File
@@ -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