func NewAzure(name string, info map[string]string) (Backend, error) {
	b := &azureBackend{
		name:      name,
		container: info["container"],
	}
	accountName := info["account_name"]
	accountKey := info["account_key"]

	if b.container == "" {
		return nil, fmt.Errorf("blobstore: missing Azure Storage container param for %s", name)
	}
	if accountName == "" {
		return nil, fmt.Errorf("blobstore: missing Azure Storage account_name param for %s", name)
	}
	if accountKey == "" {
		return nil, fmt.Errorf("blobstore: missing Azure Storage account_key param for %s", name)
	}

	client, err := storage.NewBasicClient(accountName, accountKey)
	if err != nil {
		return nil, fmt.Errorf("blobstore: error creating Azure Storage client %s: %s", name, err)
	}
	b.client = client.GetBlobService()

	ok, err := b.client.ContainerExists(b.container)
	if err != nil {
		return nil, fmt.Errorf("blobstore: error checking if Azure Storage container %q exists for %s: %s", b.container, name, err)
	}
	if !ok {
		return nil, fmt.Errorf("blobstore: Azure Storage container %q does not exists for %s", b.container, name)
	}

	return b, nil
}

func TestAzureBackend(t *testing.T) {
	if os.Getenv("AZURE_ACCOUNT_NAME") == "" ||
		os.Getenv("AZURE_ACCOUNT_KEY") == "" {
		t.SkipNow()
	}

	accountName := os.Getenv("AZURE_ACCOUNT_NAME")
	accountKey := os.Getenv("AZURE_ACCOUNT_KEY")

	ts := time.Now().UnixNano()
	container := fmt.Sprintf("vault-test-%d", ts)

	cleanupClient, _ := storage.NewBasicClient(accountName, accountKey)

	logger := log.New(os.Stderr, "", log.LstdFlags)
	backend, err := NewBackend("azure", logger, map[string]string{
		"container":   container,
		"accountName": accountName,
		"accountKey":  accountKey,
	})

	defer func() {
		cleanupClient.GetBlobService().DeleteContainerIfExists(container)
	}()

	if err != nil {
		t.Fatalf("err: %s", err)
	}

	testBackend(t, backend)
	testBackend_ListPrefix(t, backend)
}

func NewAzureClient(subscriptionID string, resourceGroupName string, storageAccountName string, servicePrincipalToken *azure.ServicePrincipalToken) (*AzureClient, error) {
	var azureClient = &AzureClient{}

	azureClient.DeploymentsClient = resources.NewDeploymentsClient(subscriptionID)
	azureClient.DeploymentsClient.Authorizer = servicePrincipalToken

	azureClient.GroupsClient = resources.NewGroupsClient(subscriptionID)
	azureClient.GroupsClient.Authorizer = servicePrincipalToken

	azureClient.PublicIPAddressesClient = network.NewPublicIPAddressesClient(subscriptionID)
	azureClient.PublicIPAddressesClient.Authorizer = servicePrincipalToken

	azureClient.VirtualMachinesClient = compute.NewVirtualMachinesClient(subscriptionID)
	azureClient.VirtualMachinesClient.Authorizer = servicePrincipalToken

	storageAccountsClient := armStorage.NewAccountsClient(subscriptionID)
	storageAccountsClient.Authorizer = servicePrincipalToken

	accountKeys, err := storageAccountsClient.ListKeys(resourceGroupName, storageAccountName)
	if err != nil {
		return nil, err
	}

	storageClient, err := storage.NewBasicClient(storageAccountName, *accountKeys.Key1)
	if err != nil {
		return nil, err
	}

	azureClient.BlobStorageClient = storageClient.GetBlobService()
	return azureClient, nil
}

