From f6e679e70e2a7bb9a766cd9ef512daba2fdb2aa9 Mon Sep 17 00:00:00 2001 From: Vishal Date: Mon, 15 Jun 2020 19:13:13 +0530 Subject: [PATCH] S3 backend for storage service (#1629) Co-authored-by: Alok Kumar --- charts/fission-all/templates/deployment.yaml | 26 ++- charts/fission-all/templates/pvc.yaml | 2 +- charts/fission-all/values.yaml | 14 +- charts/fission-core/templates/deployment.yaml | 30 ++- charts/fission-core/templates/pvc.yaml | 2 +- charts/fission-core/values.yaml | 14 +- cmd/fission-bundle/main.go | 24 ++- pkg/storagesvc/client/storagesvc_test.go | 172 ++++++++++++++++-- pkg/storagesvc/localstorage.go | 52 ++++++ pkg/storagesvc/s3storage.go | 67 +++++++ pkg/storagesvc/storage_test.go | 55 ++++++ pkg/storagesvc/storagesvc.go | 33 +++- pkg/storagesvc/stowClient.go | 42 ++--- 13 files changed, 472 insertions(+), 61 deletions(-) create mode 100644 pkg/storagesvc/localstorage.go create mode 100644 pkg/storagesvc/s3storage.go create mode 100644 pkg/storagesvc/storage_test.go diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index f91cd93e..b8e76175 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -776,7 +776,11 @@ spec: image: {{ include "fission-bundleImage" . | quote }} imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--storageServicePort", "8000", "--filePath", "/fission"] + {{- if and (.Values.persistence.enabled) (eq (.Values.persistence.storageType | default "local") "s3") }} + args: ["--storageServicePort", "8000", "--storageType", "s3"] + {{- else }} + args: ["--storageServicePort", "8000", "--storageType", "local"] + {{- end }} env: - name: TRACE_JAEGER_COLLECTOR_ENDPOINT value: "{{ .Values.traceCollectorEndpoint }}" @@ -786,9 +790,25 @@ spec: value: "{{.Values.pruneInterval}}" - name: DEBUG_ENV value: {{ .Values.debugEnv | quote }} + {{- if and (.Values.persistence.enabled) (eq (.Values.persistence.storageType | default "local") "s3") }} + - name: STORAGE_S3_ENDPOINT + value: {{ .Values.persistence.s3.endPoint }} + - name: STORAGE_S3_BUCKET_NAME + value: {{ .Values.persistence.s3.bucketName }} + - name: STORAGE_S3_SUB_DIR + value: {{ .Values.persistence.s3.subDir }} + - name: STORAGE_S3_ACCESS_KEY_ID + value: {{ .Values.persistence.s3.accessKeyId }} + - name: STORAGE_S3_SECRET_ACCESS_KEY + value: {{ .Values.persistence.s3.secretAccessKey }} + - name: STORAGE_S3_REGION + value: {{ .Values.persistence.s3.region }} + {{- end }} + {{- if ne (.Values.persistence.storageType | default "local") "s3" }} volumeMounts: - name: fission-storage mountPath: /fission + {{- end }} readinessProbe: httpGet: path: "/healthz" @@ -806,12 +826,14 @@ spec: - containerPort: 8000 name: http serviceAccountName: fission-svc + {{- if and (.Values.persistence.enabled) (ne (.Values.persistence.storageType | default "local") "s3") }} volumes: - name: fission-storage - {{- if .Values.persistence.enabled }} persistentVolumeClaim: claimName: {{ .Values.persistence.existingClaim | default "fission-storage-pvc" }} {{- else }} + volumes: + - name: fission-storage emptyDir: {} {{- end }} {{- if .Values.extraCoreComponentPodConfig }} diff --git a/charts/fission-all/templates/pvc.yaml b/charts/fission-all/templates/pvc.yaml index 7819c1db..9551f833 100644 --- a/charts/fission-all/templates/pvc.yaml +++ b/charts/fission-all/templates/pvc.yaml @@ -1,4 +1,4 @@ -{{- if and .Values.persistence.enabled (not .Values.persistence.existingClaim) }} +{{- if and (.Values.persistence.enabled) (not .Values.persistence.existingClaim) (eq (.Values.persistence.storageType | default "local") "local") }} kind: PersistentVolumeClaim apiVersion: v1 metadata: diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index e907c567..5c3dc45f 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -203,11 +203,23 @@ kafka: ## Persist data to a persistent volume. persistence: - ## If true, fission will create/use a Persistent Volume Claim + ## If true, fission will create/use a Persistent Volume Claim unless storageType is set to s3 ## If false, use emptyDir ## enabled: true + ## Must be set to either local or S3. + ## If storateType is set(other than local), one of its backend configuration must be set as below. + #storageType: local | s3 + + ## Sample configruation for AWS s3 storage backend + #s3: + # bucketName: + # subDir: + # accessKeyId: + # secretAccessKey: + # region: + ## A manually managed Persistent Volume Claim name ## Requires persistence.enabled: true ## If defined, PVC must be created manually before volume will be bound diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index a389d4ea..ff0d7e3b 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -288,7 +288,7 @@ spec: - name: TRACE_JAEGER_COLLECTOR_ENDPOINT value: "{{ .Values.traceCollectorEndpoint }}" - name: TRACING_SAMPLING_RATE - value: {{ .Values.traceSamplingRate | default "0.5" | quote }} + value: {{ .Values.traceSamplingRate | default "0.5" | quote }} - name: ENABLE_ISTIO value: "{{ .Values.enableIstio }}" - name: FETCHER_MINCPU @@ -398,27 +398,49 @@ spec: image: {{ include "fission-bundleImage" . | quote }} imagePullPolicy: {{ .Values.pullPolicy }} command: ["/fission-bundle"] - args: ["--storageServicePort", "8000", "--filePath", "/fission"] + {{- if and (.Values.persistence.enabled) (eq (.Values.persistence.storageType | default "local") "s3") }} + args: ["--storageServicePort", "8000", "--storageType", "s3"] + {{- else }} + args: ["--storageServicePort", "8000", "--storageType", "local"] + {{- end }} env: - name: PRUNE_INTERVAL value: "{{.Values.pruneInterval}}" - name: TRACE_JAEGER_COLLECTOR_ENDPOINT value: "{{ .Values.traceCollectorEndpoint }}" - name: TRACING_SAMPLING_RATE - value: {{ .Values.traceSamplingRate | default "0.5" | quote }} + value: {{ .Values.traceSamplingRate | default "0.5" | quote }} + {{- if and (.Values.persistence.enabled) (eq (.Values.persistence.storageType | default "local") "s3") }} + - name: STORAGE_S3_ENDPOINT + value: {{ .Values.persistence.s3.endPoint }} + - name: STORAGE_S3_BUCKET_NAME + value: {{ .Values.persistence.s3.bucketName }} + - name: STORAGE_S3_SUB_DIR + value: {{ .Values.persistence.s3.subDir }} + - name: STORAGE_S3_ACCESS_KEY_ID + value: {{ .Values.persistence.s3.accessKeyId }} + - name: STORAGE_S3_SECRET_ACCESS_KEY + value: {{ .Values.persistence.s3.secretAccessKey }} + - name: STORAGE_S3_REGION + value: {{ .Values.persistence.s3.region }} + {{- end }} + {{- if ne (.Values.persistence.storageType | default "local") "s3" }} volumeMounts: - name: fission-storage mountPath: /fission + {{- end }} ports: - containerPort: 8000 name: http serviceAccountName: fission-svc + {{- if and (.Values.persistence.enabled) (ne (.Values.persistence.storageType | default "local") "s3") }} volumes: - name: fission-storage - {{- if .Values.persistence.enabled }} persistentVolumeClaim: claimName: {{ .Values.persistence.existingClaim | default "fission-storage-pvc" }} {{- else }} + volumes: + - name: fission-storage emptyDir: {} {{- end }} {{- if .Values.extraCoreComponentPodConfig }} diff --git a/charts/fission-core/templates/pvc.yaml b/charts/fission-core/templates/pvc.yaml index 7819c1db..9551f833 100644 --- a/charts/fission-core/templates/pvc.yaml +++ b/charts/fission-core/templates/pvc.yaml @@ -1,4 +1,4 @@ -{{- if and .Values.persistence.enabled (not .Values.persistence.existingClaim) }} +{{- if and (.Values.persistence.enabled) (not .Values.persistence.existingClaim) (eq (.Values.persistence.storageType | default "local") "local") }} kind: PersistentVolumeClaim apiVersion: v1 metadata: diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index 04174f7e..705cb917 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -110,11 +110,23 @@ router: ## Persist data to a persistent volume. persistence: - ## If true, fission will create/use a Persistent Volume Claim + ## If true, fission will create/use a Persistent Volume Claim unless storageType is set to s3 ## If false, use emptyDir ## enabled: true + ## Must be set to either local or S3. + ## If storateType is set(other than local), one of its backend configuration must be set as below. + #storageType: local | s3 + + ## Sample configruation for AWS s3 storage backend + #s3: + # bucketName: + # subDir: + # accessKeyId: + # secretAccessKey: + # region: + ## A manually managed Persistent Volume Claim name ## Requires persistence.enabled: true ## If defined, PVC must be created manually before volume will be bound diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index 52b3ccda..e1949734 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -78,14 +78,11 @@ func runMessageQueueMgr(logger *zap.Logger, routerUrl string) { } } -func runStorageSvc(logger *zap.Logger, port int, filePath string) { - subdir := os.Getenv("SUBDIR") - if len(subdir) == 0 { - subdir = "fission-functions" +func runStorageSvc(logger *zap.Logger, port int, storage storagesvc.Storage) { + err := storagesvc.Start(logger, storage, port) + if err != nil { + logger.Fatal("error starting storage service", zap.Error(err)) } - enableArchivePruner := true - storagesvc.RunStorageService(logger, storagesvc.StorageTypeLocal, - filePath, subdir, port, enableArchivePruner) } func runBuilderMgr(logger *zap.Logger, storageSvcUrl string, envBuilderNamespace string) { @@ -200,7 +197,7 @@ Usage: fission-bundle --routerPort= [--executorUrl=] fission-bundle --executorPort= [--namespace=] [--fission-namespace=] fission-bundle --kubewatcher [--routerUrl=] - fission-bundle --storageServicePort= --filePath= + fission-bundle --storageServicePort= --storageType= fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] @@ -295,8 +292,15 @@ Options: if arguments["--storageServicePort"] != nil { port := getPort(logger, arguments["--storageServicePort"]) - filePath := arguments["--filePath"].(string) - runStorageSvc(logger, port, filePath) + + var storage storagesvc.Storage + + if arguments["--storageType"] != nil && arguments["--storageType"] == string(storagesvc.StorageTypeS3) { + storage = storagesvc.NewS3Storage() + } else if arguments["--storageType"] == string(storagesvc.StorageTypeLocal) { + storage = storagesvc.NewLocalStorage("/fission") + } + runStorageSvc(logger, port, storage) } select {} diff --git a/pkg/storagesvc/client/storagesvc_test.go b/pkg/storagesvc/client/storagesvc_test.go index ba925e0c..f49880e6 100644 --- a/pkg/storagesvc/client/storagesvc_test.go +++ b/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)) } diff --git a/pkg/storagesvc/localstorage.go b/pkg/storagesvc/localstorage.go new file mode 100644 index 00000000..484eaeee --- /dev/null +++ b/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) +} diff --git a/pkg/storagesvc/s3storage.go b/pkg/storagesvc/s3storage.go new file mode 100644 index 00000000..560ed155 --- /dev/null +++ b/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) +} diff --git a/pkg/storagesvc/storage_test.go b/pkg/storagesvc/storage_test.go new file mode 100644 index 00000000..84aca6eb --- /dev/null +++ b/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) + } +} diff --git a/pkg/storagesvc/storagesvc.go b/pkg/storagesvc/storagesvc.go index d1422143..200b4946 100644 --- a/pkg/storagesvc/storagesvc.go +++ b/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 } diff --git a/pkg/storagesvc/stowClient.go b/pkg/storagesvc/stowClient.go index 2110be7e..775555dc 100644 --- a/pkg/storagesvc/stowClient.go +++ b/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)