(PPS-411): Add bucket selection support for multi part upload

gen3-client/commonUtils/commonUtils.go

@@ -74,6 +74,7 @@ type FileUploadRequestObject struct {
 	PresignedURL string
 	Request      *http.Request
 	Bar          *pb.ProgressBar
+	Bucket 	 	 string `json:"bucket,omitempty"`
 }
 
 // FileDownloadResponseObject defines a object for file download
@@ -105,6 +106,7 @@ type RetryObject struct {
 	GUID         string
 	RetryCount   int
 	Multipart    bool
+	Bucket 		 string
 }
 
 // ParseRootPath parses dirname that has "~" in the beginning

gen3-client/g3cmd/retry-upload.go

@@ -53,6 +53,7 @@ func retryUpload(failedLogMap map[string]commonUtils.RetryObject) {
 	var guid string
 	var presignedURL string
 	var err error
+
 	fmt.Println()
 	if len(failedLogMap) == 0 {
 		log.Println("No failed file in log, no need to retry upload.")
@@ -96,7 +97,7 @@ func retryUpload(failedLogMap map[string]commonUtils.RetryObject) {
 
 		if ro.Multipart {
 			fileInfo := FileInfo{FilePath: ro.FilePath, Filename: ro.Filename}
-			err = multipartUpload(gen3Interface, fileInfo, ro.RetryCount)
+			err = multipartUpload(gen3Interface, fileInfo, ro.RetryCount, ro.Bucket)
 			if err != nil {
 				updateRetryObject(&ro, ro.FilePath, ro.Filename, ro.FileMetadata, ro.GUID, ro.RetryCount, true)
 				handleFailedRetry(ro, retryObjCh, err, true)
@@ -109,7 +110,7 @@ func retryUpload(failedLogMap map[string]commonUtils.RetryObject) {
 				}
 			}
 		} else {
-			presignedURL, guid, err = GeneratePresignedURL(gen3Interface, ro.Filename, ro.FileMetadata)
+			presignedURL, guid, err = GeneratePresignedURL(gen3Interface, ro.Filename, ro.FileMetadata, ro.Bucket)
 			if err != nil {
 				updateRetryObject(&ro, ro.FilePath, ro.Filename, ro.FileMetadata, guid, ro.RetryCount, false)
 				handleFailedRetry(ro, retryObjCh, err, true)

gen3-client/g3cmd/upload-multipart.go

@@ -37,7 +37,7 @@ func retry(attempts int, filePath string, guid string, f func() error) (err erro
 	return fmt.Errorf("After %d attempts, last error: %s", attempts, err)
 }
 
-func multipartUpload(g3 Gen3Interface, fileInfo FileInfo, retryCount int) error {
+func multipartUpload(g3 Gen3Interface, fileInfo FileInfo, retryCount int, bucketName string) error {
 	// NOTE @mpingram -- multipartUpload does not yet use the new Shepherd API
 	// because Shepherd does not yet support multipart uploads.
 	file, err := os.Open(fileInfo.FilePath)
@@ -61,7 +61,7 @@ func multipartUpload(g3 Gen3Interface, fileInfo FileInfo, retryCount int) error
 		return err
 	}
 
-	uploadID, guid, err := InitMultipartUpload(g3, fileInfo.Filename)
+	uploadID, guid, err := InitMultipartUpload(g3, fileInfo.Filename, bucketName)
 	if err != nil {
 		logs.AddToFailedLog(fileInfo.FilePath, fileInfo.Filename, fileInfo.FileMetadata, guid, retryCount, true, true)
 		err = fmt.Errorf("FAILED multipart upload for %s: %s", fileInfo.Filename, err.Error())
@@ -85,7 +85,7 @@ func multipartUpload(g3 Gen3Interface, fileInfo FileInfo, retryCount int) error
 			for chunkIndex := range chunkIndexCh {
 				var presignedURL string
 				err = retry(MaxRetryCount, fileInfo.FilePath, guid, func() (err error) {
-					presignedURL, err = GenerateMultipartPresignedURL(g3, key, uploadID, chunkIndex)
+					presignedURL, err = GenerateMultipartPresignedURL(g3, key, uploadID, chunkIndex, bucketName)
 					return
 				})
 				if err != nil {
@@ -165,7 +165,7 @@ func multipartUpload(g3 Gen3Interface, fileInfo FileInfo, retryCount int) error
 		return parts[i].PartNumber < parts[j].PartNumber // sort parts in ascending order
 	})
 
-	if err = CompleteMultipartUpload(g3, key, uploadID, parts); err != nil {
+	if err = CompleteMultipartUpload(g3, key, uploadID, parts, bucketName); err != nil {
 		logs.AddToFailedLog(fileInfo.FilePath, fileInfo.Filename, fileInfo.FileMetadata, guid, retryCount, true, true)
 		err = fmt.Errorf("FAILED multipart upload for %s: %s", fileInfo.Filename, err.Error())
 		return err

gen3-client/g3cmd/upload-multiple.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"net/http"
 	"os"
 	"path/filepath"
 	"strings"
@@ -17,20 +16,19 @@ import (
 )
 
 func init() {
+	var bucketName string
 	var manifestPath string
 	var uploadPath string
 	var batch bool
 	var numParallel int
-	var workers int
-	var respCh chan *http.Response
-	var errCh chan error
-	var batchFURObjects []commonUtils.FileUploadRequestObject
+	var forceMultipart bool
+	var includeSubDirName bool
 
 	var uploadMultipleCmd = &cobra.Command{
 		Use:     "upload-multiple",
 		Short:   "Upload multiple of files from a specified manifest",
-		Long:    `Get presigned URLs for multiple of files specified in a manifest file and then upload all of them.`,
-		Example: `./gen3-client upload-multiple --profile=<profile-name> --manifest=<path-to-manifest/manifest.json> --upload-path=<path-to-file-dir/>`,
+		Long:    `Get presigned URLs for multiple of files specified in a manifest file and then upload all of them. Options to run multipart uploads for large files and running multiple workers to batch upload available.`,
+		Example: `./gen3-client upload-multiple --profile=<profile-name> --manifest=<path-to-manifest/manifest.json> --upload-path=<path-to-file-dir/> --bucket=<bucket-name> --force-multipart=<boolean> --include-subdirname=<boolean> --batch=<boolean>`,
 		Run: func(cmd *cobra.Command, args []string) {
 			fmt.Printf("Notice: this is the upload method which requires the user to provide GUIDs. In this method files will be uploaded to specified GUIDs.\nIf your intention is to upload files without pre-existing GUIDs, consider to use \"./gen3-client upload\" instead.\n\n")
 
@@ -76,51 +74,54 @@ func init() {
 				log.Fatalln("A valid manifest can be acquired by using the \"Download Manifest\" button on " + dataExplorerURL)
 			}
 
-			furObjects := validateObject(objects, uploadPath)
+			uploadPath, err := commonUtils.GetAbsolutePath(uploadPath)
+			if err != nil {
+				log.Fatalf("Error when parsing file paths: " + err.Error())
+			}
 
-			if batch {
-				workers, respCh, errCh, batchFURObjects = initBatchUploadChannels(numParallel, len(objects))
+			filePaths := make([]string, 0)
+			for _, object := range objects {
+				// Here we are assuming the local filename will be the same as GUID
+				filePath, err := getFullFilePath(uploadPath, object.ObjectID)
+				if err != nil {
+					log.Println(err.Error())
+					continue
+				}
+				filePaths = append(filePaths, filePath)
 			}
 
-			for i, furObject := range furObjects {
-				if batch {
-					if len(batchFURObjects) < workers {
-						batchFURObjects = append(batchFURObjects, furObject)
-					} else {
-						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh)
-						batchFURObjects = make([]commonUtils.FileUploadRequestObject, 0)
-						batchFURObjects = append(batchFURObjects, furObject)
-					}
-					if i == len(furObjects)-1 { // upload remainders
-						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh)
-					}
-				} else {
-					file, err := os.Open(furObject.FilePath)
-					if err != nil {
-						log.Println("File open error: " + err.Error())
-						logs.AddToFailedLog(furObject.FilePath, furObject.Filename, commonUtils.FileMetadata{}, furObject.GUID, 0, false, true)
-						logs.IncrementScore(logs.ScoreBoardLen - 1)
-						continue
-					}
-					defer file.Close()
+			singlePartFilePaths, multipartFilePaths := separateSingleAndMultipartUploads(filePaths, forceMultipart)
 
-					furObject, err := GenerateUploadRequest(gen3Interface, furObject, file)
+			if batch {
+				workers, respCh, errCh, batchFURObjects := initBatchUploadChannels(numParallel, len(singlePartFilePaths))
+				for i, filePath := range singlePartFilePaths {
+					fileInfo, err := ProcessFilename(uploadPath, filePath, includeSubDirName, false)
 					if err != nil {
-						file.Close()
-						logs.AddToFailedLog(furObject.FilePath, furObject.Filename, commonUtils.FileMetadata{}, furObject.GUID, 0, false, true)
-						logs.IncrementScore(logs.ScoreBoardLen - 1)
-						log.Printf("Error occurred during request generation: %s", err.Error())
-						continue
+						logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
+						log.Println("Process filename error: " + err.Error())
+						return
 					}
-					err = uploadFile(furObject, 0)
-					if err != nil {
-						log.Println(err.Error())
-						logs.IncrementScore(logs.ScoreBoardLen - 1)
+					if len(batchFURObjects) < workers {
+						furObject := commonUtils.FileUploadRequestObject{FilePath: fileInfo.FilePath, Filename: fileInfo.Filename, FileMetadata: fileInfo.FileMetadata, GUID: ""}
+						batchFURObjects = append(batchFURObjects, furObject) //nolint:ineffassign
 					} else {
-						logs.IncrementScore(0)
+						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh, bucketName)
+						batchFURObjects = make([]commonUtils.FileUploadRequestObject, 0)
+						furObject := commonUtils.FileUploadRequestObject{FilePath: fileInfo.FilePath, Filename: fileInfo.Filename, FileMetadata: fileInfo.FileMetadata, GUID: ""}
+						batchFURObjects = append(batchFURObjects, furObject) //nolint:ineffassign
+					}
+					if !forceMultipart && i == len(singlePartFilePaths)-1 { // upload remainders
+						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh, bucketName)
 					}
-					file.Close()
 				}
+			} else {
+				processSingleUploads(gen3Interface, singlePartFilePaths, bucketName, includeSubDirName, uploadPath)
+			}
+			if len(multipartFilePaths) > 0 {
+				processMultipartUpload(gen3Interface, multipartFilePaths, bucketName, includeSubDirName, uploadPath)
+			}
+			if !logs.IsFailedLogMapEmpty() {
+				retryUpload(logs.GetFailedLogMap())
 			}
 			logs.PrintScoreBoard()
 			logs.CloseAll()
@@ -135,5 +136,87 @@ func init() {
 	uploadMultipleCmd.MarkFlagRequired("upload-path") //nolint:errcheck
 	uploadMultipleCmd.Flags().BoolVar(&batch, "batch", true, "Upload in parallel")
 	uploadMultipleCmd.Flags().IntVar(&numParallel, "numparallel", 3, "Number of uploads to run in parallel")
+	uploadMultipleCmd.Flags().StringVar(&bucketName, "bucket", "", "The bucket to which files will be uploaded. If not provided, defaults to Gen3's configured DATA_UPLOAD_BUCKET.")
+	uploadMultipleCmd.Flags().BoolVar(&forceMultipart, "force-multipart", false, "Force to use multipart upload when possible (file size >= 5MB)")
+	uploadMultipleCmd.Flags().BoolVar(&includeSubDirName, "include-subdirname", false, "Include subdirectory names in file name")
 	RootCmd.AddCommand(uploadMultipleCmd)
 }
+
+func processSingleUploads(gen3Interface Gen3Interface, singleFilePaths []string, bucketName string, includeSubDirName bool, uploadPath string) {
+	for _, filePath := range singleFilePaths {
+		file, err := os.Open(filePath)
+		if err != nil {
+			logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
+			log.Println("File open error: " + err.Error())
+			continue
+		}
+
+		startSingleFileUpload(gen3Interface, filePath, file, bucketName, includeSubDirName, uploadPath)
+		file.Close()
+	}
+}
+
+func startSingleFileUpload(gen3Interface Gen3Interface, filePath string, file *os.File, bucketName string, includeSubDirName bool, uploadPath string) {
+	fi, err := file.Stat()
+	if err != nil {
+		logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
+		log.Println("File stat error for file" + fi.Name() + ", file may be missing or unreadable because of permissions.\n")
+		return
+	}
+
+	fileInfo, err := ProcessFilename(uploadPath, filePath, includeSubDirName, false)
+	if err != nil {
+		logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
+		log.Println("Process filename error for file: " + err.Error())
+		return
+	}
+
+	respURL, guid, err := GeneratePresignedURL(gen3Interface, fileInfo.Filename, fileInfo.FileMetadata, bucketName)
+	if err != nil {
+		logs.AddToFailedLog(fileInfo.FilePath, fileInfo.Filename, fileInfo.FileMetadata, guid, 0, false, true)
+		log.Println(err.Error())
+		return
+	}
+
+	logs.AddToFailedLog(fileInfo.FilePath, fileInfo.Filename, fileInfo.FileMetadata, guid, 0, false, true)
+
+	furObject := commonUtils.FileUploadRequestObject{FilePath: fileInfo.FilePath, Filename: fileInfo.Filename, GUID: guid, PresignedURL: respURL}
+	furObject, err = GenerateUploadRequest(gen3Interface, furObject, file)
+	if err != nil {
+		file.Close()
+		log.Printf("Error occurred during request generation: %s\n", err.Error())
+		return
+	}
+
+	err = uploadFile(furObject, 0)
+	if err != nil {
+		log.Println(err.Error())
+	} else {
+		logs.IncrementScore(0)
+	}
+	file.Close()
+}
+
+func processMultipartUpload(gen3Interface Gen3Interface, multipartFilePaths []string, bucketName string, includeSubDirName bool, uploadPath string) {
+	profileConfig := conf.ParseConfig(profile)
+	if profileConfig.UseShepherd == "true" ||
+		profileConfig.UseShepherd == "" && commonUtils.DefaultUseShepherd == true {
+		log.Fatalf("Error: Shepherd currently does not support multipart uploads. For the moment, please disable Shepherd with\n	$ gen3-client configure --profile=%v --use-shepherd=false\nand try again.\n", profile)
+	}
+	log.Println("Multipart uploading....")
+
+	for _, filePath := range multipartFilePaths {
+		fileInfo, err := ProcessFilename(uploadPath, filePath, includeSubDirName, false)
+		if err != nil {
+			logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
+			log.Println("Process filename error for file: " + err.Error())
+			continue
+		}
+		err = multipartUpload(gen3Interface, fileInfo, 0, bucketName)
+		if err != nil {
+			log.Println(err.Error())
+		} else {
+			logs.IncrementScore(0)
+		}
+	}
+}

gen3-client/g3cmd/upload-single.go

@@ -16,6 +16,7 @@ import (
 func init() {
 	var guid string
 	var filePath string
+	var bucketName string
 
 	var uploadSingleCmd = &cobra.Command{
 		Use:     "upload-single",
@@ -65,7 +66,7 @@ func init() {
 			}
 			defer file.Close()
 
-			furObject := commonUtils.FileUploadRequestObject{FilePath: filePath, Filename: filename, GUID: guid}
+			furObject := commonUtils.FileUploadRequestObject{FilePath: filePath, Filename: filename, GUID: guid, Bucket: bucketName}
 
 			furObject, err = GenerateUploadRequest(gen3Interface, furObject, file)
 			if err != nil {
@@ -94,5 +95,6 @@ func init() {
 	uploadSingleCmd.MarkFlagRequired("guid") //nolint:errcheck
 	uploadSingleCmd.Flags().StringVar(&filePath, "file", "", "Specify file to upload to with --file=~/path/to/file")
 	uploadSingleCmd.MarkFlagRequired("file") //nolint:errcheck
+	uploadSingleCmd.Flags().StringVar(&bucketName, "bucket", "", "The bucket to which files will be uploaded. If not provided, defaults to Gen3's configured DATA_UPLOAD_BUCKET.")
 	RootCmd.AddCommand(uploadSingleCmd)
 }

gen3-client/g3cmd/upload.go

@@ -12,6 +12,7 @@ import (
 )
 
 func init() {
+	var bucketName string
 	var includeSubDirName bool
 	var uploadPath string
 	var batch bool
@@ -70,7 +71,7 @@ func init() {
 			}
 			fmt.Println()
 
-			singlepartFilePaths, multipartFilePaths := validateFilePath(filePaths, forceMultipart)
+			singlepartFilePaths, multipartFilePaths := separateSingleAndMultipartUploads(filePaths, forceMultipart)
 
 			if batch {
 				workers, respCh, errCh, batchFURObjects := initBatchUploadChannels(numParallel, len(singlepartFilePaths))
@@ -85,13 +86,13 @@ func init() {
 						furObject := commonUtils.FileUploadRequestObject{FilePath: fileInfo.FilePath, Filename: fileInfo.Filename, FileMetadata: fileInfo.FileMetadata, GUID: ""}
 						batchFURObjects = append(batchFURObjects, furObject)
 					} else {
-						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh)
+						batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh, bucketName)
 						batchFURObjects = make([]commonUtils.FileUploadRequestObject, 0)
 						furObject := commonUtils.FileUploadRequestObject{FilePath: fileInfo.FilePath, Filename: fileInfo.Filename, FileMetadata: fileInfo.FileMetadata, GUID: ""}
 						batchFURObjects = append(batchFURObjects, furObject)
 					}
 				}
-				batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh)
+				batchUpload(gen3Interface, batchFURObjects, workers, respCh, errCh, bucketName)
 
 				if len(errCh) > 0 {
 					close(errCh)
@@ -123,7 +124,7 @@ func init() {
 						continue
 					}
 					// The following flow is for singlepart upload flow
-					respURL, guid, err := GeneratePresignedURL(gen3Interface, fileInfo.Filename, fileInfo.FileMetadata)
+					respURL, guid, err := GeneratePresignedURL(gen3Interface, fileInfo.Filename, fileInfo.FileMetadata, bucketName)
 					if err != nil {
 						logs.AddToFailedLog(fileInfo.FilePath, fileInfo.Filename, fileInfo.FileMetadata, guid, 0, false, true)
 						log.Println(err.Error())
@@ -153,26 +154,7 @@ func init() {
 			if len(multipartFilePaths) > 0 {
 				// NOTE(@mpingram) - For the moment Shepherd doesn't support multipart uploads.
 				// Throw an error if Shepherd is enabled and user attempts to multipart upload.
-				profileConfig := conf.ParseConfig(profile)
-				if profileConfig.UseShepherd == "true" ||
-					profileConfig.UseShepherd == "" && commonUtils.DefaultUseShepherd == true {
-					log.Fatalf("Error: Shepherd currently does not support multipart uploads. For the moment, please disable Shepherd with\n	$ gen3-client configure --profile=%v --use-shepherd=false\nand try again.\n", profile)
-				}
-				log.Println("Multipart uploading....")
-				for _, filePath := range multipartFilePaths {
-					fileInfo, err := ProcessFilename(uploadPath, filePath, includeSubDirName, false)
-					if err != nil {
-						logs.AddToFailedLog(filePath, filepath.Base(filePath), commonUtils.FileMetadata{}, "", 0, false, true)
-						log.Println("Process filename error for file: " + err.Error())
-						continue
-					}
-					err = multipartUpload(gen3Interface, fileInfo, 0)
-					if err != nil {
-						log.Println(err.Error())
-					} else {
-						logs.IncrementScore(0)
-					}
-				}
+				processMultipartUpload(gen3Interface, multipartFilePaths, bucketName, includeSubDirName, uploadPath)
 			}
 
 			if !logs.IsFailedLogMapEmpty() {
@@ -192,5 +174,6 @@ func init() {
 	uploadCmd.Flags().BoolVar(&includeSubDirName, "include-subdirname", false, "Include subdirectory names in file name")
 	uploadCmd.Flags().BoolVar(&forceMultipart, "force-multipart", false, "Force to use multipart upload if possible")
 	uploadCmd.Flags().BoolVar(&hasMetadata, "metadata", false, "Search for and upload file metadata alongside the file")
+	uploadCmd.Flags().StringVar(&bucketName, "bucket", "", "The bucket to which files will be uploaded. If not provided, defaults to Gen3's configured DATA_UPLOAD_BUCKET.")
 	RootCmd.AddCommand(uploadCmd)
 }