func uploadBlob(cl ExtensionsClient, storageAccount, packagePath string) (string, error) {
	// Fetch keys for storage account
	svc := storageservice.NewClient(cl.client)
	keys, err := svc.GetStorageServiceKeys(storageAccount)
	if err != nil {
		return "", fmt.Errorf("Could not fetch keys for storage account. Make sure it is in publisher subscription. Error: %v", err)
	}
	log.Debug("Retrieved storage account keys.")

	// Read package
	pkg, err := os.OpenFile(packagePath, os.O_RDONLY, 0777)
	if err != nil {
		return "", fmt.Errorf("Could not reach package file: %v", err)
	}
	stat, err := pkg.Stat()
	if err != nil {
		return "", fmt.Errorf("Could not stat the package file: %v", err)
	}
	defer pkg.Close()

	// Upload blob
	sc, err := storage.NewBasicClient(storageAccount, keys.PrimaryKey)
	if err != nil {
		return "", fmt.Errorf("Could not create storage client: %v", err)
	}
	bs := sc.GetBlobService()
	if _, err := bs.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob); err != nil {
		return "", fmt.Errorf("Error creating blob container: %v", err)
	}
	blobName := fmt.Sprintf("%d.zip", time.Now().Unix())
	if err := bs.CreateBlockBlobFromReader(containerName, blobName, uint64(stat.Size()), pkg, nil); err != nil {
		return "", fmt.Errorf("Error uploading blob: %v", err)
	}
	return bs.GetBlobURL(containerName, blobName), nil
}

func getBlobClient() (storage.BlobStorageClient, error) {
	api, err := storage.NewBasicClient(c.Conf.AzureAccount, c.Conf.AzureKey)
	if err != nil {
		return storage.BlobStorageClient{}, err
	}
	return api.GetBlobService(), nil
}

func NewAzureClient(subscriptionID, resourceGroupName, storageAccountName string,
	servicePrincipalToken, servicePrincipalTokenVault *azure.ServicePrincipalToken) (*AzureClient, error) {

	var azureClient = &AzureClient{}

	maxlen := getInspectorMaxLength()

	azureClient.DeploymentsClient = resources.NewDeploymentsClient(subscriptionID)
	azureClient.DeploymentsClient.Authorizer = servicePrincipalToken
	azureClient.DeploymentsClient.RequestInspector = withInspection(maxlen)
	azureClient.DeploymentsClient.ResponseInspector = byInspecting(maxlen)
	azureClient.DeploymentsClient.UserAgent += packerUserAgent

	azureClient.GroupsClient = resources.NewGroupsClient(subscriptionID)
	azureClient.GroupsClient.Authorizer = servicePrincipalToken
	azureClient.GroupsClient.RequestInspector = withInspection(maxlen)
	azureClient.GroupsClient.ResponseInspector = byInspecting(maxlen)
	azureClient.GroupsClient.UserAgent += packerUserAgent

	azureClient.PublicIPAddressesClient = network.NewPublicIPAddressesClient(subscriptionID)
	azureClient.PublicIPAddressesClient.Authorizer = servicePrincipalToken
	azureClient.PublicIPAddressesClient.RequestInspector = withInspection(maxlen)
	azureClient.PublicIPAddressesClient.ResponseInspector = byInspecting(maxlen)
	azureClient.PublicIPAddressesClient.UserAgent += packerUserAgent

	azureClient.VirtualMachinesClient = compute.NewVirtualMachinesClient(subscriptionID)
	azureClient.VirtualMachinesClient.Authorizer = servicePrincipalToken
	azureClient.VirtualMachinesClient.RequestInspector = withInspection(maxlen)
	azureClient.VirtualMachinesClient.ResponseInspector = byConcatDecorators(byInspecting(maxlen), templateCapture(azureClient))
	azureClient.VirtualMachinesClient.UserAgent += packerUserAgent

	azureClient.AccountsClient = armStorage.NewAccountsClient(subscriptionID)
	azureClient.AccountsClient.Authorizer = servicePrincipalToken
	azureClient.AccountsClient.RequestInspector = withInspection(maxlen)
	azureClient.AccountsClient.ResponseInspector = byInspecting(maxlen)
	azureClient.AccountsClient.UserAgent += packerUserAgent

	azureClient.VaultClient = common.VaultClient{}
	azureClient.VaultClient.Authorizer = servicePrincipalTokenVault
	azureClient.VaultClient.RequestInspector = withInspection(maxlen)
	azureClient.VaultClient.ResponseInspector = byInspecting(maxlen)
	azureClient.VaultClient.UserAgent += packerUserAgent

	accountKeys, err := azureClient.AccountsClient.ListKeys(resourceGroupName, storageAccountName)
	if err != nil {
		return nil, err
	}

	storageClient, err := storage.NewBasicClient(storageAccountName, *accountKeys.Key1)
	if err != nil {
		return nil, err
	}

	azureClient.BlobStorageClient = storageClient.GetBlobService()
	return azureClient, nil
}

