diff --git a/environments/fetcher/cmd/build.sh b/environments/fetcher/cmd/build.sh index d4b6f514..813975e0 100755 --- a/environments/fetcher/cmd/build.sh +++ b/environments/fetcher/cmd/build.sh @@ -1,2 +1,2 @@ #!/bin/sh -GOOS=linux GOARCH=386 go build -o fetcher . +CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o fetcher . diff --git a/fission-bundle/build.sh b/fission-bundle/build.sh index 8f33c64f..f7378e10 100755 --- a/fission-bundle/build.sh +++ b/fission-bundle/build.sh @@ -1,2 +1,2 @@ #!/bin/sh -GOOS=linux GOARCH=386 go build +CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build diff --git a/fission-bundle/main.go b/fission-bundle/main.go index c3e7de27..45a9efcc 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -2,6 +2,7 @@ package main import ( "log" + "os" "strconv" "github.com/docopt/docopt-go" @@ -11,6 +12,7 @@ import ( "github.com/fission/fission/mqtrigger" "github.com/fission/fission/poolmgr" "github.com/fission/fission/router" + "github.com/fission/fission/storagesvc" "github.com/fission/fission/timer" ) @@ -57,6 +59,15 @@ func runMessageQueueMgr(routerUrl string) { } } +func runStorageSvc(port int, filePath string) { + subdir := os.Getenv("SUBDIR") + if len(subdir) == 0 { + subdir = "fission-functions" + } + storagesvc.RunStorageService(storagesvc.StorageTypeLocal, + filePath, subdir, port) +} + func getPort(portArg interface{}) int { portArgStr := portArg.(string) port, err := strconv.Atoi(portArgStr) diff --git a/glide.lock b/glide.lock index 347e05dc..66d2e07a 100644 --- a/glide.lock +++ b/glide.lock @@ -78,6 +78,10 @@ imports: - client/v2 - models - pkg/escape +- name: github.com/graymeta/stow + version: da285caa6daa337ae04a9ac603dbaf9009ffe687 + subpackages: + - local - name: github.com/jonboulle/clockwork version: 2eee05ed794112d45db504eb05aa693efd2b8b09 - name: github.com/juju/ratelimit diff --git a/glide.yaml b/glide.yaml index 9c324a89..b047bf53 100644 --- a/glide.yaml +++ b/glide.yaml @@ -38,3 +38,4 @@ import: version: ^v0.3.4 - package: github.com/nats-io/nats-streaming-server version: ^v0.4.0 +- package: github.com/graymeta/stow diff --git a/storagesvc/client/client.go b/storagesvc/client/client.go new file mode 100644 index 00000000..a65abf42 --- /dev/null +++ b/storagesvc/client/client.go @@ -0,0 +1,178 @@ +/* +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" + "errors" + "fmt" + "io" + "io/ioutil" + "mime/multipart" + "net/http" + "net/url" + "os" + "strings" + + "github.com/fission/fission/storagesvc" +) + +type ( + Client struct { + url string + } +) + +// Client creates a storage service client. +func MakeClient(url string) *Client { + return &Client{ + url: strings.TrimSuffix(url, "/") + "/v1", + } +} + +// Upload sends the local file pointed to by filePath to the storage +// service, along with the metadata. It returns a file ID that can be +// used to retrieve the file. +func (c *Client) Upload(filePath string, metadata *map[string]string) (string, error) { + fi, err := os.Stat(filePath) + if err != nil { + return "", err + } + fileSize := fi.Size() + + buf := &bytes.Buffer{} + bodyWriter := multipart.NewWriter(buf) + fileWriter, err := bodyWriter.CreateFormFile("uploadfile", filePath) + if err != nil { + return "", err + } + + f, err := os.Open(filePath) + if err != nil { + return "", err + } + + _, err = io.Copy(fileWriter, f) + if err != nil { + return "", err + } + + contentType := bodyWriter.FormDataContentType() + bodyWriter.Close() + + req, err := http.NewRequest(http.MethodPost, c.url+"/archive", buf) + if err != nil { + return "", err + } + req.Header["X-File-Size"] = []string{fmt.Sprintf("%v", fileSize)} + req.Header["Content-Type"] = []string{contentType} + + client := &http.Client{} + + resp, err := client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return "", err + } + if resp.StatusCode != http.StatusOK { + msg := fmt.Sprintf("Upload error %v", resp.Status) + return "", errors.New(msg) + } + + var ur storagesvc.UploadResponse + err = json.Unmarshal(body, &ur) + if err != nil { + return "", err + } + + return ur.ID, nil +} + +// GetUrl returns an HTTP URL that can be used to download the file pointed to by ID +func (c *Client) GetUrl(id string) string { + return fmt.Sprintf("%v/archive?id=%v", c.url, url.PathEscape(id)) +} + +// Download fetches the file identified by ID to the local file path. +// filePath must not exist. +func (c *Client) Download(id string, filePath string) error { + // url for id + url := c.GetUrl(id) + + // quit if file exists + _, err := os.Stat(filePath) + if err == nil || !os.IsNotExist(err) { + return errors.New(fmt.Sprintf("file already exists: %v", filePath)) + } + + // create + f, err := os.Create(filePath) + if err != nil { + return err + } + defer f.Close() + + // make request + resp, err := http.Get(url) + if err != nil { + fmt.Println(err) + os.Remove(filePath) + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + msg := fmt.Sprintf("HTTP error %v", resp.StatusCode) + os.Remove(filePath) + return errors.New(msg) + } + + // download and write data + _, err = io.Copy(f, resp.Body) + if err != nil { + return err + } + + return nil +} + +func (c *Client) Delete(id string) error { + url := c.GetUrl(id) + + req, err := http.NewRequest(http.MethodDelete, url, nil) + if err != nil { + return err + } + + client := &http.Client{} + + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return errors.New(fmt.Sprintf("HTTP error %v", resp.StatusCode)) + } + + return nil +} diff --git a/storagesvc/client/storagesvc_test.go b/storagesvc/client/storagesvc_test.go new file mode 100644 index 00000000..36bc6091 --- /dev/null +++ b/storagesvc/client/storagesvc_test.go @@ -0,0 +1,100 @@ +/* +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 client + +import ( + "bytes" + "fmt" + "io/ioutil" + "log" + "os" + "testing" + "time" + + "github.com/dchest/uniuri" + + "github.com/fission/fission/storagesvc" +) + +func panicIf(err error) { + if err != nil { + log.Panicf("Error: %v", err) + } +} + +func MakeTestFile(size int) *os.File { + f, err := ioutil.TempFile("", "storagesvc_test_") + panicIf(err) + + _, err = f.Write(bytes.Repeat([]byte("."), size)) + panicIf(err) + + return f +} + +func TestStorageService(t *testing.T) { + testId := uniuri.NewLen(8) + port := 8080 + + log.Printf("starting storage svc") + _ = storagesvc.RunStorageService( + storagesvc.StorageTypeLocal, "/tmp", testId, port) + + time.Sleep(time.Second) + client := MakeClient(fmt.Sprintf("http://localhost:%v/", port)) + + // generate a test file + tmpfile := MakeTestFile(10 * 1024) + defer os.Remove(tmpfile.Name()) + + // store it + metadata := make(map[string]string) + fileId, err := client.Upload(tmpfile.Name(), &metadata) + panicIf(err) + + // make a temp file for verification + retrievedfile, err := ioutil.TempFile("", "storagesvc_verify_") + panicIf(err) + os.Remove(retrievedfile.Name()) + + // retrieve uploaded file + err = client.Download(fileId, retrievedfile.Name()) + panicIf(err) + defer os.Remove(retrievedfile.Name()) + + // compare contents + contents1, err := ioutil.ReadFile(tmpfile.Name()) + panicIf(err) + contents2, err := ioutil.ReadFile(retrievedfile.Name()) + panicIf(err) + if bytes.Compare(contents1, contents2) != 0 { + log.Panicf("Contents don't match") + } + + // delete uploaded file + err = client.Delete(fileId) + panicIf(err) + + // make sure download fails + err = client.Download(fileId, "xxx") + if err == nil { + log.Panicf("Download succeeded but file isn't supposed to exist") + } + + // cleanup /tmp + os.RemoveAll(fmt.Sprintf("/tmp/", testId)) +} diff --git a/storagesvc/storagesvc.go b/storagesvc/storagesvc.go new file mode 100644 index 00000000..20057c22 --- /dev/null +++ b/storagesvc/storagesvc.go @@ -0,0 +1,236 @@ +/* +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 storagesvc + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "os" + "strconv" + + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + "github.com/graymeta/stow" + _ "github.com/graymeta/stow/local" + "github.com/satori/go.uuid" +) + +type ( + StorageType string + storageConfig struct { + storageType StorageType + localPath string + containerName string + // other stuff, such as google or s3 credentials, bucket names etc + } + + StorageService struct { + config storageConfig + location stow.Location + container stow.Container + port int + } + + UploadResponse struct { + ID string `json:"id"` + } +) + +const ( + StorageTypeLocal StorageType = "local" +) + +// Handle multipart file uploads. +func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) { + // handle upload + r.ParseMultipartForm(0) + file, handler, err := r.FormFile("uploadfile") + if err != nil { + http.Error(w, "missing upload file", 400) + return + } + defer file.Close() + + // stow wants the file size, but that's different from the + // content length, the content length being the size of the + // encoded file in the HTTP request. So we require an + // "X-File-Size" header in bytes. + + fileSizeS, ok := r.Header["X-File-Size"] + if !ok { + log.Printf("Missing X-File-Size") + http.Error(w, "missing X-File-Size header", 400) + return + } + + fileSize, err := strconv.Atoi(fileSizeS[0]) + if err != nil { + log.Printf("Error parsing x-file-size: '%v'", fileSizeS) + http.Error(w, "missing or bad X-File-Size header", 400) + return + } + + // TODO: allow headers to add more metadata (e.g. environment + // and function metadata) + log.Printf("Handling upload for %v", handler.Filename) + //fileMetadata := make(map[string]interface{}) + //fileMetadata["filename"] = handler.Filename + + // This is not the item ID (that's returned by Put) + // should we just use handler.Filename? what are the constraints here? + uploadName := uuid.NewV4().String() + + // save the file to the storage backend + item, err := ss.container.Put(uploadName, file, int64(fileSize), nil) + if err != nil { + log.Printf("Error saving uploaded file: '%v'", err) + http.Error(w, "Error saving uploaded file", 400) + return + } + + // respond with an ID that can be used to retrieve the file + ur := &UploadResponse{ + ID: item.ID(), + } + resp, err := json.Marshal(ur) + if err != nil { + http.Error(w, "Error marshaling response", 500) + return + } + w.Write(resp) +} + +func (ss *StorageService) getIdFromRequest(r *http.Request) (string, error) { + values := r.URL.Query() + ids, ok := values["id"] + if !ok || len(ids) == 0 { + return "", errors.New("Missing `id' query param") + } + return ids[0], nil +} + +func (ss *StorageService) deleteHandler(w http.ResponseWriter, r *http.Request) { + // get id from request + fileId, err := ss.getIdFromRequest(r) + if err != nil { + http.Error(w, err.Error(), 400) + } + + err = ss.container.RemoveItem(fileId) + if err != nil { + msg := fmt.Sprintf("Error deleting item: %v", err) + http.Error(w, msg, 500) + return + } + w.WriteHeader(http.StatusOK) +} + +func (ss *StorageService) downloadHandler(w http.ResponseWriter, r *http.Request) { + // get id from request + fileId, err := ss.getIdFromRequest(r) + if err != nil { + http.Error(w, err.Error(), 400) + } + + // Get the file (called "item" in stow's jargon), open it, + // stream it to response + + item, err := ss.container.Item(fileId) + if err != nil { + log.Printf("Error getting item id '%v': %v", fileId, err) + if err == stow.ErrNotFound { + http.Error(w, "Error retrieving item: not found", 404) + } else { + http.Error(w, "Error retrieving item", 400) + } + return + } + + f, err := item.Open() + if err != nil { + log.Printf("Error opening item %v: %v", fileId, err) + // TODO better http errors based on err + http.Error(w, "Error opening item", 400) + return + } + defer f.Close() + + _, err = io.Copy(w, f) + if err != nil { + log.Printf("Error writing response: %v", err) + http.Error(w, "Error writing response", 500) + return + } +} + +func MakeStorageService(sc *storageConfig) (*StorageService, error) { + ss := &StorageService{ + config: *sc, + } + + if sc.storageType != StorageTypeLocal { + return nil, errors.New("Storage types other than 'local' are not implemented") + } + + cfg := stow.ConfigMap{"path": sc.localPath} + loc, err := stow.Dial("local", cfg) + if err != nil { + log.Printf("Error initializing storage: %v", err) + return nil, err + } + ss.location = loc + + con, err := loc.CreateContainer(sc.containerName) + if err != nil { + log.Printf("Error initializing storage: %v", err) + return nil, err + } + ss.container = con + + return ss, nil +} + +func (ss *StorageService) Start(port int) { + r := mux.NewRouter() + r.HandleFunc("/v1/archive", ss.uploadHandler).Methods("POST") + r.HandleFunc("/v1/archive", ss.downloadHandler).Methods("GET") + r.HandleFunc("/v1/archive", ss.deleteHandler).Methods("DELETE") + + address := fmt.Sprintf(":%v", port) + log.Fatal(http.ListenAndServe(address, handlers.LoggingHandler(os.Stdout, r))) +} + +func RunStorageService(storageType StorageType, storagePath string, containerName string, port int) *StorageService { + // storage + ss, err := MakeStorageService(&storageConfig{ + storageType: storageType, + localPath: storagePath, + containerName: containerName, + }) + if err != nil { + log.Panicf("Error initializing storage: %v", err) + } + + // http handlers + go ss.Start(port) + + return ss +} diff --git a/test/test_utils.sh b/test/test_utils.sh index 219e154a..24609c1b 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -80,9 +80,11 @@ helm_install_fission() { echo "Installing fission" helm install \ --wait \ + --timeout 600 \ --name $id \ --set $helmVars \ --namespace $ns \ + --debug \ $ROOT/charts/fission-all } @@ -160,17 +162,17 @@ dump_fission_logs() { echo --- end $component logs --- } -dump_fission_resource() { +dump_fission_tpr() { type=$1 echo --- All objects of type $type --- kubectl --all-namespaces=true get $type -o yaml echo --- End objects of type $type --- } -dump_fission_resources() { - dump_fission_resource function.fission.io - dump_fission_resource httptrigger.fission.io - dump_fission_resource environment.fission.io +dump_fission_tprs() { + dump_fission_tpr function.fission.io + dump_fission_tpr httptrigger.fission.io + dump_fission_tpr environment.fission.io } dump_env_pods() { @@ -181,17 +183,27 @@ dump_env_pods() { echo --- End environment pods --- } +dump_all_fission_resources() { + ns=$1 + + echo "--- All objects in the fission namespace $ns ---" + kubectl -n $ns get all + echo "--- End objects in the fission namespace $ns ---" +} + dump_logs() { id=$1 ns=f-$id fns=f-func-$id + dump_all_fission_resources $ns dump_env_pods $fns + dump_fission_logs $ns $fns controller dump_fission_logs $ns $fns router dump_fission_logs $ns $fns poolmgr dump_function_pod_logs $ns $fns - dump_fission_resources + dump_fission_tprs } export FAILURES=0 @@ -226,7 +238,11 @@ install_and_test() { id=$(generate_test_id) trap "helm_uninstall_fission $id" EXIT - helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort + if ! helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort + then + dump_logs $id + exit 1 + fi wait_for_services $id set_environment $id