Merge pull request #68 from Azure/dev

v0.5.12

Author: giventocode

.vscode/launch.json

@@ -12,10 +12,10 @@
             "host": "127.0.0.1",
             "program": "${fileDirname}",
             "env": {
-                "ACCOUNT_NAME":"storagejaa",
-                "ACCOUNT_KEY":"/mXWG4aXMWvftzR+Ed5URccwrDSvvv5cUsKCqq5gbyPXvtlyfQ9gjq462kuH/2dErjwR8QcAS74vLaNXynd74g=="
+                "ACCOUNT_NAME":"bprelease",
+                "ACCOUNT_KEY":"bGgXq0WvdkGRKe38Icww4i+3NUsb4E0ngm9tImN/liiB9uix/Wwgua45x+uP6A6EQCv9QIAyk73mlcn2bJ+vkQ=="
             },
-            "args": ["-c", "many", "-t","blob-file","-p"] ,
+            "args": ["-c","100k","-t","blob-file","-p","-x","300"] ,
             "showLog": true
         }
     ]

README.md

@@ -29,7 +29,7 @@ Sources and targets are decoupled, this design enables the composition of variou
 Download, extract and set permissions:
 
 ```bash
-wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.10/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.12/bp_linux.tar.gz
 tar -xvf bp_linux.tar.gz linux_amd64/blobporter
 chmod +x ~/linux_amd64/blobporter
 cd ~/linux_amd64
@@ -46,7 +46,7 @@ export ACCOUNT_KEY=<STORAGE_ACCOUNT_KEY>
 
 ### Windows
 
-Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.10/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.12/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 
@@ -175,9 +175,11 @@ By default files are downloaded to the same directory where you are running blob
 
 - `q`, `--quiet_mode` *bool* if present or true, the progress indicator is not displayed. The files to transfer, errors, warnings and transfer completion summary is still displayed.
 
-- `x`, `---files_per_transfer` *int* number of files in a batch transfer. Default is 200.
+- `x`, `--files_per_transfer` *int* number of files in a batch transfer. Default is 200.
 
-- `h`, `----handles_per_file` *int* number of open handles for concurrent reads and writes per file. Default is 2.
+- `h`, `--handles_per_file` *int* number of open handles for concurrent reads and writes per file. Default is 2.
+
+- `p`, `--keep_directories` *bool* if set blobs are downloaded or uploaded keeping the directory structure from the source. Not applicable when the source is a HTTP endpoint.
 
 ## Performance Considerations
 
@@ -189,11 +191,9 @@ By default, BlobPorter creates 5 readers and 8 workers for each core on the comp
 
 - For transfers from fast disks (SSD) or HTTP sources reducing the number readers or workers could provide better performance than the default values. Reduce these values if you want to minimize resource utilization. Lowering these numbers reduces contention and the likelihood of experiencing throttling conditions.
 
-- Starting with version 0.5.10:
-
-- - Transfers are batched. Each batch transfer will concurrently read and transfer up to 200 files (default value) from the source. The batch size can be modified using the -x option, the maximum value is 500.
+- Transfers can be batched. Each batch transfer will concurrently read and transfer up to 200 files (default value) from the source. The batch size can be modified using the -x option, the maximum value is 500.
 
-- - Blobs smaller than the block size are transferred in a single operation. With relatively small files (<32MB) performance may be better if you set a block size equal to the size of the files. Setting the number of workers and readers to the number of files could yield performance gains.
+- Blobs smaller than the block size are transferred in a single operation. With relatively small files (<32MB) performance may be better if you set a block size equal to the size of the files. Setting the number of workers and readers to the number of files could yield performance gains.
 
 ## Issues and Feedback
 

blobporter.go

@@ -49,7 +49,7 @@ const (
 	// User can use environment variables to specify storage account information
 	storageAccountNameEnvVar = "ACCOUNT_NAME"
 	storageAccountKeyEnvVar  = "ACCOUNT_KEY"
-	programVersion           = "0.5.10" // version number to show in help
+	programVersion           = "0.5.12" // version number to show in help
 )
 
 const numOfWorkersFactor = 8
@@ -87,7 +87,7 @@ func init() {
 		dupcheckLevelMsg           = "Desired level of effort to detect duplicate data to minimize upload size.\n\tMust be one of " + transfer.DupeCheckLevelStr
 		transferDefMsg             = "Defines the type of source and target in the transfer.\n\tMust be one of:\n\tfile-blockblob, file-pageblob, http-blockblob, http-pageblob, blob-file,\n\tpageblock-file (alias of blob-file), blockblob-file (alias of blob-file)\n\tor http-file."
 		exactNameMatchMsg          = "If set or true only blobs that match the name exactly will be downloaded."
-		keepDirStructureMsg        = "If set blobs are downloaded to the same directory structure as in the storage account.\n\tIf the directory structure does not exists it will be created."
+		keepDirStructureMsg        = "If set blobs are downloaded or uploaded keeping the directory structure from the source.\n\tNot applicable when the source is a HTTP endpoint."
 		numberOfHandlersPerFileMsg = "Number of open handles for concurrent reads and writes per file."
 		numberOfFilesInBatchMsg    = "Maximum number of files in a transfer.\n\tIf the number is exceeded new transfers are created"
 	)
@@ -257,6 +257,7 @@ func getFileToPagePipelines() (source []pipeline.SourcePipeline, target pipeline
 		TargetAliases:    blobNames,
 		NumOfPartitions:  numberOfReaders,
 		MD5:              calculateMD5,
+		KeepDirStructure: keepDirStructure,
 		FilesPerPipeline: numberOfFilesInBatch}
 
 	source = sources.NewMultiFile(params)
@@ -275,6 +276,7 @@ func getFileToBlockPipelines() (source []pipeline.SourcePipeline, target pipelin
 			TargetAliases:    blobNames,
 			NumOfPartitions:  numberOfReaders,
 			MD5:              calculateMD5,
+			KeepDirStructure: keepDirStructure,
 			FilesPerPipeline: numberOfFilesInBatch}
 
 		source = sources.NewMultiFile(params)

sources/http.go

@@ -142,14 +142,15 @@ func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, p
 			return // no more blocks of file data to be read
 		}
 
-		if req, err = http.NewRequest("GET", p.SourceURI, nil); err != nil {
-			log.Fatal(err)
-		}
+		util.RetriableOperation(func(r int) error {
+			if req, err = http.NewRequest("GET", p.SourceURI, nil); err != nil {
+				log.Fatal(err)
+			}
 
-		header := fmt.Sprintf("bytes=%v-%v", p.Offset, p.Offset-1+uint64(p.BytesToRead))
-		req.Header.Set("Range", header)
+			header := fmt.Sprintf("bytes=%v-%v", p.Offset, p.Offset-1+uint64(p.BytesToRead))
+			req.Header.Set("Range", header)
+			req.Close = true
 
-		util.RetriableOperation(func(r int) error {
 			if res, err = f.HTTPClient.Do(req); err != nil || res.StatusCode != 206 {
 				var status int
 				if res != nil {
@@ -162,7 +163,6 @@ func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, p
 				}
 				return err
 			}
-
 			p.Data, err = ioutil.ReadAll(res.Body)
 			res.Body.Close()
 			if err != nil {

sources/multifile.go

@@ -45,6 +45,7 @@ type MultiFileParams struct {
 	NumOfPartitions  int
 	MD5              bool
 	FilesPerPipeline int
+	KeepDirStructure bool
 }
 
 // NewMultiFile creates a new MultiFilePipeline.
@@ -87,14 +88,19 @@ func NewMultiFile(params *MultiFileParams) []pipeline.SourcePipeline {
 		if len(params.TargetAliases) == len(files) {
 			targetAlias = params.TargetAliases[start : start+numOfFilesInBatch]
 		}
-		pipelines[b] = newMultiFilePipeline(files[start:start+numOfFilesInBatch], targetAlias, params.BlockSize, params.NumOfPartitions, params.MD5)
+		pipelines[b] = newMultiFilePipeline(files[start:start+numOfFilesInBatch],
+			targetAlias,
+			params.BlockSize,
+			params.NumOfPartitions,
+			params.MD5,
+			params.KeepDirStructure)
 		filesSent = filesSent - numOfFilesInBatch
 	}
 
 	return pipelines
 }
 
-func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint64, numOfPartitions int, md5 bool) pipeline.SourcePipeline {
+func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint64, numOfPartitions int, md5 bool, keepDirStructure bool) pipeline.SourcePipeline {
 	totalNumberOfBlocks := 0
 	var totalSize uint64
 	var err error
@@ -116,12 +122,16 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 		totalSize = totalSize + uint64(fileStat.Size())
 		totalNumberOfBlocks = totalNumberOfBlocks + numOfBlocks
 
-		//use the param instead of the original filename only when  single file
-		//transfer occurs.
+		//use the param instead of the original filename only when
+		//the number of targets matches the number files to transfer
 		if useTargetAlias {
 			sName = targetAliases[f]
 		} else {
 			sName = fileStat.Name()
+			if keepDirStructure {
+				sName = files[f]
+			}
+
 		}
 
 		fileInfo := FileInfo{FileStats: &fileStat, SourceURI: files[f], TargetAlias: sName, NumOfBlocks: numOfBlocks}

targets/azureblock.go

@@ -58,6 +58,7 @@ func (t *AzureBlock) CommitList(listInfo *pipeline.TargetCommittedListInfo, numb
 	//if the max retries is exceeded, panic will happen, hence no error is returned.
 	duration, _, _ := util.RetriableOperation(func(r int) error {
 		if err := t.StorageClient.PutBlockList(t.Container, targetName, blockList); err != nil {
+			t.resetClient()
 			return err
 		}
 		return nil
@@ -126,9 +127,8 @@ func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, sta
 				uint64(len(part.Data)),
 				bytes.NewReader(part.Data),
 				headers); err != nil {
-				if util.Verbose {
-					fmt.Printf("EH|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-				}
+				util.PrintfIfDebug("Error|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
+				t.resetClient()
 				return err
 			}
 			return nil
@@ -140,16 +140,17 @@ func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, sta
 			uint64(len(part.Data)),
 			part.NewBuffer(),
 			headers); err != nil {
-			if util.Verbose {
-				fmt.Printf("EH|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-			}
+			util.PrintfIfDebug("Error|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
+			t.resetClient()
 			return err
 		}
 
-		if util.Verbose {
-			fmt.Printf("OKA|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-		}
+		util.PrintfIfDebug("OK|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
 		return nil
 	})
 	return
 }
+
+func (t *AzureBlock) resetClient() {
+	t.StorageClient = util.GetBlobStorageClientWithNewHTTPClient(t.Creds.AccountName, t.Creds.AccountKey)
+}

targets/azurepage.go

@@ -47,6 +47,7 @@ func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error
 	//if the max retries is exceeded, panic will happen, hence no error is returned.
 	util.RetriableOperation(func(r int) error {
 		if err := (*t.StorageClient).PutPageBlob(t.Container, (*source).TargetAlias, size, nil); err != nil {
+			t.resetClient()
 			return err
 		}
 		return nil
@@ -90,6 +91,7 @@ func (t *AzurePage) WritePart(part *pipeline.Part) (duration time.Duration, star
 			if util.Verbose {
 				fmt.Printf("EH|S|%v|%v|%v|%v\n", part.Offset, len(part.Data), part.TargetAlias, err)
 			}
+			t.resetClient()
 			return err
 		}
 
@@ -101,3 +103,8 @@ func (t *AzurePage) WritePart(part *pipeline.Part) (duration time.Duration, star
 
 	return
 }
+
+func (t *AzurePage) resetClient() {
+	client := util.GetBlobStorageClientWithNewHTTPClient(t.Creds.AccountName, t.Creds.AccountKey)
+	t.StorageClient = &client
+}

targets/multifile.go

@@ -108,7 +108,6 @@ func (t *MultiFile) WritePart(part *pipeline.Part) (duration time.Duration, star
 	startTime = time.Now()
 
 	fh := <-t.FileHandles[part.TargetAlias]
-
 	if _, err := fh.WriteAt((*part).Data, int64((*part).Offset)); err != nil {
 		log.Fatal(err)
 	}

util/util.go

@@ -290,6 +290,27 @@ func GetBlobStorageClient(accountName string, accountKey string) storage.BlobSto
 	return bc
 }
 
+//GetBlobStorageClientWithNewHTTPClient gets a storage client with a new instace of the HTTP client.
+func GetBlobStorageClientWithNewHTTPClient(accountName string, accountKey string) storage.BlobStorageClient {
+	var bc storage.BlobStorageClient
+	var client storage.Client
+	var err error
+
+	if accountName == "" || accountKey == "" {
+		log.Fatal("Storage account and/or key not specified via options or in environment variables ACCOUNT_NAME and ACCOUNT_KEY")
+	}
+
+	if client, err = storage.NewClient(accountName, accountKey, storage.DefaultBaseURL, LargeBlockAPIVersion, true); err != nil {
+		log.Fatal(err)
+	}
+
+	client.HTTPClient = NewHTTPClient()
+
+	bc = client.GetBlobService()
+
+	return bc
+}
+
 //GetBlobStorageClientWithSASToken gets a storage client with support for large block blobs
 func GetBlobStorageClientWithSASToken(accountName string, sasToken string) storage.BlobStorageClient {
 	var bc storage.BlobStorageClient
@@ -357,8 +378,8 @@ var storageHTTPClient *http.Client
 var HTTPClientTimeout = 600
 
 const (
-	maxIdleConns        = 100
-	maxIdleConnsPerHost = 100
+	maxIdleConns        = 50
+	maxIdleConnsPerHost = 50
 )
 
 func getStorageHTTPClient() *http.Client {
@@ -392,13 +413,21 @@ func getSuggestion(err error) string {
 		return "Try using a different container or upload and then delete a small blob with the same name."
 	case strings.Contains(err.Error(), "Client.Timeout"):
 		return "Try increasing the timeout using the -s option or reducing the number of workers and readers, options: -r and -g"
-	case strings.Contains(err.Error(),"too many open files"):
+	case strings.Contains(err.Error(), "too many open files"):
 		return "Try increasing the number of open files allowed. For debian systems you can try: ulimit -Sn 2048 "
 	default:
 		return ""
 	}
 }
 
+//PrintfIfDebug TODO
+func PrintfIfDebug(format string, values ...interface{}) {
+	if Verbose {
+		msg := fmt.Sprintf(format, values...)
+		fmt.Printf("%v\t%s\n", time.Now(), msg)
+	}
+}
+
 //GetFileNameFromURL returns last part of URL (filename)
 func GetFileNameFromURL(sourceURI string) (string, error) {