Merge branch 'master' of https://github.com/Azure/blobporter

Author: giventocode

.gitignore

@@ -8,12 +8,13 @@ _obj
 _test
 _build/linux_amd64
 _build/windows_amd64
+.vscode
 
 # Architecture specific extensions/prefixes
 *.[568vq]
 [568vq].out
 debug
-.vscode
+launch.json
 *.cgo1.go
 *.cgo2.c
 _cgo_defun.c
@@ -30,4 +31,5 @@ _testmain.go
 *.out
 
 # external packages folder
-#vendor/
+#vendor/
+.vscode/launch.json

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.13/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.14/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.13/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.14/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 
@@ -84,7 +84,6 @@ If you want to rename multiple files, you can use the -n option:
 
 `./blobporter -f /datadrive/f1.tar -f /datadrive/f2.md -n b1 -n b2 -c mycontainer`
 
-
 ### Upload to Azure Page Blob Storage
 
 Same as uploading to block blob storage, but with the transfer definiton (-t option) set to file-pageblob.
@@ -132,7 +131,6 @@ By default files are downloaded to the same directory where you are running blob
 
 `./blobporter -p -c mycontainer -t blob-file`
 
-
 ### Download a file via HTTP to a local file
 
 `./blobporter -f "http://mysource/file.bam"  -n /datadrive/file.bam -t http-file`
@@ -141,7 +139,7 @@ By default files are downloaded to the same directory where you are running blob
 
 ## Command Options
 
