From 31ba992726771a2cfb1eef226d1af1020979b506 Mon Sep 17 00:00:00 2001 From: smruthi2187 <34555664+smruthi2187@users.noreply.github.com> Date: Thu, 1 Feb 2018 02:13:36 -0800 Subject: [PATCH] Archive pruner (#471) All functions have a pkg reference. This can be a package with either source and a deploy archives, or, a deploy archive. Everytime a function is updated, a new package is created. With archive pruner, the archives that are pointed to by old pkg reference can be deleted from the storage. * High level spec for package pruning. * Skeleton for archive pruning * Adding meat 1 to skeleton. * Adding meat #2. Separated storage service into a httpHandler component and Storage Layer component. * Adding meat #3. getOrphanedArchives in pruner and getItems on stowClient. * Restructured archivePruner methods. * Commiting the day's work. Ready for testing #1. * Fixing compile errors. * Test ready. added a few logs for debugging. * Adding a filter for getItems in stowClient. * After testing. * Added a test for archivePruner. * Adding helm value pruneInterval for testing. * Modified test. * Final test. * Fixing interval from seconds to minutes. * Small change. * Changing debugs to info. * Removing the WIP design * Ran gofmt on all these files. * Fixing prune_interval as string in ENV var. * Addressing all comments, but one. * changing getFile method in stowClient to stream it into a response. * All comments incorporated. * Introducing a new flag for running archivePruner. 1. This flag is disabled for archivePruner to run in unit test. 2. This flag is enabled for archivePruner to run in production. 3. Also disabling test_archive_pruner.sh in this PR. Follow up with next PR to enable it. * Addressing review comments. * Changing the command to generate a file dynamically. * Enabling arching_pruner_test * giving execute permissions to test_archive_pruner.sh * Making changes of positional parameters after recent commit. Change test case permission and removing kubectlPortForward. * Adding debug to see why test_utils.sh passed junk pruneInterval. * shell needs special handling for positional parameters from 10. --- charts/fission-all/templates/deployment.yaml | 3 + charts/fission-all/values.yaml | 5 + charts/fission-core/templates/deployment.yaml | 3 + charts/fission-core/values.yaml | 5 + fission-bundle/main.go | 3 +- storagesvc/README.md | 26 +++ storagesvc/archivePruner.go | 146 +++++++++++++ storagesvc/client/storagesvc_test.go | 9 +- storagesvc/storagesvc.go | 154 +++++--------- storagesvc/stowClient.go | 201 ++++++++++++++++++ storagesvc/util.go | 53 +++++ test/build_and_test.sh | 3 +- test/test_utils.sh | 6 +- test/tests/test_archive_pruner.sh | 117 ++++++++++ 14 files changed, 622 insertions(+), 112 deletions(-) create mode 100644 storagesvc/README.md create mode 100644 storagesvc/archivePruner.go create mode 100644 storagesvc/stowClient.go create mode 100644 storagesvc/util.go create mode 100755 test/tests/test_archive_pruner.sh 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