Resolving 500 server timeouts on append blob appends

common/fe-ste-models.go

@@ -1069,6 +1069,7 @@ const (
 	MaxAppendBlobBlockSize         = 4 * 1024 * 1024
 	DefaultPageBlobChunkSize       = 4 * 1024 * 1024
 	DefaultAzureFileChunkSize      = 4 * 1024 * 1024
+	MaxRangeGetSize                = 4 * 1024 * 1024
 	MaxNumberOfBlocksPerBlob       = 50000
 	BlockSizeThreshold             = 256 * 1024 * 1024
 	MinParallelChunkCountThreshold = 4 /* minimum number of chunks in parallel for AzCopy to be performant. */

ste/sender-appendBlob.go

@@ -21,12 +21,16 @@
 package ste
 
 import (
+	"bytes"
 	"context"
+	"crypto/md5"
 	"fmt"
-	"time"
-
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/appendblob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
+	"io"
+	"time"
 
 	"golang.org/x/sync/semaphore"
 
@@ -208,3 +212,57 @@ func (s *appendBlobSenderBase) GetDestinationLength() (int64, error) {
 	}
 	return *prop.ContentLength, nil
 }
+
+func (s *appendBlobSenderBase) GetMD5(offset, count int64) ([]byte, error) {
+	var rangeGetContentMD5 *bool
+	if count <= common.MaxRangeGetSize {
+		rangeGetContentMD5 = to.Ptr(true)
+	}
+	response, err := s.destAppendBlobClient.DownloadStream(s.jptm.Context(),
+		&blob.DownloadStreamOptions{
+			Range:              blob.HTTPRange{Offset: offset, Count: count},
+			RangeGetContentMD5: rangeGetContentMD5,
+			CPKInfo:            s.jptm.CpkInfo(),
+			CPKScopeInfo:       s.jptm.CpkScopeInfo(),
+		})
+	if err != nil {
+		return nil, err
+	}
+	if response.ContentMD5 != nil && len(response.ContentMD5) > 0 {
+		return response.ContentMD5, nil
+	} else {
+		// compute md5
+		body := response.NewRetryReader(s.jptm.Context(), &blob.RetryReaderOptions{MaxRetries: MaxRetryPerDownloadBody})
+		defer body.Close()
+		h := md5.New()
+		if _, err = io.Copy(h, body); err != nil {
+			return nil, err
+		}
+		return h.Sum(nil), nil
+	}
+}
+
+func (s *appendBlobSenderBase) transformAppendConditionMismatchError(timeoutFromCtx bool, offset, count int64, err error) (string, error) {
+	if err != nil && bloberror.HasCode(err, bloberror.AppendPositionConditionNotMet) && timeoutFromCtx {
+		if _, ok := s.sip.(benchmarkSourceInfoProvider); ok {
+			// If the source is a benchmark, then we don't need to check MD5 since the data is constantly changing.  This is OK.
+			return "", nil
+		}
+		// Download Range of last append
+		destMD5, destErr := s.GetMD5(offset, count)
+		if destErr != nil {
+			return ", get destination md5 after timeout", destErr
+		}
+		sourceMD5, sourceErr := s.sip.GetMD5(offset, count)
+		if sourceErr != nil {
+			return ", get source md5 after timeout", sourceErr
+		}
+		if destMD5 != nil && sourceMD5 != nil && len(destMD5) > 0 && len(sourceMD5) > 0 {
+			// Compare MD5
+			if bytes.Equal(destMD5, sourceMD5) {
+				return "", nil
+			}
+		}
+	}
+	return "", err
+}

ste/sender-appendBlobFromLocal.go

@@ -68,14 +68,18 @@ func (u *appendBlobUploader) GenerateUploadFunc(id common.ChunkID, blockIndex in
 		u.jptm.LogChunkStatus(id, common.EWaitReason.Body())
 		body := newPacedRequestBody(u.jptm.Context(), reader, u.pacer)
 		offset := id.OffsetInFile()
-		_, err := u.destAppendBlobClient.AppendBlock(u.jptm.Context(), body,
+		var timeoutFromCtx bool
+		ctx := withTimeoutNotification(u.jptm.Context(), &timeoutFromCtx)
+		_, err := u.destAppendBlobClient.AppendBlock(ctx, body,
 			&appendblob.AppendBlockOptions{
 				AppendPositionAccessConditions: &appendblob.AppendPositionAccessConditions{AppendPosition: &offset},
-				CPKInfo: u.jptm.CpkInfo(),
-				CPKScopeInfo: u.jptm.CpkScopeInfo(),
+				CPKInfo:                        u.jptm.CpkInfo(),
+				CPKScopeInfo:                   u.jptm.CpkScopeInfo(),
 			})
+		errString, err := u.transformAppendConditionMismatchError(timeoutFromCtx, offset, reader.Length(), err)
 		if err != nil {
-			u.jptm.FailActiveUpload("Appending block", err)
+			errString = "Appending block" + errString
+			u.jptm.FailActiveUpload(errString, err)
 			return
 		}
 	}

ste/sender-appendBlobFromURL.go

@@ -62,16 +62,20 @@ func (c *urlToAppendBlobCopier) GenerateCopyFunc(id common.ChunkID, blockIndex i
 			c.jptm.FailActiveS2SCopy("Getting source token credential", err)
 			return
 		}
-		_, err = c.destAppendBlobClient.AppendBlockFromURL(c.jptm.Context(), c.srcURL,
+		var timeoutFromCtx bool
+		ctx := withTimeoutNotification(c.jptm.Context(), &timeoutFromCtx)
+		_, err = c.destAppendBlobClient.AppendBlockFromURL(ctx, c.srcURL,
 			&appendblob.AppendBlockFromURLOptions{
-				Range: blob.HTTPRange{Offset: offset, Count: adjustedChunkSize},
+				Range:                          blob.HTTPRange{Offset: offset, Count: adjustedChunkSize},
 				AppendPositionAccessConditions: &appendblob.AppendPositionAccessConditions{AppendPosition: &offset},
-				CPKInfo: c.jptm.CpkInfo(),
-				CPKScopeInfo: c.jptm.CpkScopeInfo(),
-				CopySourceAuthorization: token,
+				CPKInfo:                        c.jptm.CpkInfo(),
+				CPKScopeInfo:                   c.jptm.CpkScopeInfo(),
+				CopySourceAuthorization:        token,
 			})
+		errString, err := c.transformAppendConditionMismatchError(timeoutFromCtx, offset, adjustedChunkSize, err)
 		if err != nil {
-			c.jptm.FailActiveS2SCopy("Appending block from URL", err)
+			errString = "Appending block from URL" + errString
+			c.jptm.FailActiveS2SCopy(errString, err)
 			return
 		}
 	}

ste/sender-appendBlob_test.go

@@ -0,0 +1,110 @@
+package ste
+
+import (
+	"context"
+	"fmt"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/appendblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
+	blobsas "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas"
+	blobservice "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
+	"github.com/Azure/azure-storage-azcopy/v10/common"
+	"github.com/stretchr/testify/assert"
+	"net/http"
+	"testing"
+	"time"
+)
+
+type appendErrorInjectionPolicy struct {
+	timedOut bool
+}
+
+func (r *appendErrorInjectionPolicy) Do(req *policy.Request) (*http.Response, error) {
+	if req.Raw().URL.Query().Get("comp") == "appendblock" && !r.timedOut {
+		req.Next()
+		var headers http.Header = make(map[string][]string)
+		headers.Add("x-ms-error-code", "OperationTimedOut")
+		r.timedOut = true
+		return &http.Response{StatusCode: 500, Header: headers}, nil
+	}
+	return req.Next()
+}
+
+func Test500FollowedBy412Logic(t *testing.T) {
+	a := assert.New(t)
+
+	// Setup source and destination
+	accountName, accountKey := getAccountAndKey()
+	rawURL := fmt.Sprintf("https://%s.blob.core.windows.net/", accountName)
+
+	credential, err := blob.NewSharedKeyCredential(accountName, accountKey)
+	a.Nil(err)
+
+	client, err := blobservice.NewClientWithSharedKeyCredential(rawURL, credential, &blobservice.ClientOptions{
+		ClientOptions: azcore.ClientOptions{
+			Transport: NewAzcopyHTTPClient(0),
+		}})
+	a.Nil(err)
+
+	cName := generateContainerName()
+	cc := client.NewContainerClient(cName)
+	_, err = cc.Create(context.Background(), nil)
+	a.Nil(err)
+	defer cc.Delete(context.Background(), nil)
+
+	sourceName := generateBlobName()
+	sourceClient := cc.NewBlockBlobClient(sourceName)
+	size := 1024 * 1024 * 10
+	dataReader, _ := getDataAndReader(t.Name(), size)
+	_, err = sourceClient.Upload(context.Background(), streaming.NopCloser(dataReader), nil)
+	a.Nil(err)
+
+	sasURL, err := cc.NewBlobClient(sourceName).GetSASURL(
+		blobsas.BlobPermissions{Read: true},
+		time.Now().Add(1*time.Hour),
+		nil)
+	a.Nil(err)
+
+	destName := generateBlobName()
+	destClient := cc.NewAppendBlobClient(destName)
+	destClient.Create(context.Background(), nil)
+	a.Nil(err)
+
+	jptm := testJobPartTransferManager{
+		info: to.Ptr(TransferInfo{
+			Source:       sasURL,
+			SrcContainer: cName,
+			SrcFilePath:  sourceName,
+		}),
+		fromTo: common.EFromTo.BlobBlob(),
+	}
+	blobSIP, err := newBlobSourceInfoProvider(jptm)
+	a.Nil(err)
+
+	injectionPolicy := &appendErrorInjectionPolicy{timedOut: false}
+	destClient, err = appendblob.NewClientWithSharedKeyCredential(destClient.URL(), credential, &appendblob.ClientOptions{
+		ClientOptions: azcore.ClientOptions{
+			PerRetryPolicies: []policy.Policy{newRetryNotificationPolicy(), injectionPolicy},
+			Transport:        NewAzcopyHTTPClient(0),
+		},
+	})
+	a.Nil(err)
+	base := appendBlobSenderBase{jptm: jptm, destAppendBlobClient: destClient, sip: blobSIP}
+
+	// Get MD5 range within service calculation
+	offset := int64(0)
+	count := int64(common.MaxRangeGetSize)
+	var timeoutFromCtx bool
+	ctx := withTimeoutNotification(context.Background(), &timeoutFromCtx)
+	_, err = base.destAppendBlobClient.AppendBlockFromURL(ctx, sasURL,
+		&appendblob.AppendBlockFromURLOptions{
+			Range:                          blob.HTTPRange{Offset: offset, Count: count},
+			AppendPositionAccessConditions: &appendblob.AppendPositionAccessConditions{AppendPosition: &offset},
+		})
+	errString, err := base.transformAppendConditionMismatchError(timeoutFromCtx, offset, count, err)
+	a.Nil(err)
+	a.Empty(errString)
+}

ste/sourceInfoProvider-Benchmark.go

@@ -21,6 +21,7 @@
 package ste
 
 import (
+	"errors"
 	"time"
 
 	"github.com/Azure/azure-storage-azcopy/v10/common"
@@ -57,3 +58,8 @@ func (b benchmarkSourceInfoProvider) GetFreshFileLastModifiedTime() (time.Time,
 func (b benchmarkSourceInfoProvider) EntityType() common.EntityType {
 	return common.EEntityType.File() // no folders in benchmark
 }
+
+// Tell append blob it doesn't matter if the MD5 is wrong since this data doesn't matter
+func (b benchmarkSourceInfoProvider) GetMD5(_, _ int64) ([]byte, error) {
+	return nil, errors.New("benchmark does not support retrieving MD5")
+}

ste/sourceInfoProvider-Blob.go

@@ -21,18 +21,18 @@
 package ste
 
 import (
-	"io"
-	"time"
-
+	"crypto/md5"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
 	"github.com/Azure/azure-storage-azcopy/v10/common"
+	"io"
+	"time"
 )
 
 // Source info provider for Azure blob
 type blobSourceInfoProvider struct {
 	defaultRemoteSourceInfoProvider
-	source               *blob.Client
+	source *blob.Client
 }
 
 func (p *blobSourceInfoProvider) IsDFSSource() bool {
@@ -147,3 +147,32 @@ func (p *blobSourceInfoProvider) GetFreshFileLastModifiedTime() (time.Time, erro
 	}
 	return common.IffNotNil(properties.LastModified, time.Time{}), nil
 }
+
+func (p *blobSourceInfoProvider) GetMD5(offset, count int64) ([]byte, error) {
+	var rangeGetContentMD5 *bool
+	if count <= common.MaxRangeGetSize {
+		rangeGetContentMD5 = to.Ptr(true)
+	}
+	response, err := p.source.DownloadStream(p.jptm.Context(),
+		&blob.DownloadStreamOptions{
+			Range:              blob.HTTPRange{Offset: offset, Count: count},
+			RangeGetContentMD5: rangeGetContentMD5,
+			CPKInfo:            p.jptm.CpkInfo(),
+			CPKScopeInfo:       p.jptm.CpkScopeInfo(),
+		})
+	if err != nil {
+		return nil, err
+	}
+	if response.ContentMD5 != nil && len(response.ContentMD5) > 0 {
+		return response.ContentMD5, nil
+	} else {
+		// compute md5
+		body := response.NewRetryReader(p.jptm.Context(), &blob.RetryReaderOptions{MaxRetries: MaxRetryPerDownloadBody})
+		defer body.Close()
+		h := md5.New()
+		if _, err = io.Copy(h, body); err != nil {
+			return nil, err
+		}
+		return h.Sum(nil), nil
+	}
+}

ste/sourceInfoProvider-File.go

@@ -22,11 +22,15 @@ package ste
 
 import (
 	"context"
+	"crypto/md5"
+	"fmt"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/directory"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/file"
+	"io"
 	"sync"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/directory"
-	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/file"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azfile/share"
 
 	"github.com/Azure/azure-storage-azcopy/v10/common"
@@ -180,10 +184,10 @@ func newFileSourceInfoProvider(jptm IJobPartTransferMgr) (ISourceInfoProvider, e
 	//TODO: Should we do that?
 	return &fileSourceInfoProvider{
 		defaultRemoteSourceInfoProvider: *base,
-		ctx: jptm.Context(),
-		cacheOnce: &sync.Once{},
-		srcShareClient: s.NewShareClient(jptm.Info().SrcContainer),
-		sourceURL: source.URL()}, nil
+		ctx:                             jptm.Context(),
+		cacheOnce:                       &sync.Once{},
+		srcShareClient:                  s.NewShareClient(jptm.Info().SrcContainer),
+		sourceURL:                       source.URL()}, nil
 }
 
 func (p *fileSourceInfoProvider) PreSignedSourceURL() (string, error) {
@@ -303,3 +307,42 @@ func (p *fileSourceInfoProvider) GetFreshFileLastModifiedTime() (time.Time, erro
 	// We ignore smblastwrite because otherwise the tx will fail s2s
 	return properties.LastModified(), nil
 }
+
+func (p *fileSourceInfoProvider) GetMD5(offset, count int64) ([]byte, error) {
+	switch p.EntityType() {
+	case common.EEntityType.File():
+		var rangeGetContentMD5 *bool
+		if count <= common.MaxRangeGetSize {
+			rangeGetContentMD5 = to.Ptr(true)
+		}
+		fsc, err := p.jptm.SrcServiceClient().FileServiceClient()
+		if err != nil {
+			return nil, err
+		}
+		shareClient := fsc.NewShareClient(p.transferInfo.SrcContainer)
+		fileClient := shareClient.NewRootDirectoryClient().NewFileClient(p.transferInfo.SrcFilePath)
+		response, err := fileClient.DownloadStream(p.ctx, &file.DownloadStreamOptions{
+			Range:              file.HTTPRange{Offset: offset, Count: count},
+			RangeGetContentMD5: rangeGetContentMD5,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if response.ContentMD5 != nil && len(response.ContentMD5) > 0 {
+			return response.ContentMD5, nil
+		} else {
+			// compute md5
+			body := response.NewRetryReader(p.jptm.Context(), &file.RetryReaderOptions{MaxRetries: MaxRetryPerDownloadBody})
+			defer body.Close()
+			h := md5.New()
+			if _, err = io.Copy(h, body); err != nil {
+				return nil, err
+			}
+			return h.Sum(nil), nil
+		}
+	case common.EEntityType.Folder():
+		return nil, fmt.Errorf("cannot get body or md5 of a folder")
+	default:
+		panic("unexpected case")
+	}
+}