diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index ad22dfad..8edfdee0 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -471,6 +471,9 @@ spec: imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] args: ["--storageServicePort", "8000", "--filePath", "/fission"] + env: + - name: PRUNE_INTERVAL + value: "{{.Values.pruneInterval}}" volumeMounts: - name: fission-storage mountPath: /fission diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index 11536b6f..e7ba1a56 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -67,3 +67,8 @@ persistence: ## Analytics let us count how many people installed fission. Set to ## false to disable analytics. analytics: true + +## Archive pruner is a garbage collector for archives on the fission storage service. +## This interval configures the frequency at which it runs inside the storagesvc pod. +## The value is in minutes. +pruneInterval: 60 \ No newline at end of file diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 8b5298f9..13b25b5e 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -287,6 +287,9 @@ spec: imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] args: ["--storageServicePort", "8000", "--filePath", "/fission"] + env: + - name: PRUNE_INTERVAL + value: "{{.Values.pruneInterval}}" volumeMounts: - name: fission-storage mountPath: /fission diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index 08b0c475..9438809a 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -51,3 +51,8 @@ persistence: ## Analytics let us count how many people installed fission. Set to ## false to disable analytics. analytics: true + +## Archive pruner is a garbage collector for archives on the fission storage service. +## This interval configures the frequency at which it runs inside the storagesvc pod. +## The value is in minutes. +pruneInterval: 60 \ No newline at end of file diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 6b00af11..66b77b96 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -60,8 +60,9 @@ func runStorageSvc(port int, filePath string) { if len(subdir) == 0 { subdir = "fission-functions" } + enableArchivePruner := true storagesvc.RunStorageService(storagesvc.StorageTypeLocal, - filePath, subdir, port) + filePath, subdir, port, enableArchivePruner) } func runBuilderMgr(storageSvcUrl string, envBuilderNamespace string) { diff --git a/storagesvc/README.md b/storagesvc/README.md new file mode 100644 index 00000000..3a087cd7 --- /dev/null +++ b/storagesvc/README.md @@ -0,0 +1,26 @@ +# StorageSvc +StorageSvc consists of 3 components +* Storage http request handler +* StowClient +* ArchivePruner + +## StorageSvc +This is the HTTP handler that serves requests to : +* upload archive into a storage +* fetch an archive from storage +* delete archive from storage + +## StowClient +This is the storage interface layer that interacts with stow package. +It provides methods to: +* write a file to storage +* retrieve a file from storage +* delete a file from storage +* get all files on storage + +## ArchivePruner +This acts like a cron job to clean up orphaned archives from storage. +By default configured to run every hour. The value can be set in Values.yaml to any preferred interval. + + + diff --git a/storagesvc/archivePruner.go b/storagesvc/archivePruner.go new file mode 100644 index 00000000..f0d8bf4b --- /dev/null +++ b/storagesvc/archivePruner.go @@ -0,0 +1,146 @@ +/* +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 ( + "time" + + log "github.com/sirupsen/logrus" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission/crd" +) + +type ArchivePruner struct { + crdClient *crd.FissionClient + archiveChan chan (string) + stowClient *StowClient + pruneInterval time.Duration +} + +const defaultPruneInterval int = 60 // in minutes + +func MakeArchivePruner(stowClient *StowClient, pruneInterval time.Duration) (*ArchivePruner, error) { + crdClient, _, _, err := crd.MakeFissionClient() + if err != nil { + return nil, err + } + + return &ArchivePruner{ + crdClient: crdClient, + archiveChan: make(chan string), + stowClient: stowClient, + pruneInterval: pruneInterval, + }, nil +} + +// pruneArchives listens to archiveChannel for archive ids that need to be deleted +func (pruner *ArchivePruner) pruneArchives() { + log.Info("listening to archiveChannel to prune archives") + for { + select { + case archiveID := <-pruner.archiveChan: + log.WithField("archive ID", archiveID).Info("sending delete request") + if err := pruner.stowClient.removeFileByID(archiveID); err != nil { + // logging the error and continuing with other deletions. + // hopefully this archive will be deleted in the next iteration. + log.WithField("archiveID", archiveID).WithError(err).Error("Ignoring error while deleting archive") + } + } + } +} + +// insertArchive method just writes the archive ID into the channel. +func (pruner *ArchivePruner) insertArchive(archiveID string) { + pruner.archiveChan <- archiveID +} + +// A user may have deleted pkgs with kubectl or fission cli. That only deletes crd.Package objects from kubernetes +// and not the archives that are referenced by them, leaving the archives as orphans. +// getOrphanArchives reaps the orphaned archives. +func (pruner *ArchivePruner) getOrphanArchives() error { + log.Info("getting orphan archives") + archivesRefByPkgs := make([]string, 0) + var archiveID string + + // get all pkgs from kubernetes + pkgList, err := pruner.crdClient.Packages(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + log.WithError(err).Error("Error getting package list from kubernetes") + return err + } + + // extract archives referenced by these pkgs + for _, pkg := range pkgList.Items { + if pkg.Spec.Deployment.URL != "" { + archiveID, err = getQueryParamValue(pkg.Spec.Deployment.URL, "id") + if err != nil { + log.WithError(err).Error("Error extracting value of archiveID from url") + return err + } + archivesRefByPkgs = append(archivesRefByPkgs, archiveID) + } + if pkg.Spec.Source.URL != "" { + archiveID, err = getQueryParamValue(pkg.Spec.Source.URL, "id") + if err != nil { + log.WithError(err).Error("Error extracting value of archiveID from url") + return err + } + archivesRefByPkgs = append(archivesRefByPkgs, archiveID) + } + } + + log.WithField("list", "archives referenced by packages").Debugf("%s", archivesRefByPkgs) + + // get all archives on storage + // out of them, there may be some just created but not referenced by packages yet. + // need to filter them out. + archivesInStorage, err := pruner.stowClient.getItemIDsWithFilter(filterItemCreatedAMinuteAgo, time.Now()) + if err != nil { + log.WithError(err).Error("Error getting items from storage") + return err + } + log.WithField("list", "archives in storage").Debugf("%s", archivesInStorage) + + // difference of the two lists gives us the list of orphan archives. This is just a brute force approach. + // need to do something more optimal at scale. + orphanedArchives := getDifferenceOfLists(archivesInStorage, archivesRefByPkgs) + log.WithField("list", "orphan archives").Debugf("%s", orphanedArchives) + + // send each orphan archive away for deletion + for _, archiveID = range orphanedArchives { + pruner.insertArchive(archiveID) + } + + return nil +} + +// Start starts a go routine that listens to a channel for archive IDs that need to deleted. +// Also wakes up at regular intervals to make a list of archive IDs that need to be reaped +// and sends them over to the channel for deletion +func (pruner *ArchivePruner) Start() { + ticker := time.NewTicker(pruner.pruneInterval * time.Minute) + go pruner.pruneArchives() + for { + select { + case <-ticker.C: + // This method fetches unused archive IDs and sends them to archiveChannel for deletion + // silencing the errors, hoping they go away in next iteration. + pruner.getOrphanArchives() + } + } +} diff --git a/storagesvc/client/storagesvc_test.go b/storagesvc/client/storagesvc_test.go index d94a60c5..a0f2b684 100644 --- a/storagesvc/client/storagesvc_test.go +++ b/storagesvc/client/storagesvc_test.go @@ -49,10 +49,11 @@ func MakeTestFile(size int) *os.File { func TestStorageService(t *testing.T) { testId := uniuri.NewLen(8) port := 8080 + enableArchivePruner := false - log.Printf("starting storage svc") + log.Println("starting storage svc") _ = storagesvc.RunStorageService( - storagesvc.StorageTypeLocal, "/tmp", testId, port) + storagesvc.StorageTypeLocal, "/tmp", testId, port, enableArchivePruner) time.Sleep(time.Second) client := MakeClient(fmt.Sprintf("http://localhost:%v/", port)) @@ -82,7 +83,7 @@ func TestStorageService(t *testing.T) { contents2, err := ioutil.ReadFile(retrievedfile.Name()) panicIf(err) if !bytes.Equal(contents1, contents2) { - log.Panicf("Contents don't match") + log.Panic("Contents don't match") } // delete uploaded file @@ -92,7 +93,7 @@ func TestStorageService(t *testing.T) { // make sure download fails err = client.Download(fileId, "xxx") if err == nil { - log.Panicf("Download succeeded but file isn't supposed to exist") + log.Panic("Download succeeded but file isn't supposed to exist") } // cleanup /tmp diff --git a/storagesvc/storagesvc.go b/storagesvc/storagesvc.go index d4ebf5ac..4f253f71 100644 --- a/storagesvc/storagesvc.go +++ b/storagesvc/storagesvc.go @@ -20,33 +20,21 @@ import ( "encoding/json" "errors" "fmt" - "io" - "log" "net/http" "os" "strconv" + "time" "github.com/gorilla/handlers" "github.com/gorilla/mux" - "github.com/graymeta/stow" _ "github.com/graymeta/stow/local" - "github.com/satori/go.uuid" + log "github.com/sirupsen/logrus" ) 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 + storageClient *StowClient + port int } UploadResponse struct { @@ -54,10 +42,6 @@ type ( } ) -const ( - StorageTypeLocal StorageType = "local" -) - // Handle multipart file uploads. func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) { // handle upload @@ -76,39 +60,34 @@ func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) fileSizeS, ok := r.Header["X-File-Size"] if !ok { - log.Printf("Missing X-File-Size") + log.Error("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) + log.WithError(err).Errorf("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) + log.Infof("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) + id, err := ss.storageClient.putFile(file, int64(fileSize)) if err != nil { - log.Printf("Error saving uploaded file: '%v'", err) - http.Error(w, "Error saving uploaded file", 400) + log.WithError(err).Error("Error saving uploaded file") + http.Error(w, "Error saving uploaded file", 500) return } // respond with an ID that can be used to retrieve the file ur := &UploadResponse{ - ID: item.ID(), + ID: id, } resp, err := json.Marshal(ur) if err != nil { @@ -135,7 +114,7 @@ func (ss *StorageService) deleteHandler(w http.ResponseWriter, r *http.Request) return } - err = ss.container.RemoveItem(fileId) + err = ss.storageClient.removeFileByID(fileId) if err != nil { msg := fmt.Sprintf("Error deleting item: %v", err) http.Error(w, msg, 500) @@ -154,75 +133,27 @@ func (ss *StorageService) downloadHandler(w http.ResponseWriter, r *http.Request // Get the file (called "item" in stow's jargon), open it, // stream it to response - - item, err := ss.container.Item(fileId) + err = ss.storageClient.copyFileToStream(fileId, w) if err != nil { - log.Printf("Error getting item id '%v': %v", fileId, err) - if err == stow.ErrNotFound { + log.WithError(err).Errorf("Error getting item id '%v'", fileId) + if err == ErrNotFound { http.Error(w, "Error retrieving item: not found", 404) - } else { + } else if err == ErrRetrievingItem { http.Error(w, "Error retrieving item", 400) + } else if err == ErrOpeningItem { + http.Error(w, "Error opening item", 400) + } else if err == ErrWritingFileIntoResponse { + http.Error(w, "Error writing response", 500) } 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, +func MakeStorageService(storageClient *StowClient, port int) *StorageService { + return &StorageService{ + storageClient: storageClient, + port: port, } - - 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 os.IsExist(err) { - var cons []stow.Container - var cursor string - - // use location.Containers to find containers that match the prefix (container name) - cons, cursor, err = loc.Containers(sc.containerName, stow.CursorStart, 1) - if err == nil { - if !stow.IsCursorEnd(cursor) { - // Should only have one storage container - err = errors.New("Found more than one matched storage containers") - } else { - con = cons[0] - } - } - } - 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) { @@ -235,19 +166,34 @@ func (ss *StorageService) Start(port int) { 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, - }) +func RunStorageService(storageType StorageType, storagePath string, containerName string, port int, enablePruner bool) *StorageService { + // initialize logger + log.SetLevel(log.InfoLevel) + + // create a storage client + storageClient, err := MakeStowClient(storageType, storagePath, containerName) if err != nil { - log.Panicf("Error initializing storage: %v", err) + log.Fatalf("Error creating stowClient: %v", err) } - // http handlers - go ss.Start(port) + // create http handlers + storageService := MakeStorageService(storageClient, port) + go storageService.Start(port) - return ss + // enablePruner prevents storagesvc unit test from needing to talk to kubernetes + if enablePruner { + // get the prune interval and start the archive pruner + pruneInterval, err := strconv.Atoi(os.Getenv("PRUNE_INTERVAL")) + if err != nil { + pruneInterval = defaultPruneInterval + } + pruner, err := MakeArchivePruner(storageClient, time.Duration(pruneInterval)) + if err != nil { + log.Fatalf("Error creating archivePruner: %v", err) + } + go pruner.Start() + } + + log.Info("Storage service started") + return storageService } diff --git a/storagesvc/stowClient.go b/storagesvc/stowClient.go new file mode 100644 index 00000000..e2a87ef1 --- /dev/null +++ b/storagesvc/stowClient.go @@ -0,0 +1,201 @@ +/* +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 ( + "errors" + "io" + "mime/multipart" + "os" + "time" + + "github.com/graymeta/stow" + _ "github.com/graymeta/stow/local" + "github.com/satori/go.uuid" + log "github.com/sirupsen/logrus" +) + +type ( + StorageType string + + storageConfig struct { + storageType StorageType + localPath string + containerName string + // other stuff, such as google or s3 credentials, bucket names etc + } + + StowClient struct { + config *storageConfig + location stow.Location + container stow.Container + } +) + +const ( + StorageTypeLocal StorageType = "local" + PaginationSize int = 10 +) + +var ( + ErrNotFound = errors.New("not found") + ErrRetrievingItem = errors.New("unable to retrieve item") + ErrOpeningItem = errors.New("unable to open item") + ErrWritingFile = errors.New("unable to write file") + ErrWritingFileIntoResponse = errors.New("unable to copy item into http response") +) + +func MakeStowClient(storageType StorageType, storagePath string, containerName string) (*StowClient, error) { + if storageType != StorageTypeLocal { + return nil, errors.New("Storage types other than 'local' are not implemented") + } + + config := &storageConfig{ + storageType: storageType, + localPath: storagePath, + containerName: containerName, + } + + stowClient := &StowClient{ + config: config, + } + + cfg := stow.ConfigMap{"path": config.localPath} + loc, err := stow.Dial("local", cfg) + if err != nil { + log.WithError(err).Error("Error initializing storage") + return nil, err + } + stowClient.location = loc + + con, err := loc.CreateContainer(config.containerName) + if os.IsExist(err) { + var cons []stow.Container + var cursor string + + // use location.Containers to find containers that match the prefix (container name) + cons, cursor, err = loc.Containers(config.containerName, stow.CursorStart, 1) + if err == nil { + if !stow.IsCursorEnd(cursor) { + // Should only have one storage container + err = errors.New("Found more than one matched storage containers") + } else { + con = cons[0] + } + } + } + if err != nil { + log.WithError(err).Error("Error initializing storage") + return nil, err + } + stowClient.container = con + + return stowClient, nil +} + +// putFile writes the file on the storage +func (client *StowClient) putFile(file multipart.File, fileSize int64) (string, error) { + // 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 := client.container.Put(uploadName, file, int64(fileSize), nil) + if err != nil { + log.WithError(err).Errorf("Error writing file: %s on storage", uploadName) + return "", ErrWritingFile + } + + log.Debugf("Successfully wrote file:%s on storage", uploadName) + return item.ID(), nil +} + +// copyFileToStream gets the file contents into a stream +func (client *StowClient) copyFileToStream(fileId string, w io.Writer) error { + item, err := client.container.Item(fileId) + if err != nil { + if err == stow.ErrNotFound { + return ErrNotFound + } else { + return ErrRetrievingItem + } + } + + f, err := item.Open() + if err != nil { + return ErrOpeningItem + } + defer f.Close() + + _, err = io.Copy(w, f) + if err != nil { + return ErrWritingFileIntoResponse + } + + log.Debugf("successfully wrote file: %s into httpresponse", fileId) + return nil +} + +// removeFileByID deletes the file from storage +func (client *StowClient) removeFileByID(itemID string) error { + return client.container.RemoveItem(itemID) +} + +// filter defines an interface to filter out items from a set of items +type filter func(stow.Item, interface{}) bool + +// This method returns all items in a container, filtering out items based on the filter function passed to it +func (client *StowClient) getItemIDsWithFilter(filterFunc filter, filterFuncParam interface{}) ([]string, error) { + cursor := stow.CursorStart + var items []stow.Item + var err error + + archiveIDList := make([]string, 0) + + for { + items, cursor, err = client.container.Items(stow.NoPrefix, cursor, PaginationSize) + if err != nil { + log.WithError(err).Error("Error getting items from container") + return nil, err + } + + for _, item := range items { + isItemFilterable := filterFunc(item, filterFuncParam) + if isItemFilterable { + continue + } + archiveIDList = append(archiveIDList, item.ID()) + } + + if stow.IsCursorEnd(cursor) { + break + } + } + + return archiveIDList, nil +} + +// filterItemCreatedAMinuteAgo is one type of filter function that filters out items created less than a minute ago. +// More filter functions can be written if needed, as long as they are of type filter +func filterItemCreatedAMinuteAgo(item stow.Item, currentTime interface{}) bool { + itemLastModTime, _ := item.LastMod() + if currentTime.(time.Time).Sub(itemLastModTime) < 1*time.Minute { + log.Debugf("item: %s created less than a minute ago: %v", item.ID(), itemLastModTime) + return true + } + return false +} diff --git a/storagesvc/util.go b/storagesvc/util.go new file mode 100644 index 00000000..a0fa2f5d --- /dev/null +++ b/storagesvc/util.go @@ -0,0 +1,53 @@ +/* +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 ( + "log" + "net/url" +) + +func getQueryParamValue(urlString string, queryParam string) (string, error) { + url, err := url.Parse(urlString) + if err != nil { + log.Printf("Error parsing URL string: %s into URL", urlString) + return "", err + } + return url.Query().Get(queryParam), nil +} + +func getDifferenceOfLists(firstList []string, secondList []string) []string { + tempMap := make(map[string]int) + differenceList := make([]string, 0) + + for _, item := range firstList { + tempMap[item] = 1 + } + + for _, item := range secondList { + _, ok := tempMap[item] + if ok { + delete(tempMap, item) + } + } + + for k := range tempMap { + differenceList = append(differenceList, k) + } + + return differenceList +} diff --git a/test/build_and_test.sh b/test/build_and_test.sh index 1f4549dc..be23396e 100755 --- a/test/build_and_test.sh +++ b/test/build_and_test.sh @@ -16,6 +16,7 @@ FETCHER_IMAGE=$REPO/fetcher FLUENTD_IMAGE=gcr.io/fission-ci/fluentd BUILDER_IMAGE=$REPO/builder TAG=test +PRUNE_INTERVAL=1 # this variable controls the interval to run archivePruner. The unit is in minutes. dump_system_info @@ -35,4 +36,4 @@ build_and_push_fluentd $FLUENTD_IMAGE:$TAG build_fission_cli -install_and_test $IMAGE $TAG $FETCHER_IMAGE $TAG $FLUENTD_IMAGE $TAG +install_and_test $IMAGE $TAG $FETCHER_IMAGE $TAG $FLUENTD_IMAGE $TAG $PRUNE_INTERVAL diff --git a/test/test_utils.sh b/test/test_utils.sh index 195669a3..7513bfb5 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -162,11 +162,12 @@ helm_install_fission() { routerNodeport=$7 fluentdImage=$8 fluentdImageTag=$9 + pruneInterval="${10}" ns=f-$id fns=f-func-$id - helmVars=image=$image,imageTag=$imageTag,fetcherImage=$fetcherImage,fetcherImageTag=$fetcherImageTag,functionNamespace=$fns,controllerPort=$controllerNodeport,routerPort=$routerNodeport,pullPolicy=Always,analytics=false,logger.fluentdImage=$fluentdImage,logger.fluentdImageTag=$fluentdImageTag + helmVars=image=$image,imageTag=$imageTag,fetcherImage=$fetcherImage,fetcherImageTag=$fetcherImageTag,functionNamespace=$fns,controllerPort=$controllerNodeport,routerPort=$routerNodeport,pullPolicy=Always,analytics=false,logger.fluentdImage=$fluentdImage,logger.fluentdImageTag=$fluentdImageTag,pruneInterval=$pruneInterval timeout 30 bash -c "helm_setup" @@ -369,6 +370,7 @@ install_and_test() { fetcherImageTag=$4 fluentdImage=$5 fluentdImageTag=$6 + pruneInterval=$7 controllerPort=31234 routerPort=31235 @@ -377,7 +379,7 @@ install_and_test() { id=$(generate_test_id) trap "helm_uninstall_fission $id" EXIT - if ! helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort $fluentdImage $fluentdImageTag + if ! helm_install_fission $id $image $imageTag $fetcherImage $fetcherImageTag $controllerPort $routerPort $fluentdImage $fluentdImageTag $pruneInterval then dump_logs $id exit 1 diff --git a/test/tests/test_archive_pruner.sh b/test/tests/test_archive_pruner.sh new file mode 100755 index 00000000..cc85ea02 --- /dev/null +++ b/test/tests/test_archive_pruner.sh @@ -0,0 +1,117 @@ +#!/bin/bash +set -euo pipefail + +# global variables +pkg="" +http_status="" +url="" + +cleanup() { + if [ -e "test-deploy-pkg.zip" ]; then + rm -rf test-deploy-pkg.zip test_dir + fi + if [ -e "/tmp/file" ]; then + rm -rf /tmp/file + fi +} + +create_archive() { + echo "Creating an archive" + mkdir test_dir + dd if=/dev/urandom of=test_dir/dynamically_generated_file bs=256k count=1 + printf 'def main():\n return "Hello, world!"' > test_dir/hello.py + zip -jr test-deploy-pkg.zip test_dir/ +} + +create_package() { + echo "Creating package" + pkg=$(fission package create --deploy "test-deploy-pkg.zip" --env python| cut -f2 -d' '| tr -d \') +} + +delete_package() { + echo "Deleting package: $1" + fission package delete --name $1 +} + +get_archive_url_from_package() { + echo "Getting archive URL from package: $1" + url=`kubectl get package $1 -ojsonpath='{.spec.deployment.url}'` +} + +get_archive_from_storage() { + http_status=`curl -sw "%{http_code}" $1 -o /tmp/file` +} + +#1. declare trap to cleanup for all the required signals +#2. create an archives with large files such that total size of archive is > 256KB +#3. create 2 pkgs referencing those archives +#4. delete both the packages +#5. verify archives are not recycled . this handles the case where archives are just created but not referenced by pkgs yet. +#6. sleep for two minutes +#7. now verify that both get deleted. +main() { + # trap + trap cleanup EXIT + + # create a huge archive + create_archive + echo "created archive test-deploy-pkg.zip" + + # create packages with the huge archive + create_package + pkg_1=$pkg + get_archive_url_from_package $pkg_1 + url_1=$url + echo "pkg: $pkg_1, archive_url : $url_1" + + create_package + pkg_2=$pkg + get_archive_url_from_package $pkg_2 + url_2=$url + echo "pkg: $pkg_2, archive_url : $url_2" + + # delete packages + delete_package $pkg_1 + delete_package $pkg_2 + echo "deleted packages : $pkg_1 $pkg_2" + + # curl on the archive url + get_archive_from_storage $url_1 + echo "http_status for $url_1 : $http_status" + if [ "$http_status" -ne "200" ]; then + echo "Archive $url_1 absent on storage, while expected to be present" + exit 1 + fi + + # curl on the archive url + get_archive_from_storage $url_2 + echo "http_status for $url_2 : $http_status" + if [ "$http_status" -ne "200" ]; then + echo "Archive $url_2 absent on storage, while expected to be present" + exit 1 + fi + + # archivePruner is set to run every minute for test. In production, its set to run every hour. + echo "waiting for packages to get recycled" + sleep 120 + + # curl on the archive url + get_archive_from_storage $url_1 + echo "http_status for $url_1 : $http_status" + if [ "$http_status" -ne "404" ]; then + echo "Archive $url_1 should have been recycled, but curl returned $http_status, while expected status is 404." + exit 1 + fi + + # curl on the archive url + get_archive_from_storage $url_2 + echo "http_status for $url_2 : $http_status" + if [ "$http_status" -ne "404" ]; then + echo "Archive $url_2 should have been recycled, but curl returned $http_status, while expected status is 404." + exit 1 + fi + + echo "Test archive pruner PASSED" +} + +main \ No newline at end of file