Migrate to azure-sdk-for-go

Makefile

@@ -46,6 +46,13 @@ swagger-install:
 	# Install swag
 	@test -f $(BINPATH)/swag || GOOS=linux GOARCH=amd64 go install github.com/swaggo/swag/cmd/swag@latest
 
+azurite-start:
+	azurite & \
+	echo $$! > ~/.azurite.pid
+
+azurite-stop:
+	kill `cat ~/.azurite.pid`
+
 swagger: swagger-install
 	# Generate swagger docs
 	@PATH=$(BINPATH)/:$(PATH) swag init --markdownFiles docs
@@ -160,10 +167,22 @@ docker-deb:  ## Build debian packages in docker container
 	@docker run -it --rm -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper dpkg DEBARCH=amd64
 
 docker-unit-test:  ## Run unit tests in docker container
-	@docker run -it --rm -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper test
+	@docker run -it --rm -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper \
+		azurite-start \
+		AZURE_STORAGE_ENDPOINT=http://127.0.0.1:10000/devstoreaccount1 \
+		AZURE_STORAGE_ACCOUNT=devstoreaccount1 \
+		AZURE_STORAGE_ACCESS_KEY="Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" \
+		test \
+		azurite-stop
 
 docker-system-test:  ## Run system tests in docker container (add TEST=t04_mirror or TEST=UpdateMirror26Test to run only specific tests)
-	@docker run -it --rm -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper system-test TEST=$(TEST)
+	@docker run -it --rm -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper \
+		azurite-start \
+		AZURE_STORAGE_ENDPOINT=http://127.0.0.1:10000/devstoreaccount1 \
+		AZURE_STORAGE_ACCOUNT=devstoreaccount1 \
+		AZURE_STORAGE_ACCESS_KEY="Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" \
+		system-test TEST=$(TEST) \
+		azurite-stop
 
 docker-serve:  ## Run development server (auto recompiling) on http://localhost:3142
 	@docker run -it --rm -p 3142:3142 -v ${PWD}:/work/src aptly-dev /work/src/system/docker-wrapper serve || true

azure/azure.go

@@ -5,28 +5,35 @@
 import (
 	"context"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
-	"net/url"
+	"os"
 	"path/filepath"
 	"time"
 
-	"github.com/Azure/azure-storage-blob-go/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
 	"github.com/aptly-dev/aptly/aptly"
 )
 
 func isBlobNotFound(err error) bool {
-	storageError, ok := err.(azblob.StorageError)
-	return ok && storageError.ServiceCode() == azblob.ServiceCodeBlobNotFound
+	var respErr *azcore.ResponseError
+	if errors.As(err, &respErr) {
+		return respErr.StatusCode == 404 // BlobNotFound
+	}
+	return false
 }
 
 type azContext struct {
-	container azblob.ContainerURL
+	client    *azblob.Client
+	container string
 	prefix    string
 }
 
 func newAzContext(accountName, accountKey, container, prefix, endpoint string) (*azContext, error) {
-	credential, err := azblob.NewSharedKeyCredential(accountName, accountKey)
+	cred, err := azblob.NewSharedKeyCredential(accountName, accountKey)
 	if err != nil {
 		return nil, err
 	}
@@ -35,15 +42,14 @@
 		endpoint = fmt.Sprintf("https://%s.blob.core.windows.net", accountName)
 	}
 
-	url, err := url.Parse(fmt.Sprintf("%s/%s", endpoint, container))
+	serviceClient, err := azblob.NewClientWithSharedKeyCredential(endpoint, cred, nil)
 	if err != nil {
 		return nil, err
 	}
 
-	containerURL := azblob.NewContainerURL(*url, azblob.NewPipeline(credential, azblob.PipelineOptions{}))
-
 	result := &azContext{
-		container: containerURL,
+		client:    serviceClient,
+		container: container,
 		prefix:    prefix,
 	}
 
@@ -54,10 +60,6 @@
 	return filepath.Join(az.prefix, path)
 }
 
-func (az *azContext) blobURL(path string) azblob.BlobURL {
-	return az.container.NewBlobURL(az.blobPath(path))
-}
-
 func (az *azContext) internalFilelist(prefix string, progress aptly.Progress) (paths []string, md5s []string, err error) {
 	const delimiter = "/"
 	paths = make([]string, 0, 1024)
@@ -67,27 +69,33 @@
 		prefix += delimiter
 	}
 
