Merge pull request #92 from Azure/dev

v.0.6.12

Author: giventocode

.gitignore

@@ -2,15 +2,18 @@
 *.o
 *.a
 *.so
+__*.*
 
 # Folders
 _obj
 _test
+_build/
 _build/linux_amd64
 _build/windows_amd64
 _wd
 _old
 .vscode
+*_testdata
 
 # Architecture specific extensions/prefixes
 *.[568vq]

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.6.09/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.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.6.09/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.6.12/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 

args.go

@@ -72,6 +72,7 @@ type arguments struct {
 	readTokenExp             int
 	numberOfHandlesPerFile   int //numberOfHandlesPerFile = defaultNumberOfHandlesPerFile
 	numberOfFilesInBatch     int //numberOfFilesInBatch = defaultNumberOfFilesInBatch
+	transferStatusPath       string
 }
 
 //represents validated parameters
@@ -95,6 +96,7 @@ type validatedParameters struct {
 	blobSource             blobParams
 	blobTarget             blobParams
 	perfSourceDefinitions  []sources.SourceDefinition
+	tracker                *internal.TransferTracker
 }
 
 type s3Source struct {
@@ -166,12 +168,14 @@ func (p *paramParserValidator) parseAndValidate() error {
 	p.targetSegment = t
 	err = p.runParseAndValidationRules(
 		p.pvgCalculateReadersAndWorkers,
+		p.pvgTransferStatusPathIsPresent,
 		p.pvgBatchLimits,
 		p.pvgHTTPTimeOut,
 		p.pvgDupCheck,
 		p.pvgParseBlockSize,
 		p.pvgQuietMode,
-		p.pvgKeepDirectoryStructure)
+		p.pvgKeepDirectoryStructure,
+		p.pvgUseExactMatch)
 
 	if err != nil {
 		return err
@@ -254,6 +258,27 @@ func (p *paramParserValidator) getSourceRules() ([]parseAndValidationRule, error
 //**************************
 
 //Global rules....
+func (p *paramParserValidator) pvgUseExactMatch() error {
+	p.params.useExactMatch = p.args.exactNameMatch
+	return nil
+}
+
+func (p *paramParserValidator) pvgTransferStatusPathIsPresent() error {
+
+	if p.args.transferStatusPath != "" {
+		if !p.args.quietMode{
+			fmt.Printf("Transfer is resumable. Transfer status file:%v \n", p.args.transferStatusPath)
+		}
+		tracker, err := internal.NewTransferTracker(p.args.transferStatusPath)
+
+		if err != nil {
+			return err
+		}
+
+		p.params.tracker = tracker
+	}
+	return nil
+}
 func (p *paramParserValidator) pvgKeepDirectoryStructure() error {
 	p.params.keepDirStructure = !p.args.removeDirStructure
 	return nil
@@ -503,7 +528,7 @@ func (p *paramParserValidator) pvSourceInfoForS3IsReq() error {
 	burl, err := url.Parse(p.params.sourceURIs[0])
 
 	if err != nil {
-		return fmt.Errorf("Invalid S3 endpoint URL. Parsing error: %v.\nThe format is s3://[END_POINT]/[BUCKET]/[OBJECT]", err)
+		return fmt.Errorf("Invalid S3 endpoint URL. Parsing error: %v.\nThe format is s3://[END_POINT]/[BUCKET]/[PREFIX]", err)
 	}
 
 	p.params.s3Source.endpoint = burl.Hostname()
@@ -514,10 +539,14 @@ func (p *paramParserValidator) pvSourceInfoForS3IsReq() error {
 
 	segments := strings.Split(burl.Path, "/")
 
+	if len(segments) < 2 {
+		return fmt.Errorf("Invalid S3 endpoint URL. Bucket not specified. The format is s3://[END_POINT]/[BUCKET]/[PREFIX]")		
+	}
+
 	p.params.s3Source.bucket = segments[1]
 
 	if p.params.s3Source.bucket == "" {
-		return fmt.Errorf("Invalid source S3 URI. Bucket name could be parsed")
+		return fmt.Errorf("Invalid source S3 URI. Bucket name could not be parsed")
 	}
 
 	prefix := ""

blobporter.go

@@ -10,13 +10,12 @@ import (
 	"strconv"
 	"sync/atomic"
 
+	"github.com/Azure/blobporter/internal"
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/transfer"
 	"github.com/Azure/blobporter/util"
-	"github.com/Azure/blobporter/internal"
 )
 
-
 var argsUtil paramParserValidator
 
 func init() {
@@ -46,6 +45,7 @@ func init() {
 		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"
 		readTokenExpMsg            = "Expiration in minutes of the read-only access token that will be generated to read from S3 or Azure Blob sources."
+		transferStatusFileMsg      = "Transfer status file location. If set, blobporter will use this file to track the status of the transfer.\n\tIn case of failure and if the option is set the same status file, source files that were transferred will be skipped.\n\tIf the transfer is successful a summary will be appended."
 	)
 
 	flag.Usage = func() {
@@ -68,6 +68,7 @@ func init() {
 		util.PrintUsageDefaults("h", "handles_per_file", strconv.Itoa(argsUtil.args.numberOfHandlesPerFile), numberOfHandlersPerFileMsg)
 		util.PrintUsageDefaults("x", "files_per_transfer", strconv.Itoa(argsUtil.args.numberOfFilesInBatch), numberOfFilesInBatchMsg)
 		util.PrintUsageDefaults("o", "read_token_exp", strconv.Itoa(defaultReadTokenExp), readTokenExpMsg)
+		util.PrintUsageDefaults("l", "transfer_status", "", transferStatusFileMsg)
 	}
 
 	util.StringListVarAlias(&argsUtil.args.sourceURIs, "f", "source_file", "", fileMsg)
@@ -89,39 +90,35 @@ func init() {
 	util.IntVarAlias(&argsUtil.args.numberOfHandlesPerFile, "h", "handles_per_file", defaultNumberOfHandlesPerFile, numberOfHandlersPerFileMsg)
 	util.IntVarAlias(&argsUtil.args.numberOfFilesInBatch, "x", "files_per_transfer", defaultNumberOfFilesInBatch, numberOfFilesInBatchMsg)
 	util.IntVarAlias(&argsUtil.args.readTokenExp, "o", "read_token_exp", defaultReadTokenExp, readTokenExpMsg)
-
+	util.StringVarAlias(&argsUtil.args.transferStatusPath, "l", "transfer_status", "", transferStatusFileMsg)
 }
 
 var dataTransferred uint64
 var targetRetries int32
 
-func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo, numOfBatches int, batchNumber int) {
-	if numOfBatches == 1 {
-		fmt.Printf("Files to Transfer (%v) :\n", argsUtil.params.transferType)
-		var totalSize uint64
-		summary := ""
-
-		for _, source := range sourcesInfo {
-			//if the source is URL, remove the QS
-			display := source.SourceName
-			if u, err := url.Parse(source.SourceName); err == nil {
-				display = fmt.Sprintf("%v%v", u.Hostname(), u.Path)
-			}
-			summary = summary + fmt.Sprintf("Source: %v Size:%v \n", display, source.Size)
-			totalSize = totalSize + source.Size
-		}
+func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo) {
+	fmt.Printf("\nFiles to Transfer (%v) :\n", argsUtil.params.transferType)
+	var totalSize uint64
+	summary := ""
 
-		if len(sourcesInfo) < 10 {
-			fmt.Printf(summary)
-			return
+	for _, source := range sourcesInfo {
+		//if the source is URL, remove the QS
+		display := source.SourceName
+		if u, err := url.Parse(source.SourceName); err == nil {
+			display = fmt.Sprintf("%v%v", u.Hostname(), u.Path)
 		}
+		summary = summary + fmt.Sprintf("Source: %v Size:%v \n", display, source.Size)
+		totalSize = totalSize + source.Size
+	}
 
-		fmt.Printf("%v files. Total size:%v\n", len(sourcesInfo), totalSize)
-
+	if len(sourcesInfo) < 10 {
+		fmt.Printf(summary)
 		return
 	}
 
-	fmt.Printf("\nBatch transfer (%v).\nFiles per Batch: %v.\nBatch: %v of %v\n ", argsUtil.params.transferType, len(sourcesInfo), batchNumber+1, numOfBatches)
+	fmt.Printf("%v files. Total size:%v\n", len(sourcesInfo), totalSize)
+
+	return
 }
 
 func main() {
@@ -141,21 +138,32 @@ func main() {
 
 	stats := transfer.NewStats(argsUtil.params.numberOfWorkers, argsUtil.params.numberOfReaders)
 
-	for b, sourcePipeline := range sourcePipelines {
-		sourcesInfo := sourcePipeline.GetSourcesInfo()
+	for sourcePipeline := range sourcePipelines {
+
+		if sourcePipeline.Err != nil {
+			log.Fatal(sourcePipeline.Err)
+		}
+
+		sourcesInfo := sourcePipeline.Source.GetSourcesInfo()
 
-		tfer := transfer.NewTransfer(&sourcePipeline, &targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
+		tfer := transfer.NewTransfer(sourcePipeline.Source, targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
+		tfer.SetTransferTracker(argsUtil.params.tracker)
 
-		displayFilesToTransfer(sourcesInfo, len(sourcePipelines), b)
+		displayFilesToTransfer(sourcesInfo)
 		pb := getProgressBarDelegate(tfer.TotalSize, argsUtil.params.quietMode)
 
 		tfer.StartTransfer(argsUtil.params.dedupeLevel, pb)
-
 		tfer.WaitForCompletion()
 
 		stats.AddTransferInfo(tfer.GetStats())
 	}
 
+	if argsUtil.params.tracker != nil {
+		if err = argsUtil.params.tracker.TrackTransferComplete(); err != nil {
+			log.Fatal(err)
+		}
+	}
+
 	stats.DisplaySummary()
 
 }

docs/bptransfer.png

@@ -15,12 +15,12 @@
 # import os
 # import sys
 # sys.path.insert(0, os.path.abspath('.'))
-
+import sphinx_bootstrap_theme
 
 # -- Project information -----------------------------------------------------
 
 project = u'BlobPorter'
-#copyright = u'2018, Jesus Aguilar'
+copyright = u'2018, BlobPorter Contributors'
 author = u'BlobPorter Contributors'
 
 # The short X.Y version
@@ -74,7 +74,9 @@
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 #
-html_theme = 'alabaster'
+html_theme = 'sphinx_rtd_theme'
+#html_theme = 'bootstrap'
+#html_theme_path = sphinx_bootstrap_theme.get_html_theme_path()
 
 # Theme options are theme-specific and customize the look and feel of a theme
 # further.  For a list of options available for each theme, see the