azure: Add new storage_create_container configuration property

In some cases account does not have permissions to read container properties or create one but has all permissions to do CRUD operations inside the container, e.g. SAS tokens. To solve such use case the new configuration property was added for the Azure object storage config which creates a new container explicitly by settings `storage_create_container` to `true`. To keep backward compatibility with existing object storage configurations the default value for it is always `true`.

Signed-off-by: Andrey Pleskach <ples@aiven.io>

providers/azure/azure.go

@@ -29,7 +29,8 @@ import (
 
 // DefaultConfig for Azure objstore client.
 var DefaultConfig = Config{
-	Endpoint: "blob.core.windows.net",
+	Endpoint:               "blob.core.windows.net",
+	StorageCreateContainer: true,
 	HTTPConfig: exthttp.HTTPConfig{
 		IdleConnTimeout:       model.Duration(90 * time.Second),
 		ResponseHeaderTimeout: model.Duration(2 * time.Minute),
@@ -47,6 +48,7 @@ type Config struct {
 	StorageAccountName      string             `yaml:"storage_account"`
 	StorageAccountKey       string             `yaml:"storage_account_key"`
 	StorageConnectionString string             `yaml:"storage_connection_string"`
+	StorageCreateContainer  bool               `yaml:"storage_create_container"`
 	ContainerName           string             `yaml:"container"`
 	Endpoint                string             `yaml:"endpoint"`
 	UserAssignedID          string             `yaml:"user_assigned_id"`
@@ -167,19 +169,20 @@ func NewBucketWithConfig(logger log.Logger, conf Config, component string) (*Buc
 	}
 
 	// Check if storage account container already exists, and create one if it does not.
-	ctx := context.Background()
-	_, err = containerClient.GetProperties(ctx, &container.GetPropertiesOptions{})
-	if err != nil {
-		if !bloberror.HasCode(err, bloberror.ContainerNotFound) {
-			return nil, err
-		}
-		_, err := containerClient.Create(ctx, nil)
+	if conf.StorageCreateContainer {
+		ctx := context.Background()
+		_, err = containerClient.GetProperties(ctx, &container.GetPropertiesOptions{})
 		if err != nil {
-			return nil, errors.Wrapf(err, "error creating Azure blob container: %s", conf.ContainerName)
+			if !bloberror.HasCode(err, bloberror.ContainerNotFound) {
+				return nil, err
+			}
+			_, err := containerClient.Create(ctx, nil)
+			if err != nil {
+				return nil, errors.Wrapf(err, "error creating Azure blob container: %s", conf.ContainerName)
+			}
+			level.Info(logger).Log("msg", "Azure blob container successfully created", "address", conf.ContainerName)
 		}
-		level.Info(logger).Log("msg", "Azure blob container successfully created", "address", conf.ContainerName)
 	}
-
 	bkt := &Bucket{
 		logger:           logger,
 		containerClient:  containerClient,