func setup(t *testing.T, conf map[string]string) {
	creds, err := getCredentialsFromConf(conf)
	if err != nil {
		t.Fatalf("Error getting credentials from conf: %v", err)
	}
	rivieraClient, err := getRivieraClient(creds)
	if err != nil {
		t.Fatalf("Error instantiating the riviera client: %v", err)
	}

	// Create resource group
	r := rivieraClient.NewRequest()
	r.Command = riviera.CreateResourceGroup{
		Name:     conf["resource_group_name"],
		Location: riviera.WestUS,
	}
	response, err := r.Execute()
	if err != nil {
		t.Fatalf("Error creating a resource group: %v", err)
	}
	if !response.IsSuccessful() {
		t.Fatalf("Error creating a resource group: %v", response.Error.Error())
	}

	// Create storage account
	r = rivieraClient.NewRequest()
	r.Command = storage.CreateStorageAccount{
		ResourceGroupName: conf["resource_group_name"],
		Name:              conf["storage_account_name"],
		AccountType:       riviera.String("Standard_LRS"),
		Location:          riviera.WestUS,
	}
	response, err = r.Execute()
	if err != nil {
		t.Fatalf("Error creating a storage account: %v", err)
	}
	if !response.IsSuccessful() {
		t.Fatalf("Error creating a storage account: %v", response.Error.Error())
	}

	// Create container
	accessKey, err := getStorageAccountAccessKey(conf, conf["resource_group_name"], conf["storage_account_name"])
	if err != nil {
		t.Fatalf("Error creating a storage account: %v", err)
	}
	storageClient, err := mainStorage.NewBasicClient(conf["storage_account_name"], accessKey)
	if err != nil {
		t.Fatalf("Error creating storage client for storage account %q: %s", conf["storage_account_name"], err)
	}
	blobClient := storageClient.GetBlobService()
	_, err = blobClient.CreateContainerIfNotExists(conf["container_name"], mainStorage.ContainerAccessTypePrivate)
	if err != nil {
		t.Fatalf("Couldn't create container with name %s: %s.", conf["container_name"], err)
	}
}

// Create a new StorageClient object based on a configuration file.
func (c *Config) NewClient() (dialects.StorageClient, error) {
	serviceClient, err := storage.NewBasicClient(c.Account, c.AccessKey)
	if err != nil {
		return nil, err
	}
	return &QueueStorage{
		Account:   c.Account,
		AccessKey: c.AccessKey,
		QueueName: c.QueueName,
		Client:    serviceClient.GetQueueService()}, nil
}

// newAzureBackend constructs an Azure backend using a pre-existing
// bucket. Credentials can be provided to the backend, sourced
// from the environment, AWS credential files or by IAM role.
func newAzureBackend(conf map[string]string, logger log.Logger) (Backend, error) {

	container := os.Getenv("AZURE_BLOB_CONTAINER")
	if container == "" {
		container = conf["container"]
		if container == "" {
			return nil, fmt.Errorf("'container' must be set")
		}
	}

	accountName := os.Getenv("AZURE_ACCOUNT_NAME")
	if accountName == "" {
		accountName = conf["accountName"]
		if accountName == "" {
			return nil, fmt.Errorf("'accountName' must be set")
		}
	}

	accountKey := os.Getenv("AZURE_ACCOUNT_KEY")
	if accountKey == "" {
		accountKey = conf["accountKey"]
		if accountKey == "" {
			return nil, fmt.Errorf("'accountKey' must be set")
		}
	}

	client, err := storage.NewBasicClient(accountName, accountKey)

	if err != nil {
		return nil, fmt.Errorf("Failed to create Azure client: %v", err)
	}

	client.GetBlobService().CreateContainerIfNotExists(container, storage.ContainerAccessTypePrivate)

	maxParStr, ok := conf["max_parallel"]
	var maxParInt int
	if ok {
		maxParInt, err = strconv.Atoi(maxParStr)
		if err != nil {
			return nil, errwrap.Wrapf("failed parsing max_parallel parameter: {{err}}", err)
		}
		if logger.IsDebug() {
			logger.Debug("azure: max_parallel set", "max_parallel", maxParInt)
		}
	}

	a := &AzureBackend{
		container:  container,
		client:     client.GetBlobService(),
		logger:     logger,
		permitPool: NewPermitPool(maxParInt),
	}
	return a, nil
}

