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

@@ -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

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

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

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

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) {

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.
+
+
+

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()
+		}
+	}
+}

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

storagesvc/storagesvc.go

@@ -20,32 +20,20 @@ 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
+		storageClient *StowClient
 		port          int
 	}
 
@@ -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)
-		}
-		return
-	}
-
-	f, err := item.Open()
-	if err != nil {
-		log.Printf("Error opening item %v: %v", fileId, err)
-		// TODO better http errors based on err
+		} else if err == ErrOpeningItem {
 			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)
+		} else if err == ErrWritingFileIntoResponse {
 			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
 }

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
+}

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
+}

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

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

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