S3 backend for storage service (#1629)

Co-authored-by: Alok Kumar <rajalokan@gmail.com>

pkg/storagesvc/client/storagesvc_test.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"os"
@@ -27,9 +28,17 @@ import (
 	"time"
 
 	"github.com/dchest/uniuri"
-	"go.uber.org/zap"
-
 	"github.com/fission/fission/pkg/storagesvc"
+	"github.com/minio/minio-go/v6"
+	"github.com/ory/dockertest"
+	dc "github.com/ory/dockertest/docker"
+	"go.uber.org/zap"
+)
+
+const (
+	minioAccessKeyID     = "minioadmin"
+	minioSecretAccessKey = "minioadmin"
+	minioRegion          = "ap-south-1"
 )
 
 func panicIf(err error) {
@@ -48,17 +57,152 @@ func MakeTestFile(size int) *os.File {
 	return f
 }
 
-func TestStorageService(t *testing.T) {
-	testId := uniuri.NewLen(8)
-	port := 8080
-	enableArchivePruner := false
+func runMinioDockerContainer(pool *dockertest.Pool) *dockertest.Resource {
+	options := &dockertest.RunOptions{
+		Repository: "minio/minio",
+		Tag:        "latest",
+		Cmd:        []string{"server", "/data"},
+		PortBindings: map[dc.Port][]dc.PortBinding{
+			"9000/tcp": {{HostIP: "", HostPort: "9000"}},
+		},
+	}
+
+	// pulls an image, creates a container based on it and runs it
+	resource, err := pool.RunWithOptions(options)
+	if err != nil {
+		log.Fatalf("Could not start resource: %s", err)
+	}
+	return resource
+}
+
+func startS3StorageService(endpoint, bucketName, subDir string) {
+	// testID := uniuri.NewLen(8)
+	port := 8081
 
 	logger, err := zap.NewDevelopment()
 	panicIf(err)
 
 	log.Println("starting storage svc")
-	_ = storagesvc.RunStorageService(
-		logger, storagesvc.StorageTypeLocal, "/tmp", testId, port, enableArchivePruner)
+	os.Setenv("STORAGE_S3_ENDPOINT", endpoint)
+	os.Setenv("STORAGE_S3_BUCKET_NAME", bucketName)
+	os.Setenv("STORAGE_S3_SUB_DIR", subDir)
+	os.Setenv("STORAGE_S3_ACCESS_KEY_ID", minioAccessKeyID)
+	os.Setenv("STORAGE_S3_SECRET_ACCESS_KEY", minioSecretAccessKey)
+	os.Setenv("STORAGE_S3_REGION", minioRegion)
+
+	storage := storagesvc.NewS3Storage()
+	_ = storagesvc.Start(logger, storage, port)
+
+}
+
+func TestS3StorageService(t *testing.T) {
+	fmt.Println("Test S3 Storage service")
+	var minioClient *minio.Client
+
+	// Start minio docker container
+	pool, err := dockertest.NewPool("")
+	resource := runMinioDockerContainer(pool)
+
+	endpoint := fmt.Sprintf("localhost:%s", resource.GetPort("9000/tcp"))
+
+	if err := pool.Retry(func() error {
+		minioClient, err = minio.New(endpoint, minioAccessKeyID, minioSecretAccessKey, false)
+		if err != nil {
+			return err
+		}
+
+		// This is to ensure container is up. Just getting minioClient
+		// isn't suffcient to assume container is up.
+		_, err = minioClient.ListBuckets()
+		if err != nil {
+			return err
+		}
+
+		return nil
+	}); err != nil {
+		log.Fatalf("Could not connect to docker: %s", err)
+	}
+
+	defer pool.Purge(resource)
+
+	// Start storagesvc
+	bucketName := "test-s3-service"
+	subDir := "x/y/z"
+	startS3StorageService(endpoint, bucketName, subDir)
+
+	time.Sleep(time.Second)
+	client := MakeClient(fmt.Sprintf("http://localhost:%v/", 8081))
+
+	// generate a test file
+	tmpfile := MakeTestFile(10 * 1024)
+	defer os.Remove(tmpfile.Name())
+
+	// store it
+	metadata := make(map[string]string)
+	ctx := context.Background()
+	fileID, err := client.Upload(ctx, tmpfile.Name(), &metadata)
+	panicIf(err)
+
+	time.Sleep(10 * time.Second)
+
+	// Retrive file trhough minioClient
+	reader, err := minioClient.GetObject(bucketName, fileID, minio.GetObjectOptions{})
+	panicIf(err)
+	defer reader.Close()
+
+	retThroughMinio, err := ioutil.TempFile("", "storagesvc_verify_minio_")
+	panicIf(err)
+	defer os.Remove(retThroughMinio.Name())
+
+	stat, err := reader.Stat()
+	panicIf(err)
+
+	if _, err := io.CopyN(retThroughMinio, reader, stat.Size); err != nil {
+		log.Fatalln(err)
+	}
+
+	// Retrieve file through API
+	retThroughAPI, err := ioutil.TempFile("", "storagesvc_verify_")
+	panicIf(err)
+	os.Remove(retThroughAPI.Name())
+
+	err = client.Download(ctx, fileID, retThroughAPI.Name())
+	panicIf(err)
+	defer os.Remove(retThroughAPI.Name())
+
+	// compare contents
+	contentsMinio, err := ioutil.ReadFile(retThroughMinio.Name())
+	panicIf(err)
+	contentsAPI, err := ioutil.ReadFile(retThroughAPI.Name())
+	panicIf(err)
+	if !bytes.Equal(contentsMinio, contentsAPI) {
+		log.Panic("Contents don't match")
+	}
+
+	// delete uploaded file
+	err = client.Delete(ctx, fileID)
+	panicIf(err)
+
+	// make sure download fails
+	err = client.Download(ctx, fileID, "xxx")
+	if err == nil {
+		log.Panic("Download succeeded but file isn't supposed to exist")
+	}
+
+}
+
+func TestLocalStorageService(t *testing.T) {
+	testID := uniuri.NewLen(8)
+	port := 8080
+
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	log.Println("starting storage svc")
+	localPath := fmt.Sprintf("/tmp/%v", testID)
+	_ = os.Mkdir(localPath, os.ModePerm)
+	storage := storagesvc.NewLocalStorage(localPath)
+	_ = storagesvc.Start(logger, storage, port)
 
 	time.Sleep(time.Second)
 	client := MakeClient(fmt.Sprintf("http://localhost:%v/", port))
@@ -70,7 +214,7 @@ func TestStorageService(t *testing.T) {
 	// store it
 	metadata := make(map[string]string)
 	ctx := context.Background()
-	fileId, err := client.Upload(ctx, tmpfile.Name(), &metadata)
+	fileID, err := client.Upload(ctx, tmpfile.Name(), &metadata)
 	panicIf(err)
 
 	// make a temp file for verification
@@ -79,7 +223,7 @@ func TestStorageService(t *testing.T) {
 	os.Remove(retrievedfile.Name())
 
 	// retrieve uploaded file
-	err = client.Download(ctx, fileId, retrievedfile.Name())
+	err = client.Download(ctx, fileID, retrievedfile.Name())
 	panicIf(err)
 	defer os.Remove(retrievedfile.Name())
 
@@ -93,15 +237,15 @@ func TestStorageService(t *testing.T) {
 	}
 
 	// delete uploaded file
-	err = client.Delete(ctx, fileId)
+	err = client.Delete(ctx, fileID)
 	panicIf(err)
 
 	// make sure download fails
-	err = client.Download(ctx, fileId, "xxx")
+	err = client.Download(ctx, fileID, "xxx")
 	if err == nil {
 		log.Panic("Download succeeded but file isn't supposed to exist")
 	}
 
-	// cleanup /tmp
-	os.RemoveAll(fmt.Sprintf("/tmp/%v", testId))
+	// // cleanup /tmp
+	os.RemoveAll(fmt.Sprintf("/tmp/%v", testID))
 }

pkg/storagesvc/localstorage.go

@@ -0,0 +1,52 @@
+package storagesvc
+
+import (
+	"os"
+
+	"github.com/graymeta/stow"
+	_ "github.com/graymeta/stow/local"
+	uuid "github.com/satori/go.uuid"
+)
+
+type localStorage struct {
+	storageType   StorageType
+	containerName string
+	localPath     string
+}
+
+// NewLocalStorage return new local storage struct
+func NewLocalStorage(localPath string) Storage {
+	subdir := os.Getenv("SUBDIR")
+	if len(subdir) == 0 {
+		subdir = "fission-functions"
+	}
+	return localStorage{
+		storageType:   StorageTypeLocal,
+		containerName: subdir,
+		localPath:     localPath,
+	}
+}
+
+// Local
+func (ls localStorage) getStorageType() StorageType {
+	return ls.storageType
+}
+
+func (ls localStorage) SetLocalPath(path string) {
+	ls.localPath = path
+}
+
+func (ls localStorage) getUploadFileName() string {
+	// This is not the item ID (that's returned by Put)
+	// should we just use handler.Filename? what are the constraints here?
+	return uuid.NewV4().String()
+}
+
+func (ls localStorage) getContainerName() string {
+	return ls.containerName
+}
+
+func (ls localStorage) dial() (stow.Location, error) {
+	cfg := stow.ConfigMap{"path": ls.localPath}
+	return stow.Dial("local", cfg)
+}

pkg/storagesvc/s3storage.go

@@ -0,0 +1,67 @@
+package storagesvc
+
+import (
+	"os"
+	"path"
+
+	"github.com/graymeta/stow"
+	"github.com/graymeta/stow/s3"
+	uuid "github.com/satori/go.uuid"
+)
+
+type (
+	s3Storage struct {
+		storageType     StorageType
+		endpoint        string
+		bucketName      string
+		subDir          string
+		accessKeyID     string
+		secretAccessKey string
+		region          string
+	}
+)
+
+// NewS3Storage returns a new s3 storage struct
+func NewS3Storage(args ...string) Storage {
+	endpoint := os.Getenv("STORAGE_S3_ENDPOINT")
+	bucketName := os.Getenv("STORAGE_S3_BUCKET_NAME")
+	subDir := os.Getenv("STORAGE_S3_SUB_DIR")
+	accessKeyID := os.Getenv("STORAGE_S3_ACCESS_KEY_ID")
+	secretAccessKey := os.Getenv("STORAGE_S3_SECRET_ACCESS_KEY")
+	region := os.Getenv("STORAGE_S3_REGION")
+
+	return s3Storage{
+		endpoint:        endpoint,
+		storageType:     StorageTypeS3,
+		bucketName:      bucketName,
+		subDir:          subDir,
+		accessKeyID:     accessKeyID,
+		secretAccessKey: secretAccessKey,
+		region:          region,
+	}
+}
+
+func (ss s3Storage) getStorageType() StorageType {
+	return ss.storageType
+}
+
+func (ss s3Storage) getContainerName() string {
+	return ss.bucketName
+}
+
+func (ss s3Storage) getUploadFileName() string {
+	uploadName := uuid.NewV4().String()
+	return path.Join(ss.subDir, uploadName)
+}
+
+func (ss s3Storage) dial() (stow.Location, error) {
+	kind := "s3"
+	config := stow.ConfigMap{
+		s3.ConfigEndpoint:    ss.endpoint,
+		s3.ConfigAccessKeyID: ss.accessKeyID,
+		s3.ConfigSecretKey:   ss.secretAccessKey,
+		s3.ConfigRegion:      ss.region,
+		s3.ConfigDisableSSL:  "true",
+	}
+	return stow.Dial(kind, config)
+}

pkg/storagesvc/storage_test.go

@@ -0,0 +1,55 @@
+package storagesvc
+
+import (
+	"os"
+	"reflect"
+	"testing"
+)
+
+func TestNewS3Storage(t *testing.T) {
+	input := map[string]string{
+		"bucketName":      "tmpBucket",
+		"subDir":          "a/b/c",
+		"accessKeyID":     "tmpAccessKeyID",
+		"secretAccessKey": "tmpSecretAccessKey",
+		"region":          "ap-south-1",
+	}
+
+	os.Setenv("STORAGE_S3_BUCKET_NAME", input["bucketName"])
+	os.Setenv("STORAGE_S3_SUB_DIR", input["subDir"])
+	os.Setenv("STORAGE_S3_ACCESS_KEY_ID", input["accessKeyID"])
+	os.Setenv("STORAGE_S3_SECRET_ACCESS_KEY", input["secretAccessKey"])
+	os.Setenv("STORAGE_S3_REGION", input["region"])
+
+	storage := NewS3Storage().(s3Storage)
+
+	for k, v := range input {
+		valueInStruct := reflect.Indirect(reflect.ValueOf(storage)).FieldByName(k).String()
+
+		if valueInStruct != v {
+			t.Errorf("Incorrect s3Storage field. Got: %s, Want %s", valueInStruct, v)
+		}
+	}
+
+	if storage.storageType != StorageTypeS3 {
+		t.Errorf("Incorrect storageType field. Got: %s, Want %s", storage.storageType, StorageTypeS3)
+	}
+
+	// TestGetStorageType
+	if storage.getStorageType() != storage.storageType {
+		t.Errorf("Incorrect getStorateType() method implementation. Got: %s, Want %s", storage.getStorageType(), storage.storageType)
+	}
+
+}
+
+func TestNewLocalStorage(t *testing.T) {
+	storage := NewLocalStorage("/fission").(localStorage)
+
+	// // When SUBDIR env is not set, expect a default "fission-functions" value.
+	// if storage.subDir != "fission-functions" {
+	// 	t.Errorf("Incorrect subDir field. Got: %s, Want %s", storage.subDir, "fission-functions")
+	// }
+	if storage.storageType != StorageTypeLocal {
+		t.Errorf("Incorrect storageType field. Got: %s, Want %s", storage.storageType, StorageTypeLocal)
+	}
+}

pkg/storagesvc/storagesvc.go

@@ -25,13 +25,23 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
-	_ "github.com/graymeta/stow/local"
+	"github.com/graymeta/stow"
 	"github.com/pkg/errors"
 	"go.opencensus.io/plugin/ochttp"
 	"go.uber.org/zap"
 )
 
 type (
+	// Storage is an interface to force storage level details implementation.
+	Storage interface {
+		getStorageType() StorageType
+		dial() (stow.Location, error)
+		// getSubDir() string
+		getContainerName() string
+		getUploadFileName() string
+	}
+
+	// StorageService is a struct to hold all things for storage service
 	StorageService struct {
 		logger        *zap.Logger
 		storageClient *StowClient
@@ -43,6 +53,15 @@ type (
 	}
 )
 
+// Functions handling storage interface
+func getStorageType(storage Storage) string {
+	return string(storage.getStorageType())
+}
+
+func getStorageLocation(config *storageConfig) (stow.Location, error) {
+	return config.storage.dial()
+}
+
 // Handle multipart file uploads.
 func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) {
 	// handle upload
@@ -185,11 +204,13 @@ func (ss *StorageService) Start(port int) {
 	ss.logger.Fatal("done listening", zap.Error(err))
 }
 
-func RunStorageService(logger *zap.Logger, storageType StorageType, storagePath string, containerName string, port int, enablePruner bool) *StorageService {
+// Start runs storage service
+func Start(logger *zap.Logger, storage Storage, port int) error {
+	enablePruner := true
 	// create a storage client
-	storageClient, err := MakeStowClient(logger, storageType, storagePath, containerName)
+	storageClient, err := MakeStowClient(logger, storage)
 	if err != nil {
-		logger.Fatal("error creating stowClient", zap.Error(err))
+		return errors.Wrap(err, "Error creating stowClient")
 	}
 
 	// create http handlers
@@ -205,11 +226,11 @@ func RunStorageService(logger *zap.Logger, storageType StorageType, storagePath
 		}
 		pruner, err := MakeArchivePruner(logger, storageClient, time.Duration(pruneInterval))
 		if err != nil {
-			logger.Fatal("error creating archivePruner", zap.Error(err))
+			return errors.Wrap(err, "Error creating archivePruner")
 		}
 		go pruner.Start()
 	}
 
 	logger.Info("storage service started")
-	return storageService
+	return nil
 }

pkg/storagesvc/stowClient.go

@@ -20,25 +20,23 @@ import (
 	"io"
 	"mime/multipart"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/graymeta/stow"
-	_ "github.com/graymeta/stow/local"
 	"github.com/pkg/errors"
-	"github.com/satori/go.uuid"
 	"go.uber.org/zap"
 )
 
 type (
+	// StorageType contains all different types of supported storage
 	StorageType string
 
 	storageConfig struct {
-		storageType   StorageType
-		localPath     string
-		containerName string
-		// other stuff, such as google or s3 credentials, bucket names etc
+		storage Storage
 	}
 
+	//StowClient is the wraper client for stow (Cloud storage abstraction package)
 	StowClient struct {
 		logger    *zap.Logger
 		config    *storageConfig
@@ -48,8 +46,12 @@ type (
 )
 
 const (
+	// StorageTypeLocal is a constant to hold local storate type name literal
 	StorageTypeLocal StorageType = "local"
-	PaginationSize   int         = 10
+	// StorageTypeS3 is a constant to hold S3 storage type name literal
+	StorageTypeS3 StorageType = "s3"
+	// PaginationSize is a constant to hold no of pages
+	PaginationSize int = 10
 )
 
 var (
@@ -60,15 +62,15 @@ var (
 	ErrWritingFileIntoResponse = errors.New("unable to copy item into http response")
 )
 
-func MakeStowClient(logger *zap.Logger, storageType StorageType, storagePath string, containerName string) (*StowClient, error) {
-	if storageType != StorageTypeLocal {
-		return nil, errors.New("Storage types other than 'local' are not implemented")
+// MakeStowClient create a new StowClient for given storage
+func MakeStowClient(logger *zap.Logger, storage Storage) (*StowClient, error) {
+	storageType := getStorageType(storage)
+	if strings.Compare(storageType, "local") == 1 && strings.Compare(storageType, "s3") == 1 {
+		return nil, errors.New("Storage types other than 'local' and 's3' are not implemented")
 	}
 
 	config := &storageConfig{
-		storageType:   storageType,
-		localPath:     storagePath,
-		containerName: containerName,
+		storage: storage,
 	}
 
 	stowClient := &StowClient{
@@ -76,21 +78,21 @@ func MakeStowClient(logger *zap.Logger, storageType StorageType, storagePath str
 		config: config,
 	}
 
-	cfg := stow.ConfigMap{"path": config.localPath}
-	loc, err := stow.Dial("local", cfg)
+	loc, err := getStorageLocation(config)
 	if err != nil {
 		return nil, err
 	}
 	stowClient.location = loc
 
-	con, err := loc.CreateContainer(config.containerName)
-	if os.IsExist(err) {
+	con, err := loc.CreateContainer(config.storage.getContainerName())
+	if err != nil && (os.IsExist(err) || strings.Contains(err.Error(), "BucketAlreadyOwnedByYou")) {
 		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)
+		cons, cursor, err = loc.Containers(config.storage.getContainerName(), stow.CursorStart, 1)
 		if err == nil {
+			con = cons[0]
 			if !stow.IsCursorEnd(cursor) {
 				// Should only have one storage container
 				err = errors.New("Found more than one matched storage containers")
@@ -109,9 +111,7 @@ func MakeStowClient(logger *zap.Logger, storageType StorageType, storagePath str
 
 // 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()
+	uploadName := client.config.storage.getUploadFileName()
 
 	// save the file to the storage backend
 	item, err := client.container.Put(uploadName, file, int64(fileSize), nil)