Merge pull request #61 from Azure/dev

504 release

Author: giventocode

README.md

@@ -135,7 +135,7 @@ Without the -n option all files in the container will be downloaded.
 
 ## Command Options
 
-- `-f`, `--file` *string* URL, file or files (e.g. /data/*.gz) to upload. Destination file for download scenarios.
+- `-f`, `--file` *string* URL, file or files (e.g. /data/*.gz) to upload.
 
 - `-c`, `--container_name` *string* container name (e.g. `mycontainer`).
 

blobporter.go

@@ -41,12 +41,13 @@ var storageAccountKey string
 var storageClientHTTPTimeout int
 var quietMode bool
 var calculateMD5 bool
+var exactNameMatch bool
 
 const (
 	// User can use environment variables to specify storage account information
 	storageAccountNameEnvVar = "ACCOUNT_NAME"
 	storageAccountKeyEnvVar  = "ACCOUNT_KEY"
-	programVersion           = "0.5.03" // version number to show in help
+	programVersion           = "0.5.04" // version number to show in help
 )
 
 const numOfWorkersFactor = 9
@@ -68,8 +69,8 @@ func init() {
 	)
 
 	const (
-		fileMsg              = "URL, file or files (e.g. /data/*.gz) to upload.\n\tDestination file for download."
-		nameMsg              = "Blob name for upload or download scenarios from Azure Blob Storage.\n\tDestination file name for downloads from a URL."
+		fileMsg              = "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."
 		concurrentReadersMsg = "Number of readers for parallel reading of the input file(s)."
@@ -82,6 +83,7 @@ func init() {
 		accountKeyMsg        = "Storage account key string.\n\tCan also be specified via the " + storageAccountKeyEnvVar + " environment variable."
 		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."
 	)
 
 	flag.Usage = func() {
@@ -99,6 +101,7 @@ func init() {
 		util.PrintUsageDefaults("k", "account_key", "", accountKeyMsg)
 		util.PrintUsageDefaults("d", "dup_check_level", dedupeLevelOptStr, dupcheckLevelMsg)
 		util.PrintUsageDefaults("t", "transfer_definition", string(defaultTransferDef), transferDefMsg)
+		util.PrintUsageDefaults("e", "exact_name", "false", exactNameMatchMsg)
 	}
 
 	util.StringListVarAlias(&sourceURIs, "f", "file", "", fileMsg)
@@ -115,7 +118,7 @@ func init() {
 	util.StringVarAlias(&storageAccountKey, "k", "account_key", "", accountKeyMsg)
 	util.StringVarAlias(&dedupeLevelOptStr, "d", "dup_check_level", dedupeLevelOptStr, dupcheckLevelMsg)
 	util.StringVarAlias(&transferDefStr, "t", "transfer_definition", string(defaultTransferDef), transferDefMsg)
-
+	util.BoolVarAlias(&exactNameMatch, "e", "exact_name", false, exactNameMatchMsg)
 }
 
 var dataTransferred uint64
@@ -236,7 +239,7 @@ func getPipelines() (pipeline.SourcePipeline, pipeline.TargetPipeline) {
 			blobNames = []string{""}
 		}
 
-		source = sources.NewAzureBlob(containerName, blobNames, storageAccountName, storageAccountKey, calculateMD5)
+		source = sources.NewAzureBlob(containerName, blobNames, storageAccountName, storageAccountKey, calculateMD5, exactNameMatch)
 		target = targets.NewMultiFile(true, numberOfWorkers)
 
 	case transfer.HTTPToFile:

sources/azureblob.go

@@ -20,17 +20,22 @@ import (
 
 //AzureBlob constructs parts  channel and implements data readers for Azure Blobs exposed via HTTP
 type AzureBlob struct {
-	HTTPSource    HTTPPipeline
-	Container     string
-	BlobNames     []string
-	storageClient storage.BlobStorageClient
+	HTTPSource     HTTPPipeline
+	Container      string
+	BlobNames      []string
+	exactNameMatch bool
+	storageClient  storage.BlobStorageClient
 }
 
 //NewAzureBlob creates a new instance of HTTPPipeline
 //Creates a SASURI to the blobName with an expiration of sasTokenNumberOfHours.
 //blobName is used as the target alias.
-func NewAzureBlob(container string, blobNames []string, accountName string, accountKey string, md5 bool) pipeline.SourcePipeline {
-	azureSource := AzureBlob{Container: container, BlobNames: blobNames, storageClient: util.GetBlobStorageClient(accountName, accountKey)}
+func NewAzureBlob(container string, blobNames []string, accountName string, accountKey string, md5 bool, useExactNameMatch bool) pipeline.SourcePipeline {
+	azureSource := AzureBlob{Container: container,
+		BlobNames:      blobNames,
+		storageClient:  util.GetBlobStorageClient(accountName, accountKey),
+		exactNameMatch: useExactNameMatch}
+
 	if sourceURIs, err := azureSource.getSourceURIs(); err == nil {
 		httpSource := NewHTTP(sourceURIs, nil, md5)
 		azureSource.HTTPSource = httpSource.(HTTPPipeline)
@@ -73,17 +78,29 @@ func (f AzureBlob) getSourceURIs() ([]string, error) {
 	var sourceURI string
 	for _, blobList := range blobLists {
 		for _, blob := range blobList.Blobs {
-			sourceURI, err = f.storageClient.GetBlobSASURI(f.Container, blob.Name, date, "r")
 
-			if err != nil {
-				return nil, err
+			include := true
+			if f.exactNameMatch {
+				include = blob.Name == blobList.Prefix
+			}
+
+			if include {
+				sourceURI, err = f.storageClient.GetBlobSASURI(f.Container, blob.Name, date, "r")
+
+				if err != nil {
+					return nil, err
+				}
+				sourceURIs = append(sourceURIs, sourceURI)
 			}
-			sourceURIs = append(sourceURIs, sourceURI)
 		}
 	}
 
 	if len(sourceURIs) == 0 {
-		return nil, fmt.Errorf("the prefix %v (-n option) did not match any blob names", f.BlobNames)
+		nameMatchMode := "prefix"
+		if f.exactNameMatch {
+			nameMatchMode = "name"
+		}
+		return nil, fmt.Errorf("the %v %s did not match any blob names", nameMatchMode, f.BlobNames)
 	}
 
 	return sourceURIs, nil
@@ -100,12 +117,29 @@ func (f AzureBlob) getBlobLists() ([]storage.BlobListResponse, error) {
 	listOfLists := make([]storage.BlobListResponse, listLength)
 
 	for i, blobname := range f.BlobNames {
+		var list *storage.BlobListResponse
 		params := storage.ListBlobsParameters{Prefix: blobname}
-		var list storage.BlobListResponse
-		if list, err = f.storageClient.ListBlobs(f.Container, params); err != nil {
-			return nil, err
+
+		for {
+			var listpage storage.BlobListResponse
+			if listpage, err = f.storageClient.ListBlobs(f.Container, params); err != nil {
+				return nil, err
+			}
+
+			if list == nil {
+				list = &listpage
+			} else {
+				(*list).Blobs = append((*list).Blobs, listpage.Blobs...)
+			}
+
+			if listpage.NextMarker == "" {
+				break
+			}
+
+			params.Marker = listpage.NextMarker
 		}
-		listOfLists[i] = list
+
+		listOfLists[i] = *list
 	}
 
 	return listOfLists, nil

targets/azureblock.go

@@ -70,9 +70,9 @@ func convertToStorageBlockList(list interface{}, numOfBlocks int) []storage.Bloc
 }
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
-//Passthrough no need to pre-process for blob blocks.
+//Checks if uncommitted blocks are present and cleans them by creating an empty blob.
 func (t AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
-	return nil
+	return util.CleanUncommittedBlocks(&t.StorageClient, t.Container, source.TargetAlias)
 }
 
 //ProcessWrittenPart implements ProcessWrittenPart from the pipeline.TargetPipeline interface.

targets/azurepage.go

@@ -32,7 +32,7 @@ func NewAzurePage(accountName string, accountKey string, container string) pipel
 const PageSize int64 = 512
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
-//Passthrough no need to pre-process for blob blocks.
+//initializes the page blob.
 func (t AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
 	size := int64(source.Size)
 

transfer/transfer_test.go

@@ -187,7 +187,7 @@ func TestFileToBlobToFile(t *testing.T) {
 	tfer.WaitForCompletion()
 
 	ap = targets.NewMultiFile(true, numOfWorkers)
-	fp = sources.NewAzureBlob(container, []string{sourceFile}, accountName, accountKey, true)
+	fp = sources.NewAzureBlob(container, []string{sourceFile}, accountName, accountKey, true, false)
 	tfer = NewTransfer(&fp, &ap, numOfReaders, numOfWorkers, blockSize)
 	tfer.StartTransfer(None, delegate)
 	tfer.WaitForCompletion()
@@ -207,7 +207,7 @@ func TestFileToBlobToFileWithAlias(t *testing.T) {
 	tfer.WaitForCompletion()
 
 	ap = targets.NewMultiFile(true, numOfWorkers)
-	fp = sources.NewAzureBlob(container, []string{alias}, accountName, accountKey, true)
+	fp = sources.NewAzureBlob(container, []string{alias}, accountName, accountKey, true, false)
 	tfer = NewTransfer(&fp, &ap, numOfReaders, numOfWorkers, blockSize)
 	tfer.StartTransfer(None, delegate)
 	tfer.WaitForCompletion()

util/util.go

@@ -171,7 +171,7 @@ func BoolVarAlias(varPtr *bool, shortflag string, longflag string, defaultVal bo
 // Retriable execution of a function
 ///////////////////////////////////////////////////////////////////
 
-const retryLimit = 50                             // max retries for an operation in retriableOperation
+const retryLimit = 100                            // max retries for an operation in retriableOperation
 const retrySleepDuration = time.Millisecond * 200 // Retry wait interval in retriableOperation
 
 //RetriableOperation executes a function, retrying up to "retryLimit" times and waiting "retrySleepDuration" between attempts
@@ -244,6 +244,31 @@ func CreateContainerIfNotExists(container string, accountName string, accountKey
 
 }
 
+//CleanUncommittedBlocks clears uncommitted blocks if an existing blob does not exists.
+func CleanUncommittedBlocks(client *storage.BlobStorageClient, container string, blobName string) error {
+	list, _ := client.GetBlockList(container, blobName, "Uncommitted")
+
+	if len(list.UncommittedBlocks) == 0 {
+		return nil
+	}
+
+	fmt.Printf("Warning! uncommitted blocks detected for blob %v \nAttempting to clean them up\n", blobName)
+
+	exists, err := client.BlobExists(container, blobName)
+
+	if exists {
+		fmt.Printf("Can't delete uncommitted blocks for the blob:%v. A committed blob with the same name already exists \n", blobName)
+		return nil
+	}
+
+	if err != nil {
+		return err
+	}
+
+	return client.CreateBlockBlob(container, blobName)
+
+}
+
 //GetBlobStorageClient gets a storage client with support for larg block blobs
 func GetBlobStorageClient(accountName string, accountKey string) storage.BlobStorageClient {
 	var bc storage.BlobStorageClient