func New(config Config) (result *comm, err error) {
	storageClient, err := storage.NewBasicClient(config.StorageAccountName, config.StorageAccountKey)
	if err != nil {
		return nil, err
	}
	config.blobClient = storageClient.GetBlobService()

	result = &comm{
		config: config,
	}
	return
}

func (armClient *ArmClient) getQueueServiceClientForStorageAccount(resourceGroupName, storageAccountName string) (*mainStorage.QueueServiceClient, error) {
	key, err := armClient.getKeyForStorageAccount(resourceGroupName, storageAccountName)
	if err != nil {
		return nil, err
	}

	storageClient, err := mainStorage.NewBasicClient(storageAccountName, key)
	if err != nil {
		return nil, fmt.Errorf("Error creating storage client for storage account %q: %s", storageAccountName, err)
	}

	queueClient := storageClient.GetQueueService()
	return &queueClient, nil
}

// getStorageClientForStorageService is helper method which returns the
// storage.Client associated to the given storage service name.
func (c Client) getStorageClientForStorageService(serviceName string) (storage.Client, error) {
	var storageClient storage.Client

	keys, err := c.storageServiceClient.GetStorageServiceKeys(serviceName)
	if err != nil {
		return storageClient, fmt.Errorf("Failed getting Storage Service keys for %s: %s", serviceName, err)
	}

	storageClient, err = storage.NewBasicClient(serviceName, keys.PrimaryKey)
	if err != nil {
		return storageClient, fmt.Errorf("Failed creating Storage Service client for %s: %s", serviceName, err)
	}

	return storageClient, err
}

func initAzureStorage() error {
	var storageKey = os.Getenv("AZURE_STORAGE_ACCESS_KEY")
	var storageName = os.Getenv("AZURE_STORAGE_ACCOUNT")
	client, err := storage.NewBasicClient(storageName, storageKey)
	if err != nil {
		return err
	}

	blobService = client.GetBlobService()
	_, err = blobService.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypePrivate)
	if err != nil {
		return err
	}

	return nil
}

func (armClient *ArmClient) getBlobStorageClientForStorageAccount(resourceGroupName, storageAccountName string) (*mainStorage.BlobStorageClient, bool, error) {
	key, accountExists, err := armClient.getKeyForStorageAccount(resourceGroupName, storageAccountName)
	if err != nil {
		return nil, accountExists, err
	}
	if accountExists == false {
		return nil, false, nil
	}

	storageClient, err := mainStorage.NewBasicClient(storageAccountName, key)
	if err != nil {
		return nil, true, fmt.Errorf("Error creating storage client for storage account %q: %s", storageAccountName, err)
	}

	blobClient := storageClient.GetBlobService()
	return &blobClient, true, nil
}

// Create a new StorageClient object based on a configuration file.
func (c *Config) NewClient() (dialects.StorageClient, error) {
	serviceClient, err := storage.NewBasicClient(c.Account, c.AccessKey)
	if err != nil {
		return nil, err
	}
	converterFunction, err := dialects.GetBatchConverterFunction(c.FileFormat)
	if err != nil {
		return nil, err
	}
	return &BlobStorage{
		Account:        c.Account,
		AccessKey:      c.AccessKey,
		BlobPath:       c.BlobPath,
		Container:      c.Container,
		FileFormat:     c.FileFormat,
		BatchConverter: converterFunction,
		Client:         serviceClient.GetBlobService()}, nil
}

func newVolumeDriver(accountName, accountKey, mountpoint, metadataRoot string, removeShares bool) (*volumeDriver, error) {
	storageClient, err := azure.NewBasicClient(accountName, accountKey)
	if err != nil {
		return nil, fmt.Errorf("error creating azure client: %v", err)
	}
	metaDriver, err := newMetadataDriver(metadataRoot)
	if err != nil {
		return nil, fmt.Errorf("cannot initialize metadata driver: %v", err)
	}
	return &volumeDriver{
		cl:           storageClient.GetFileService(),
		meta:         metaDriver,
		accountName:  accountName,
		accountKey:   accountKey,
		mountpoint:   mountpoint,
		removeShares: removeShares,
	}, nil
}

