-Expiration of the readonly access token for S3/blob is configurable

-Winsock timeout is now retriable.
-Code refactoring

Author: giventocode

.gitignore

@@ -9,6 +9,7 @@ _test
 _build/linux_amd64
 _build/windows_amd64
 _wd
+_old
 .vscode
 
 # Architecture specific extensions/prefixes

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.07/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.09/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.07/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.6.09/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 
@@ -208,6 +208,8 @@ By default files are downloaded to the same directory where you are running blob
 
 - `i`, `--remove_directories` *bool* if set blobs are downloaded or uploaded without keeping the directory structure of the source. Not applicable when the source is a HTTP endpoint.
 
+- `o`, `--read_token_exp` *int* Expiration in minutes of the read-only access token that will be generated to read from S3 or Azure Blob sources. Default value: 360.
+
 ## Performance Considerations
 
 By default, BlobPorter creates 5 readers and 8 workers for each core on the computer. You can overwrite these values by using the options -r (number of readers) and -g (number of workers). When overriding these options there are few considerations:

_build/release0609.zip

@@ -8,6 +8,7 @@ import (
 	"runtime"
 	"strings"
 
+	"github.com/Azure/blobporter/internal"
 	"github.com/Azure/blobporter/sources"
 	"github.com/Azure/blobporter/targets"
 	"github.com/Azure/blobporter/transfer"
@@ -31,9 +32,10 @@ const numOfWorkersFactor = 8
 const numOfReadersFactor = 5
 const defaultNumberOfFilesInBatch = 500
 const defaultNumberOfHandlesPerFile = 2
-const defaultHTTPClientTimeout = 30
+const defaultHTTPClientTimeout = 60
 const defaultBlockSizeStr = "8MB"
 const defaultDedupeLevelStr = "None"
+const defaultReadTokenExp = 360
 
 //sets the parameters by parsing and validating the arguments
 type paramParserValidator struct {
@@ -67,6 +69,7 @@ type arguments struct {
 	exactNameMatch           bool
 	removeDirStructure       bool
 	hTTPClientTimeout        int
+	readTokenExp             int
 	numberOfHandlesPerFile   int //numberOfHandlesPerFile = defaultNumberOfHandlesPerFile
 	numberOfFilesInBatch     int //numberOfFilesInBatch = defaultNumberOfFilesInBatch
 }
@@ -109,6 +112,7 @@ type blobParams struct {
 	container   string
 	prefixes    []string
 	baseBlobURL string
+	sasExpMin   int
 }
 
 func (b blobParams) isSourceAuthAndContainerInfoSet() bool {
@@ -134,14 +138,16 @@ func newParamParserValidator() paramParserValidator {
 		numberOfWorkers:        defaultNumberOfWorkers,
 		blockSizeStr:           defaultBlockSizeStr,
 		dedupeLevelOptStr:      defaultDedupeLevelStr,
+		readTokenExp:           defaultReadTokenExp,
 		transferDefStr:         string(transfer.FileToBlock),
 		numberOfHandlesPerFile: defaultNumberOfHandlesPerFile,
 		hTTPClientTimeout:      defaultHTTPClientTimeout,
 		numberOfFilesInBatch:   defaultNumberOfFilesInBatch}
 	params := &validatedParameters{
 		s3Source: s3Source{
-			preSignedExpMin: defaulPreSignedExpMins},
-		blobSource: blobParams{},
+			preSignedExpMin: defaultReadTokenExp},
+		blobSource: blobParams{
+			sasExpMin: defaultReadTokenExp},
 		blobTarget: blobParams{}}
 
 	p := paramParserValidator{args: args, params: params}
@@ -256,6 +262,7 @@ func (p *paramParserValidator) pvgQuietMode() error {
 	p.params.quietMode = p.args.quietMode
 	return nil
 }
+
 func (p *paramParserValidator) pvgCalculateReadersAndWorkers() error {
 
 	if p.args.numberOfWorkers <= 0 {
@@ -289,10 +296,10 @@ func (p *paramParserValidator) pvgBatchLimits() error {
 }
 func (p *paramParserValidator) pvgHTTPTimeOut() error {
 	if p.args.hTTPClientTimeout < defaultHTTPClientTimeout {
-		fmt.Printf("Warning! The storage HTTP client timeout is too low (<30). Setting value to default (%v)s \n", defaultHTTPClientTimeout)
+		fmt.Printf("Warning! The HTTP timeout is too low (<30). Setting value to default (%v)s \n", defaultHTTPClientTimeout)
 		p.args.hTTPClientTimeout = defaultHTTPClientTimeout
 	}
-	util.HTTPClientTimeout = p.args.hTTPClientTimeout
+	internal.HTTPClientTimeout = p.args.hTTPClientTimeout
 	return nil
 }
 func (p *paramParserValidator) pvgDupCheck() error {
@@ -431,6 +438,12 @@ func (p *paramParserValidator) pvSourceInfoForBlobIsReq() error {
 
 		p.params.blobSource.prefixes = p.args.blobNames
 
+		if p.args.readTokenExp < 1 {
+			return fmt.Errorf("Invalid read token expiration value. Minimum is 1 (1m)")
+		}
+
+		p.params.blobSource.sasExpMin = p.args.readTokenExp
+
 		if p.params.blobSource.isSourceAuthAndContainerInfoSet() {
 			return nil
 		}
@@ -477,6 +490,12 @@ func (p *paramParserValidator) pvSourceInfoForBlobIsReq() error {
 		return fmt.Errorf("The source storage acccount key (env variable: %v) is required for this transfer type", sourceAuthorizationEnvVar)
 	}
 
+	if p.args.readTokenExp < 1 {
+		return fmt.Errorf("Invalid read token expiration value. Minimum is 1 (1m)")
+	}
+
+	p.params.blobSource.sasExpMin = p.args.readTokenExp
+
 	return nil
 }
 
@@ -519,6 +538,12 @@ func (p *paramParserValidator) pvSourceInfoForS3IsReq() error {
 		return fmt.Errorf("The S3 secret key is required for this transfer type. Environment variable name:%v", s3SecretKeyEnvVar)
 	}
 
+	if p.args.readTokenExp < 1 {
+		return fmt.Errorf("Invalid read token expiration value. Minimum is 1 (1m)")
+	}
+
+	p.params.s3Source.preSignedExpMin = p.args.readTokenExp
+
 	return nil
 }
 

args.go

@@ -40,6 +40,31 @@ func TestBasicUploadWithAlias(t *testing.T) {
 	assert.Equal(t, val.args.sourceURIs[0], val.params.sourceURIs[0], "source is missing")
 	assert.Equal(t, val.args.blobNames[0], val.params.targetAliases[0], "target alias is missing")
 }
+func TestReadTokenExpForBlobSource(t *testing.T) {
+	val := newParamParserValidator()
+	val.args.blobNames = []string{"data"}
+	val.args.storageAccountName = "myaccount"
+	val.args.storageAccountKey = "mykey"
+	val.args.containerName = "mycont"
+	val.args.transferDefStr = "blob-file"
+
+	err := val.parseAndValidate()
+	assert.NoError(t, err, "un expected error, all params should be set")
+	assert.Equal(t, defaultReadTokenExp, val.params.blobSource.sasExpMin, "expiration time is not the default")
+
+	val = newParamParserValidator()
+	val.args.blobNames = []string{"data"}
+	val.args.storageAccountName = "myaccount"
+	val.args.storageAccountKey = "mykey"
+	val.args.containerName = "mycont"
+	val.args.transferDefStr = "blob-file"
+	val.args.readTokenExp = 10
+
+	err = val.parseAndValidate()
+	assert.NoError(t, err, "un expected error, all params should be set")
+	assert.Equal(t, val.args.readTokenExp, val.params.blobSource.sasExpMin, "expiration time is not the expected value")
+
+}
 func TestBasicUploaBlockSizeLimits(t *testing.T) {
 	val := newParamParserValidator()
 	val.args.sourceURIs = []string{"data"}
@@ -154,4 +179,92 @@ func TestLongOptionDownload(t *testing.T) {
 	assert.Equal(t, tempval, val.params.blobSource.accountKey, "account key is not set")
 	assert.Equal(t, c, val.params.blobSource.container, "container is not set")
 	assert.Equal(t, b, val.params.blobSource.prefixes[0], "blobname is missing")
+	assert.Equal(t, val.args.readTokenExp, val.params.blobSource.sasExpMin, "expiration time is not the expected value")
+
 }
+
+func TestS3Transfer(t *testing.T) {
+	val := newParamParserValidator()
+	url := "mys3.myurl.com"
+	bucket := "bucket"
+	val.args.sourceURIs = []string{fmt.Sprintf("s3://%v/%v", url, bucket)}
+	val.args.storageAccountName = "myaccount"
+	val.args.storageAccountKey = "mykey"
+	val.args.containerName = "mycont"
+	val.args.transferDefStr = "s3-blockblob"
+
+	back := os.Getenv(s3AccessKeyEnvVar)
+
+	s3access := back
+	if s3access == "" {
+		s3access = "TEST"
+	}
+	os.Setenv(s3AccessKeyEnvVar, s3access)
+
+	defer os.Setenv(s3AccessKeyEnvVar, back)
+
+	back = os.Getenv(s3SecretKeyEnvVar)
+
+	s3key := back
+	if s3key == "" {
+		s3key = "TEST"
+	}
+	os.Setenv(s3SecretKeyEnvVar, s3access)
+
+	defer os.Setenv(s3SecretKeyEnvVar, back)
+
+	err := val.parseAndValidate()
+	assert.NoError(t, err, "unexpected error, all params should be set")
+	assert.Equal(t, val.args.storageAccountName, val.params.blobTarget.accountName, "account name is not set")
+	assert.Equal(t, val.args.storageAccountKey, val.params.blobTarget.accountKey, "account key is not set")
+	assert.Equal(t, val.args.containerName, val.params.blobTarget.container, "container is not set")
+	assert.Equal(t, url, val.params.s3Source.endpoint, "url/endpoint is invalid")
+	assert.Equal(t, bucket, val.params.s3Source.bucket, "bucket is invalid")
+	assert.Equal(t, s3access, val.params.s3Source.accessKey, "access key is invalid")
+	assert.Equal(t, s3key, val.params.s3Source.secretKey, "key is invalid")
+	assert.Equal(t, val.args.readTokenExp, val.params.s3Source.preSignedExpMin, "exp time invalid")
+}
+
+func TestS3TransferWithCustomExp(t *testing.T) {
+	val := newParamParserValidator()
+	url := "mys3.myurl.com"
+	bucket := "bucket"
+	val.args.sourceURIs = []string{fmt.Sprintf("s3://%v/%v", url, bucket)}
+	val.args.storageAccountName = "myaccount"
+	val.args.storageAccountKey = "mykey"
+	val.args.containerName = "mycont"
+	val.args.transferDefStr = "s3-blockblob"
+	val.args.readTokenExp = 10
+	
+	back := os.Getenv(s3AccessKeyEnvVar)
+
+	s3access := back
+	if s3access == "" {
+		s3access = "TEST"
+	}
+	os.Setenv(s3AccessKeyEnvVar, s3access)
+
+	defer os.Setenv(s3AccessKeyEnvVar, back)
+
+	back = os.Getenv(s3SecretKeyEnvVar)
+
+	s3key := back
+	if s3key == "" {
+		s3key = "TEST"
+	}
+	os.Setenv(s3SecretKeyEnvVar, s3access)
+
+	defer os.Setenv(s3SecretKeyEnvVar, back)
+
+	err := val.parseAndValidate()
+	assert.NoError(t, err, "unexpected error, all params should be set")
+	assert.Equal(t, val.args.storageAccountName, val.params.blobTarget.accountName, "account name is not set")
+	assert.Equal(t, val.args.storageAccountKey, val.params.blobTarget.accountKey, "account key is not set")
+	assert.Equal(t, val.args.containerName, val.params.blobTarget.container, "container is not set")
+	assert.Equal(t, url, val.params.s3Source.endpoint, "url/endpoint is invalid")
+	assert.Equal(t, bucket, val.params.s3Source.bucket, "bucket is invalid")
+	assert.Equal(t, s3access, val.params.s3Source.accessKey, "access key is invalid")
+	assert.Equal(t, s3key, val.params.s3Source.secretKey, "key is invalid")
+	assert.Equal(t, val.args.readTokenExp, val.params.s3Source.preSignedExpMin, "exp time invalid")
+}
+

args_test.go

@@ -13,22 +13,19 @@ import (
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/transfer"
 	"github.com/Azure/blobporter/util"
+	"github.com/Azure/blobporter/internal"
 )
 
-const programVersion = "0.6.07"
 
 var argsUtil paramParserValidator
 
 func init() {
 
 	//Show blobporter banner
-	fmt.Printf("BlobPorter \nCopyright (c) Microsoft Corporation. \nVersion: %v\n---------------\n", programVersion)
+	fmt.Printf("BlobPorter \nCopyright (c) Microsoft Corporation. \nVersion: %v\n---------------\n", internal.ProgramVersion)
 
 	argsUtil = newParamParserValidator()
 
-	// set user agent info
-	util.SetUserAgentInfo(programVersion)
-
 	const (
 		fileMsg                    = "Source URL, file or files (e.g. /data/*.gz) to upload."
 		nameMsg                    = "Blob name (e.g. myblob.txt) or prefix for download scenarios."
@@ -48,6 +45,7 @@ func init() {
 		removeDirStructureMsg      = "If set the directory structure from the source is not kept.\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"
+		readTokenExpMsg            = "Expiration in minutes of the read-only access token that will be generated to read from S3 or Azure Blob sources."
 	)
 
 	flag.Usage = func() {
@@ -69,7 +67,7 @@ func init() {
 		util.PrintUsageDefaults("i", "remove_directories", "false", removeDirStructureMsg)
 		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.StringListVarAlias(&argsUtil.args.sourceURIs, "f", "source_file", "", fileMsg)
@@ -90,6 +88,8 @@ func init() {
 	util.BoolVarAlias(&argsUtil.args.removeDirStructure, "i", "remove_directories", false, removeDirStructureMsg)
 	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)
+
 }
 
 var dataTransferred uint64