- client resets

Author: giventocode

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.10" // version number to show in help
+	programVersion           = "0.5.11a" // version number to show in help
 )
 
 const numOfWorkersFactor = 8

targets/azureblock.go

@@ -58,6 +58,7 @@ func (t *AzureBlock) CommitList(listInfo *pipeline.TargetCommittedListInfo, numb
 	//if the max retries is exceeded, panic will happen, hence no error is returned.
 	duration, _, _ := util.RetriableOperation(func(r int) error {
 		if err := t.StorageClient.PutBlockList(t.Container, targetName, blockList); err != nil {
+			t.resetClient()
 			return err
 		}
 		return nil
@@ -126,9 +127,8 @@ func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, sta
 				uint64(len(part.Data)),
 				bytes.NewReader(part.Data),
 				headers); err != nil {
-				if util.Verbose {
-					fmt.Printf("EH|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-				}
+				util.PrintfIfDebug("Error|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
+				t.resetClient()
 				return err
 			}
 			return nil
@@ -140,16 +140,17 @@ func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, sta
 			uint64(len(part.Data)),
 			part.NewBuffer(),
 			headers); err != nil {
-			if util.Verbose {
-				fmt.Printf("EH|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-			}
+			util.PrintfIfDebug("Error|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
+			t.resetClient()
 			return err
 		}
 
-		if util.Verbose {
-			fmt.Printf("OKA|S|%v|%v|%v|%v\n", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
-		}
+		util.PrintfIfDebug("OK|S|%v|%v|%v|%v", (*part).BlockID, len((*part).Data), (*part).TargetAlias, err)
 		return nil
 	})
 	return
 }
+
+func (t *AzureBlock) resetClient() {
+	t.StorageClient = util.GetBlobStorageClientWithNewHTTPClient(t.Creds.AccountName, t.Creds.AccountKey)
+}

targets/azurepage.go

@@ -47,6 +47,7 @@ func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error
 	//if the max retries is exceeded, panic will happen, hence no error is returned.
 	util.RetriableOperation(func(r int) error {
 		if err := (*t.StorageClient).PutPageBlob(t.Container, (*source).TargetAlias, size, nil); err != nil {
+			t.resetClient()
 			return err
 		}
 		return nil
@@ -90,6 +91,7 @@ func (t *AzurePage) WritePart(part *pipeline.Part) (duration time.Duration, star
 			if util.Verbose {
 				fmt.Printf("EH|S|%v|%v|%v|%v\n", part.Offset, len(part.Data), part.TargetAlias, err)
 			}
+			t.resetClient()
 			return err
 		}
 
@@ -101,3 +103,8 @@ func (t *AzurePage) WritePart(part *pipeline.Part) (duration time.Duration, star
 
 	return
 }
+
+func (t *AzurePage) resetClient() {
+	client := util.GetBlobStorageClientWithNewHTTPClient(t.Creds.AccountName, t.Creds.AccountKey)
+	t.StorageClient = &client
+}

util/util.go

@@ -290,6 +290,27 @@ func GetBlobStorageClient(accountName string, accountKey string) storage.BlobSto
 	return bc
 }
 
+//GetBlobStorageClientWithNewHTTPClient gets a storage client with a new instace of the HTTP client.
+func GetBlobStorageClientWithNewHTTPClient(accountName string, accountKey string) storage.BlobStorageClient {
+	var bc storage.BlobStorageClient
+	var client storage.Client
+	var err error
+
+	if accountName == "" || accountKey == "" {
+		log.Fatal("Storage account and/or key not specified via options or in environment variables ACCOUNT_NAME and ACCOUNT_KEY")
+	}
+
+	if client, err = storage.NewClient(accountName, accountKey, storage.DefaultBaseURL, LargeBlockAPIVersion, true); err != nil {
+		log.Fatal(err)
+	}
+
+	client.HTTPClient = NewHTTPClient()
+
+	bc = client.GetBlobService()
+
+	return bc
+}
+
 //GetBlobStorageClientWithSASToken gets a storage client with support for large block blobs
 func GetBlobStorageClientWithSASToken(accountName string, sasToken string) storage.BlobStorageClient {
 	var bc storage.BlobStorageClient
@@ -357,8 +378,8 @@ var storageHTTPClient *http.Client
 var HTTPClientTimeout = 600
 
 const (
-	maxIdleConns        = 100
-	maxIdleConnsPerHost = 100
+	maxIdleConns        = 50
+	maxIdleConnsPerHost = 50
 )
 
 func getStorageHTTPClient() *http.Client {
@@ -392,13 +413,21 @@ func getSuggestion(err error) string {
 		return "Try using a different container or upload and then delete a small blob with the same name."
 	case strings.Contains(err.Error(), "Client.Timeout"):
 		return "Try increasing the timeout using the -s option or reducing the number of workers and readers, options: -r and -g"
-	case strings.Contains(err.Error(),"too many open files"):
+	case strings.Contains(err.Error(), "too many open files"):
 		return "Try increasing the number of open files allowed. For debian systems you can try: ulimit -Sn 2048 "
 	default:
 		return ""
 	}
 }
 
+//PrintfIfDebug TODO
+func PrintfIfDebug(format string, values ...interface{}) {
+	if Verbose {
+		msg := fmt.Sprintf(format, values...)
+		fmt.Printf("%v\t%s\n", time.Now(), msg)
+	}
+}
+
 //GetFileNameFromURL returns last part of URL (filename)
 func GetFileNameFromURL(sourceURI string) (string, error) {