-- `-f`, `--file` *string* URL, file or files (e.g. /data/*.gz) to upload.
+- `-f`, `--source_file` *string* URL, file or files (e.g. /data/*.gz) to upload.
 
 - `-c`, `--container_name` *string* container name (e.g. `mycontainer`).
 

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.13" // version number to show in help
+	programVersion           = "0.5.14" // version number to show in help
 )
 
 const numOfWorkersFactor = 8
@@ -72,7 +72,7 @@ func init() {
 	)
 
 	const (
-		fileMsg                    = "URL, file or files (e.g. /data/*.gz) to upload."
+		fileMsg                    = "Source URL, file or files (e.g. /data/*.gz) to upload."
 		nameMsg                    = "Blob name (e.g. myblob.txt) or prefix for download scenarios."
 		containerNameMsg           = "Container name (e.g. mycontainer).\n\tIf the container does not exist, it will be created."
 		concurrentWorkersMsg       = "Number of workers for parallel upload."
@@ -93,7 +93,7 @@ func init() {
 	)
 
 	flag.Usage = func() {
-		util.PrintUsageDefaults("f", "file", "", fileMsg)
+		util.PrintUsageDefaults("f", "source_file", "", fileMsg)
 		util.PrintUsageDefaults("n", "name", "", nameMsg)
 		util.PrintUsageDefaults("c", "container_name", "", containerNameMsg)
 		util.PrintUsageDefaults("g", "concurrent_workers", strconv.Itoa(defaultNumberOfWorkers), concurrentWorkersMsg)
@@ -114,7 +114,7 @@ func init() {
 
 	}
 
-	util.StringListVarAlias(&sourceURIs, "f", "file", "", fileMsg)
+	util.StringListVarAlias(&sourceURIs, "f", "source_file", "", fileMsg)
 	util.StringListVarAlias(&blobNames, "n", "name", "", nameMsg)
 	util.StringVarAlias(&containerName, "c", "container_name", "", containerNameMsg)
 	util.IntVarAlias(&numberOfWorkers, "g", "concurrent_workers", defaultNumberOfWorkers, concurrentWorkersMsg)

pipeline/pipeline.go

@@ -4,7 +4,6 @@ import (
 	"crypto/md5"
 	"encoding/base64"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
@@ -48,7 +47,7 @@ type SourceInfo struct {
 
 //TargetPipeline operations that abstract how parts a written and processed to a given target
 type TargetPipeline interface {
-	PreProcessSourceInfo(source *SourceInfo) (err error)
+	PreProcessSourceInfo(source *SourceInfo, blockSize uint64) (err error)
 	CommitList(listInfo *TargetCommittedListInfo, numberOfBlocks int, targetName string) (msg string, err error)
 	WritePart(part *Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error)
 	ProcessWrittenPart(result *WorkerResult, listInfo *TargetCommittedListInfo) (requeue bool, err error)
@@ -154,11 +153,6 @@ func ConstructPartsPartition(numberOfPartitions int, size int64, blockSize int64
 	//bsib := uint64(blockSize)
 	numOfBlocks := int((size + blockSize - 1) / blockSize)
 
-	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + util.MaxBlockCount - 1) / util.MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
 	Partitions := make([]PartsPartition, numberOfPartitions)
 	//the size of the partition needs to be a multiple (blockSize * int) to make sure all but the last part/block
 	//are the same size
@@ -191,11 +185,6 @@ func ConstructPartsQueue(size uint64, blockSize uint64, sourceURI string, target
 	var bsib = blockSize
 	numOfBlocks = int((size + (bsib - 1)) / bsib)
 
-	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + util.MaxBlockCount - 1) / util.MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
 	parts = make([]Part, numOfBlocks)
 
 	var curFileOffset uint64

sources/http.go

@@ -67,7 +67,13 @@ func getSourceSize(sourceURI string) (size int) {
 	resp, err := client.Head(sourceURI)
 
 	if err != nil || resp.StatusCode != 200 {
-		err = fmt.Errorf("HEAD request failed. Please check the URL. Status:%d Error: %v", resp.StatusCode, err)
+		statusCode := ""
+		if resp != nil {
+			statusCode = fmt.Sprintf(" Status:%d ", resp.StatusCode)
+		}
+		err = fmt.Errorf("HEAD request failed. Please check the URL.%s Error: %v", statusCode, err)
+
+		util.PrintfIfDebug("getSourceSize -> %v", err)
 
 		size = getSourceSizeFromByteRangeHeader(sourceURI)
 		return

sources/multifile.go

@@ -117,8 +117,7 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 		if fileStat.Size() == 0 {
 			log.Fatalf("Empty files are not allowed. The file %v is empty", files[f])
 		}
-
-		numOfBlocks := util.GetNumberOfBlocks(uint64(fileStat.Size()), blockSize)
+		numOfBlocks := int(uint64(fileStat.Size())+(blockSize-1)) / int(blockSize)
 		totalSize = totalSize + uint64(fileStat.Size())
 		totalNumberOfBlocks = totalNumberOfBlocks + numOfBlocks
 

targets/azureblock.go

@@ -81,7 +81,14 @@ func convertToStorageBlockList(list interface{}, numOfBlocks int) []storage.Bloc
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //Checks if uncommitted blocks are present and cleans them by creating an empty blob.
-func (t *AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
+func (t *AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
+	numOfBlocks := int(source.Size+(blockSize-1)) / int(blockSize)
+
+	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
+		var minBlkSize = (source.Size + util.MaxBlockCount - 1) / util.MaxBlockCount
+		return fmt.Errorf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
+	}
+
 	return util.CleanUncommittedBlocks(&t.StorageClient, t.Container, source.TargetAlias)
 }
 

targets/azurepage.go

@@ -28,20 +28,28 @@ func NewAzurePage(accountName string, accountKey string, container string) pipel
 	return &AzurePage{Creds: &creds, Container: container, StorageClient: &client}
 }
 
-//PageSize size of page in Azure Page Blob storage
-const PageSize int64 = 512
+//Page blobs limits and units
+
+//PageSize page size for page blobs
+const PageSize uint64 = 512
+const maxPageSize uint64 = 4 * util.MB
+const maxPageBlobSize uint64 = 8 * util.TB
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //initializes the page blob.
-func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
+func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
 	size := int64(source.Size)
 
-	if size%PageSize != 0 {
-		return fmt.Errorf("Invalid size for a page blob. The size of the file %v (%v) is not a multiple of %v", source.SourceName, source.Size, PageSize)
+	if size%int64(PageSize) != 0 {
+		return fmt.Errorf(" invalid size for a page blob. The size of the file %v (%v) is not a multiple of %v ", source.SourceName, source.Size, PageSize)
+	}
+
+	if size > int64(maxPageBlobSize) {
+		return fmt.Errorf(" the file %v is too big (%v). Tha maximum size of a page blob is %v ", source.SourceName, source.Size, maxPageBlobSize)
 	}
 
-	if size > int64(util.TB) {
-		return fmt.Errorf("The file %v is too big (%v). Tha maximum size of a page blob is %v ", source.SourceName, source.Size, util.TB)
+	if blockSize > maxPageSize || blockSize < PageSize {
+		return fmt.Errorf(" invalid block size for page blob: %v. The value must be greater than %v and less than %v", PageSize, maxPageSize)
 	}
 
 	//if the max retries is exceeded, panic will happen, hence no error is returned.

targets/multifile.go

@@ -32,7 +32,7 @@ func NewMultiFile(overwrite bool, numberOfHandles int) pipeline.TargetPipeline {
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //Passthrough no need to pre-process for a file target.
-func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
+func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
 	t.Lock()
 	defer t.Unlock()
 

transfer/transfer.go

@@ -323,18 +323,18 @@ func (t *Transfer) StartTransfer(dupeLevel DupeCheckLevel, progressBarDelegate P
 	go t.processAndCommitResults(t.ControlChannels.Results, progressBarDelegate, t.TargetPipeline, &t.SyncWaitGroups.Commits)
 }
 
-//Sequentially calls the PreProcessSourceInfo implementation of the target pipeline for each source in the transfer.
+//Concurrely calls the PreProcessSourceInfo implementation of the target pipeline for each source in the transfer.
 func (t *Transfer) preprocessSources() {
 	sourcesInfo := (*t.SourcePipeline).GetSourcesInfo()
 	var wg sync.WaitGroup
 	wg.Add(len(sourcesInfo))
 	for i := 0; i < len(sourcesInfo); i++ {
-		go func(s *pipeline.SourceInfo) {
+		go func(s *pipeline.SourceInfo, b uint64) {
 			defer wg.Done()
-			if err := (*t.TargetPipeline).PreProcessSourceInfo(s); err != nil {
+			if err := (*t.TargetPipeline).PreProcessSourceInfo(s, b); err != nil {
 				log.Fatal(err)
 			}
-		}(&sourcesInfo[i])
+		}(&sourcesInfo[i], t.blockSize)
 	}
 
 	wg.Wait()

util/util.go

@@ -206,20 +206,6 @@ func RetriableOperation(operation func(r int) error) (duration time.Duration, st
 	}
 }
 
-///////////////////////////////////////////////////////////////////
-
-//GetNumberOfBlocks calculates the number of blocks from filesize and checks if the number is greater than what's allowed (MaxBlockCount).
-func GetNumberOfBlocks(size uint64, blockSize uint64) int {
-	numOfBlocks := int(size+(blockSize-1)) / int(blockSize)
-
-	if numOfBlocks > MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + MaxBlockCount - 1) / MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
-	return numOfBlocks
-}
-
 //CreateContainerIfNotExists Creates a new container if doesn't exists. Validates the name of the container.
 func CreateContainerIfNotExists(container string, accountName string, accountKey string) {