func (c *AzureClient) createContainer(storageAccountName, primaryAccessKey, containerName string, containerAccessType storageclient.ContainerAccessType) error {
	storageClient, err1 := storageclient.NewBasicClient(storageAccountName, primaryAccessKey)
	if err1 != nil {
		fmt.Println("Creating storage client failed")
		return err1
	}

	blobStorageClient := storageClient.GetBlobService()
	ok, err2 := blobStorageClient.CreateContainerIfNotExists(containerName, containerAccessType)
	if err2 != nil {
		fmt.Println("Creating storage container failed")
		return err2
	}
	if !ok {
		fmt.Println("Storage container already existed")
	}

	return nil
}

func azureFactory(conf map[string]string) (Client, error) {
	storageAccountName, ok := conf["storage_account_name"]
	if !ok {
		return nil, fmt.Errorf("missing 'storage_account_name' configuration")
	}
	containerName, ok := conf["container_name"]
	if !ok {
		return nil, fmt.Errorf("missing 'container_name' configuration")
	}
	keyName, ok := conf["key"]
	if !ok {
		return nil, fmt.Errorf("missing 'key' configuration")
	}

	accessKey, ok := confOrEnv(conf, "access_key", "ARM_ACCESS_KEY")
	if !ok {
		resourceGroupName, ok := conf["resource_group_name"]
		if !ok {
			return nil, fmt.Errorf("missing 'resource_group' configuration")
		}

		var err error
		accessKey, err = getStorageAccountAccessKey(conf, resourceGroupName, storageAccountName)
		if err != nil {
			return nil, fmt.Errorf("Couldn't read access key from storage account: %s.", err)
		}
	}

	storageClient, err := mainStorage.NewBasicClient(storageAccountName, accessKey)
	if err != nil {
		return nil, fmt.Errorf("Error creating storage client for storage account %q: %s", storageAccountName, err)
	}

	blobClient := storageClient.GetBlobService()
	leaseID, _ := confOrEnv(conf, "lease_id", "ARM_LEASE_ID")

	return &AzureClient{
		blobClient:    &blobClient,
		containerName: containerName,
		keyName:       keyName,
		leaseID:       leaseID,
	}, nil
}

// newAzureBackend constructs an Azure backend using a pre-existing
// bucket. Credentials can be provided to the backend, sourced
// from the environment, AWS credential files or by IAM role.
func newAzureBackend(conf map[string]string, logger *log.Logger) (Backend, error) {

	container := os.Getenv("AZURE_BLOB_CONTAINER")
	if container == "" {
		container = conf["container"]
		if container == "" {
			return nil, fmt.Errorf("'container' must be set")
		}
	}

	accountName := os.Getenv("AZURE_ACCOUNT_NAME")
	if accountName == "" {
		accountName = conf["accountName"]
		if accountName == "" {
			return nil, fmt.Errorf("'accountName' must be set")
		}
	}

	accountKey := os.Getenv("AZURE_ACCOUNT_KEY")
	if accountKey == "" {
		accountKey = conf["accountKey"]
		if accountKey == "" {
			return nil, fmt.Errorf("'accountKey' must be set")
		}
	}

	client, err := storage.NewBasicClient(accountName, accountKey)

	if err != nil {
		return nil, fmt.Errorf("Failed to create Azure client: %v", err)
	}

	client.GetBlobService().CreateContainerIfNotExists(container, storage.ContainerAccessTypePrivate)

	a := &AzureBackend{
		container: container,
		client:    client.GetBlobService(),
		logger:    logger,
	}
	return a, nil
}

// AcsInit sets up an Azure Client that can talk to the storage account
func AcsInit() (err error) {
	azureAccountName := os.Getenv("AZURE_STORAGE_ACCOUNT_NAME")
	if azureAccountName == "" {
		log.Error("You need to pass in environment variable AZURE_STORAGE_ACCOUNT_NAME")
		err = fmt.Errorf("Azure storage account name not configured")
		return
	}
	azureKey := os.Getenv("AZURE_STORAGE_ACCOUNT_KEY")
	if azureAccountName == "" {
		log.Error("You need to pass in environment variable AZURE_STORAGE_ACCOUNT_KEY")
		err = fmt.Errorf("Azure storage account key not configured")
		return
	}

	azureClient, err = storage.NewBasicClient(azureAccountName, azureKey)
	if err == nil {
		azureQueueClient = azureClient.GetQueueService()
	}

	azureInitialized = true
	return
}