-	for marker := (azblob.Marker{}); marker.NotDone(); {
-		listBlob, err := az.container.ListBlobsFlatSegment(
-			context.Background(), marker, azblob.ListBlobsSegmentOptions{
-				Prefix:     prefix,
-				MaxResults: 1,
-				Details:    azblob.BlobListingDetails{Metadata: true}})
+	ctx := context.Background()
+	maxResults := int32(1)
+	pager := az.client.NewListBlobsFlatPager(az.container, &azblob.ListBlobsFlatOptions{
+		Prefix:     &prefix,
+		MaxResults: &maxResults,
+		Include:    azblob.ListBlobsInclude{Metadata: true},
+	})
+
+	// Iterate over each page
+	for pager.More() {
+		page, err := pager.NextPage(ctx)
 		if err != nil {
 			return nil, nil, fmt.Errorf("error listing under prefix %s in %s: %s", prefix, az, err)
 		}
 
-		marker = listBlob.NextMarker
-
-		for _, blob := range listBlob.Segment.BlobItems {
+		for _, blob := range page.Segment.BlobItems {
 			if prefix == "" {
-				paths = append(paths, blob.Name)
+				paths = append(paths, *blob.Name)
 			} else {
-				paths = append(paths, blob.Name[len(prefix):])
+				name := *blob.Name
+				paths = append(paths, name[len(prefix):])
 			}
-			md5s = append(md5s, fmt.Sprintf("%x", blob.Properties.ContentMD5))
-		}
+			b := *blob
+			md5 := b.Properties.ContentMD5
+			md5s = append(md5s, fmt.Sprintf("%x", md5))
 
+		}
 		if progress != nil {
 			time.Sleep(time.Duration(500) * time.Millisecond)
 			progress.AddBar(1)
@@ -97,28 +105,27 @@
 	return paths, md5s, nil
 }
 
-func (az *azContext) putFile(blob azblob.BlobURL, source io.Reader, sourceMD5 string) error {
-	uploadOptions := azblob.UploadStreamToBlockBlobOptions{
-		BufferSize: 4 * 1024 * 1024,
-		MaxBuffers: 8,
+func (az *azContext) putFile(blobName string, source io.Reader, sourceMD5 string) error {
+	uploadOptions := &azblob.UploadFileOptions{
+		BlockSize:   4 * 1024 * 1024,
+		Concurrency: 8,
 	}
 
+	path := az.blobPath(blobName)
 	if len(sourceMD5) > 0 {
 		decodedMD5, err := hex.DecodeString(sourceMD5)
 		if err != nil {
 			return err
 		}
-		uploadOptions.BlobHTTPHeaders = azblob.BlobHTTPHeaders{
-			ContentMD5: decodedMD5,
+		uploadOptions.HTTPHeaders = &blob.HTTPHeaders{
+			BlobContentMD5: decodedMD5,
 		}
 	}
 
-	_, err := azblob.UploadStreamToBlockBlob(
-		context.Background(),
-		source,
-		blob.ToBlockBlobURL(),
-		uploadOptions,
-	)
+	var err error
+	if file, ok := source.(*os.File); ok {
+		_, err = az.client.UploadFile(context.TODO(), az.container, path, file, uploadOptions)
+	}
 
 	return err
 }

azure/package_pool.go

@@ -5,7 +5,6 @@
 	"os"
 	"path/filepath"
 
-	"github.com/Azure/azure-storage-blob-go/azblob"
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/utils"
 	"github.com/pkg/errors"
@@ -41,19 +40,15 @@
 	return filepath.Join(hash[0:2], hash[2:4], hash[4:32]+"_"+filename)
 }
 
-func (pool *PackagePool) ensureChecksums(
-	poolPath string,
-	checksumStorage aptly.ChecksumStorage,
-) (*utils.ChecksumInfo, error) {
+func (pool *PackagePool) ensureChecksums(poolPath string, checksumStorage aptly.ChecksumStorage) (*utils.ChecksumInfo, error) {
 	targetChecksums, err := checksumStorage.Get(poolPath)
 	if err != nil {
 		return nil, err
 	}
 
 	if targetChecksums == nil {
 		// we don't have checksums stored yet for this file
-		blob := pool.az.blobURL(poolPath)
-		download, err := blob.Download(context.Background(), 0, 0, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
+		download, err := pool.az.client.DownloadStream(context.Background(), pool.az.container, poolPath, nil)
 		if err != nil {
 			if isBlobNotFound(err) {
 				return nil, nil
@@ -63,7 +58,7 @@
 		}
 
 		targetChecksums = &utils.ChecksumInfo{}
-		*targetChecksums, err = utils.ChecksumsForReader(download.Body(azblob.RetryReaderOptions{}))
+		*targetChecksums, err = utils.ChecksumsForReader(download.Body)
 		if err != nil {
 			return nil, errors.Wrapf(err, "error checksumming blob at %s", poolPath)
 		}
@@ -92,46 +87,49 @@
 }
 
 func (pool *PackagePool) Size(path string) (int64, error) {
-	blob := pool.az.blobURL(path)
-	props, err := blob.GetProperties(context.Background(), azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{})
+	serviceClient := pool.az.client.ServiceClient()
+	containerClient := serviceClient.NewContainerClient(pool.az.container)
+	blobClient := containerClient.NewBlobClient(path)
+
+	props, err := blobClient.GetProperties(context.TODO(), nil)
 	if err != nil {
 		return 0, errors.Wrapf(err, "error examining %s from %s", path, pool)
 	}
 
-	return props.ContentLength(), nil
+	return *props.ContentLength, nil
 }
 
 func (pool *PackagePool) Open(path string) (aptly.ReadSeekerCloser, error) {
-	blob := pool.az.blobURL(path)
-
 	temp, err := os.CreateTemp("", "blob-download")
 	if err != nil {
-		return nil, errors.Wrap(err, "error creating temporary file for blob download")
+		return nil, errors.Wrapf(err, "error creating tempfile for %s", path)
 	}
-
 	defer os.Remove(temp.Name())
 
-	err = azblob.DownloadBlobToFile(context.Background(), blob, 0, 0, temp, azblob.DownloadFromBlobOptions{})
+	_, err = pool.az.client.DownloadFile(context.TODO(), pool.az.container, path, temp, nil)
 	if err != nil {
-		return nil, errors.Wrapf(err, "error downloading blob at %s", path)
+		return nil, errors.Wrapf(err, "error downloading blob %s", path)
 	}
 
 	return temp, nil
 }
 
 func (pool *PackagePool) Remove(path string) (int64, error) {
-	blob := pool.az.blobURL(path)
-	props, err := blob.GetProperties(context.Background(), azblob.BlobAccessConditions{}, azblob.ClientProvidedKeyOptions{})
+	serviceClient := pool.az.client.ServiceClient()
+	containerClient := serviceClient.NewContainerClient(pool.az.container)
+	blobClient := containerClient.NewBlobClient(path)
+
+	props, err := blobClient.GetProperties(context.TODO(), nil)
 	if err != nil {
-		return 0, errors.Wrapf(err, "error getting props of %s from %s", path, pool)
+		return 0, errors.Wrapf(err, "error examining %s from %s", path, pool)
 	}
 
-	_, err = blob.Delete(context.Background(), azblob.DeleteSnapshotsOptionNone, azblob.BlobAccessConditions{})
+	_, err = pool.az.client.DeleteBlob(context.Background(), pool.az.container, path, nil)
 	if err != nil {
 		return 0, errors.Wrapf(err, "error deleting %s from %s", path, pool)
 	}
 
-	return props.ContentLength(), nil
+	return *props.ContentLength, nil
 }
 
 func (pool *PackagePool) Import(srcPath, basename string, checksums *utils.ChecksumInfo, _ bool, checksumStorage aptly.ChecksumStorage) (string, error) {
@@ -145,7 +143,6 @@
 	}
 
 	path := pool.buildPoolPath(basename, checksums)
-	blob := pool.az.blobURL(path)
 	targetChecksums, err := pool.ensureChecksums(path, checksumStorage)
 	if err != nil {
 		return "", err
@@ -161,7 +158,7 @@
 	}
 	defer source.Close()
 
-	err = pool.az.putFile(blob, source, checksums.MD5)
+	err = pool.az.putFile(path, source, checksums.MD5)
 	if err != nil {
 		return "", err
 	}

azure/package_pool_test.go

@@ -7,7 +7,7 @@ import (
 	"path/filepath"
 	"runtime"
 
-	"github.com/Azure/azure-storage-blob-go/azblob"
+        "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 	"github.com/aptly-dev/aptly/aptly"
 	"github.com/aptly-dev/aptly/files"
 	"github.com/aptly-dev/aptly/utils"
@@ -50,8 +50,10 @@ func (s *PackagePoolSuite) SetUpTest(c *C) {
 
 	s.pool, err = NewPackagePool(s.accountName, s.accountKey, container, "", s.endpoint)
 	c.Assert(err, IsNil)
-	cnt := s.pool.az.container
-	_, err = cnt.Create(context.Background(), azblob.Metadata{}, azblob.PublicAccessContainer)
+        publicAccessType := azblob.PublicAccessTypeContainer
+        _, err = s.pool.az.client.CreateContainer(context.TODO(), s.pool.az.container, &azblob.CreateContainerOptions{
+            Access: &publicAccessType,
+        })
 	c.Assert(err, IsNil)
 
 	s.prefixedPool, err = NewPackagePool(s.accountName, s.accountKey, container, prefix, s.endpoint)