- transfers now resumable

- file stats are read async to define the shape of the transfer.
- refactoring.

Author: giventocode

.gitignore

@@ -2,6 +2,7 @@
 *.o
 *.a
 *.so
+__*.*
 
 # Folders
 _obj
@@ -11,6 +12,7 @@ _build/windows_amd64
 _wd
 _old
 .vscode
+*_testdata
 
 # Architecture specific extensions/prefixes
 *.[568vq]

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,6 +168,7 @@ func (p *paramParserValidator) parseAndValidate() error {
 	p.targetSegment = t
 	err = p.runParseAndValidationRules(
 		p.pvgCalculateReadersAndWorkers,
+		p.pvgTransferStatusPathIsPresent,
 		p.pvgBatchLimits,
 		p.pvgHTTPTimeOut,
 		p.pvgDupCheck,
@@ -254,6 +257,22 @@ func (p *paramParserValidator) getSourceRules() ([]parseAndValidationRule, error
 //**************************
 
 //Global rules....
+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

blobporter.go

@@ -45,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 created at then."
 	)
 
 	flag.Usage = func() {
@@ -67,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)
@@ -88,7 +90,7 @@ 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
@@ -144,18 +146,24 @@ func main() {
 
 		sourcesInfo := sourcePipeline.Source.GetSourcesInfo()
 
-		tfer := transfer.NewTransfer(&sourcePipeline.Source, &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)
 		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()
 
 }

internal/const.go

@@ -6,7 +6,7 @@ import (
 )
 
 //ProgramVersion blobporter version
-const ProgramVersion = "0.6.10"
+const ProgramVersion = "0.6.11"
 
 //HTTPClientTimeout  HTTP client timeout when reading from HTTP sources and try timeout for blob storage operations.
 var HTTPClientTimeout = 90