Azure Blob Storage v2 state store (#3203)

Signed-off-by: ItalyPaleAle <[email protected]> Signed-off-by: Alessandro (Ale) Segala <[email protected]> Co-authored-by: Bernd Verst <[email protected]>
dapr · Nov 2, 2023 · 0d488c2 · 0d488c2
1 parent d6908e8
commit 0d488c2
Show file tree

Hide file tree

Showing 19 changed files with 329 additions and 99 deletions.
diff --git a/.github/scripts/test-info.mjs b/.github/scripts/test-info.mjs
@@ -509,8 +509,37 @@ const components = {
         conformanceDestroy: 'conformance-state.aws.dynamodb-destroy.sh',
         sourcePkg: 'state/aws/dynamodb',
     },
-    'state.azure.blobstorage': {
+    'state.azure.blobstorage.v2': {
+        conformance: true,
+        requiredSecrets: [
+            'AzureBlobStorageAccount',
+            'AzureBlobStorageAccessKey',
+            'AzureCertificationTenantId',
+            'AzureCertificationServicePrincipalClientId',
+            'AzureCertificationServicePrincipalClientSecret',
+            'AzureBlobStorageContainer',
+        ],
+        sourcePkg: [
+            'state/azure/blobstorage',
+            'internal/component/azure/blobstorage',
+        ],
+    },
+    'state.azure.blobstorage.v1': {
         conformance: true,
+        requiredSecrets: [
+            'AzureBlobStorageAccount',
+            'AzureBlobStorageAccessKey',
+            'AzureCertificationTenantId',
+            'AzureCertificationServicePrincipalClientId',
+            'AzureCertificationServicePrincipalClientSecret',
+            'AzureBlobStorageContainer',
+        ],
+        sourcePkg: [
+            'state/azure/blobstorage',
+            'internal/component/azure/blobstorage',
+        ],
+    },
+    'state.azure.blobstorage': {
         certification: true,
         requiredSecrets: [
             'AzureBlobStorageAccount',

diff --git a/state/azure/blobstorage/blobstorage.go → ...azure/blobstorage/internal/blobstorage.go b/state/azure/blobstorage/blobstorage.go → ...azure/blobstorage/internal/blobstorage.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Dapr Authors
+Copyright 2023 The Dapr Authors
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
@@ -11,36 +11,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-/*
-Azure Blob Storage state store.
-
-Sample configuration in yaml:
-
-	apiVersion: dapr.io/v1alpha1
-	kind: Component
-	metadata:
-	  name: statestore
-	spec:
-	  type: state.azure.blobstorage
-	  metadata:
-	  - name: accountName
-		value: <storage account name>
-	  - name: accountKey
-		value: <key>
-	  - name: containerName
-		value: <container Name>
-
-Concurrency is supported with ETags according to https://docs.microsoft.com/en-us/azure/storage/common/storage-concurrency#managing-concurrency-in-blob-storage
-*/
-
-package blobstorage
+package internal
 
 import (
 	"context"
 	"fmt"
 	"io"
 	"reflect"
-	"strings"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
@@ -56,18 +33,24 @@ import (
 	"github.com/dapr/kit/ptr"
 )
 
-const (
-	keyDelimiter = "||"
-)
-
 // StateStore Type.
 type StateStore struct {
 	state.BulkStore
 
+	getFileNameFn   func(string) string
 	containerClient *container.Client
 	logger          logger.Logger
 }
 
+func NewAzureBlobStorageStore(logger logger.Logger, getFileNameFn func(string) string) state.Store {
+	s := &StateStore{
+		logger:        logger,
+		getFileNameFn: getFileNameFn,
+	}
+	s.BulkStore = state.NewDefaultBulkStore(s)
+	return s
+}
+
 // Init the connection to blob storage, optionally creates a blob container if it doesn't exist.
 func (r *StateStore) Init(ctx context.Context, metadata state.Metadata) error {
 	var err error
@@ -100,7 +83,7 @@ func (r *StateStore) Set(ctx context.Context, req *state.SetRequest) error {
 
 func (r *StateStore) Ping(ctx context.Context) error {
 	if _, err := r.containerClient.GetProperties(ctx, nil); err != nil {
-		return fmt.Errorf("blob storage: error connecting to Blob storage at %s: %s", r.containerClient.URL(), err)
+		return fmt.Errorf("error connecting to Azure Blob Storage at '%s': %w", r.containerClient.URL(), err)
 	}
 
 	return nil
@@ -112,17 +95,8 @@ func (r *StateStore) GetComponentMetadata() (metadataInfo mdutils.MetadataMap) {
 	return
 }
 
-// NewAzureBlobStorageStore instance.
-func NewAzureBlobStorageStore(logger logger.Logger) state.Store {
-	s := &StateStore{
-		logger: logger,
-	}
-	s.BulkStore = state.NewDefaultBulkStore(s)
-	return s
-}
-
 func (r *StateStore) readFile(ctx context.Context, req *state.GetRequest) (*state.GetResponse, error) {
-	blockBlobClient := r.containerClient.NewBlockBlobClient(getFileName(req.Key))
+	blockBlobClient := r.containerClient.NewBlockBlobClient(r.getFileNameFn(req.Key))
 	blobDownloadResponse, err := blockBlobClient.DownloadStream(ctx, nil)
 	if err != nil {
 		if isNotFoundError(err) {
@@ -132,19 +106,16 @@ func (r *StateStore) readFile(ctx context.Context, req *state.GetRequest) (*stat
 		return &state.GetResponse{}, err
 	}
 
-	reader := blobDownloadResponse.Body
-	defer reader.Close()
-	blobData, err := io.ReadAll(reader)
+	defer blobDownloadResponse.Body.Close()
+	blobData, err := io.ReadAll(blobDownloadResponse.Body)
 	if err != nil {
-		return &state.GetResponse{}, fmt.Errorf("error reading az blob: %w", err)
+		return &state.GetResponse{}, fmt.Errorf("error reading blob: %w", err)
 	}
 
-	contentType := blobDownloadResponse.ContentType
-
 	return &state.GetResponse{
 		Data:        blobData,
 		ETag:        ptr.Of(string(*blobDownloadResponse.ETag)),
-		ContentType: contentType,
+		ContentType: blobDownloadResponse.ContentType,
 	}, nil
 }
 
@@ -158,22 +129,20 @@ func (r *StateStore) writeFile(ctx context.Context, req *state.SetRequest) error
 		modifiedAccessConditions.IfNoneMatch = ptr.Of(azcore.ETagAny)
 	}
 
-	accessConditions := blob.AccessConditions{
-		ModifiedAccessConditions: &modifiedAccessConditions,
-	}
-
 	blobHTTPHeaders, err := storageinternal.CreateBlobHTTPHeadersFromRequest(req.Metadata, req.ContentType, r.logger)
 	if err != nil {
 		return err
 	}
 
 	uploadOptions := azblob.UploadBufferOptions{
-		AccessConditions: &accessConditions,
-		Metadata:         storageinternal.SanitizeMetadata(r.logger, req.Metadata),
-		HTTPHeaders:      &blobHTTPHeaders,
+		AccessConditions: &blob.AccessConditions{
+			ModifiedAccessConditions: &modifiedAccessConditions,
+		},
+		Metadata:    storageinternal.SanitizeMetadata(r.logger, req.Metadata),
+		HTTPHeaders: &blobHTTPHeaders,
 	}
 
-	blockBlobClient := r.containerClient.NewBlockBlobClient(getFileName(req.Key))
+	blockBlobClient := r.containerClient.NewBlockBlobClient(r.getFileNameFn(req.Key))
 	_, err = blockBlobClient.UploadBuffer(ctx, r.marshal(req), &uploadOptions)
 
 	if err != nil {
@@ -182,14 +151,14 @@ func (r *StateStore) writeFile(ctx context.Context, req *state.SetRequest) error
 			return state.NewETagError(state.ETagMismatch, err)
 		}
 
-		return fmt.Errorf("error uploading az blob: %w", err)
+		return fmt.Errorf("error uploading blob: %w", err)
 	}
 
 	return nil
 }
 
 func (r *StateStore) deleteFile(ctx context.Context, req *state.DeleteRequest) error {
-	blockBlobClient := r.containerClient.NewBlockBlobClient(getFileName(req.Key))
+	blockBlobClient := r.containerClient.NewBlockBlobClient(r.getFileNameFn(req.Key))
 
 	modifiedAccessConditions := blob.ModifiedAccessConditions{}
 	if req.HasETag() {
@@ -218,25 +187,13 @@ func (r *StateStore) deleteFile(ctx context.Context, req *state.DeleteRequest) e
 	return nil
 }
 
-func getFileName(key string) string {
-	pr := strings.Split(key, keyDelimiter)
-	if len(pr) != 2 {
-		return pr[0]
-	}
-
-	return pr[1]
-}
-
 func (r *StateStore) marshal(req *state.SetRequest) []byte {
-	var v string
 	b, ok := req.Value.([]byte)
-	if ok {
-		v = string(b)
-	} else {
-		v, _ = jsoniter.MarshalToString(req.Value)
+	if !ok {
+		b, _ = jsoniter.Marshal(req.Value)
 	}
 
-	return []byte(v)
+	return b
 }
 
 func isNotFoundError(err error) bool {

diff --git a/state/azure/blobstorage/blobstorage_test.go → .../blobstorage/internal/blobstorage_test.go b/state/azure/blobstorage/blobstorage_test.go → .../blobstorage/internal/blobstorage_test.go
@@ -11,7 +11,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package blobstorage
+package internal
 
 import (
 	"context"
@@ -39,15 +39,3 @@ func TestInit(t *testing.T) {
 		assert.Equal(t, err, fmt.Errorf("missing or empty accountName field from metadata"))
 	})
 }
-
-func TestFileName(t *testing.T) {
-	t.Run("Valid composite key", func(t *testing.T) {
-		key := getFileName("app_id||key")
-		assert.Equal(t, "key", key)
-	})
-
-	t.Run("No delimiter present", func(t *testing.T) {
-		key := getFileName("key")
-		assert.Equal(t, "key", key)
-	})
-}
diff --git a/state/azure/blobstorage/metadata.yaml → state/azure/blobstorage/v1/metadata.yaml b/state/azure/blobstorage/metadata.yaml → state/azure/blobstorage/v1/metadata.yaml
@@ -1,4 +1,4 @@
-# yaml-language-server: $schema=../../../component-metadata-schema.json
+# yaml-language-server: $schema=../../../../component-metadata-schema.json
 schemaVersion: v1
 type: state
 name: azure.blobstorage

diff --git a/state/azure/blobstorage/v1/v1.go b/state/azure/blobstorage/v1/v1.go
@@ -0,0 +1,42 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package blobstorage
+
+import (
+	"strings"
+
+	"github.com/dapr/components-contrib/state"
+	"github.com/dapr/components-contrib/state/azure/blobstorage/internal"
+	"github.com/dapr/kit/logger"
+)
+
+const (
+	keyDelimiter = "||"
+)
+
+func getFileName(key string) string {
+	// This function splits the prefix, such as the Dapr app ID, from the key.
+	// This is wrong, as state stores should honor the prefix, but it is kept here for backwards-compatibility.
+	// The v2 of the component fixes thie behavior.
+	pr := strings.Split(key, keyDelimiter)
+	if len(pr) != 2 {
+		return pr[0]
+	}
+
+	return pr[1]
+}
+
+func NewAzureBlobStorageStore(log logger.Logger) state.Store {
+	return internal.NewAzureBlobStorageStore(log, getFileName)
+}
diff --git a/state/azure/blobstorage/v1/v1_test.go b/state/azure/blobstorage/v1/v1_test.go
@@ -0,0 +1,32 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package blobstorage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFileName(t *testing.T) {
+	t.Run("Valid composite key", func(t *testing.T) {
+		key := getFileName("app_id||key")
+		assert.Equal(t, "key", key)
+	})
+
+	t.Run("No delimiter present", func(t *testing.T) {
+		key := getFileName("key")
+		assert.Equal(t, "key", key)
+	})
+}