Merge pull request kubernetes#54687 from andyzhangx/createvolume-fix

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

fix CreateVolume func: use search mode instead

**What this PR does / why we need it**:
This is a little fall back for CreateVolume func: use search mode for Dedicated kind as @rootfs suggested.

**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes kubernetes#52396

**Special notes for your reviewer**:
I reference the implmentation of v1.6 in the same CreateVolume func
https://github.com/kubernetes/kubernetes/blob/release-1.6/pkg/cloudprovider/providers/azure/azure_storage.go#L213-L247

**Release note**:

```
fix azure storage account exhausting issue by using azure disk mount
```
/sig azure

@rootfs @feiskyer @karataliu

Author: Kubernetes Submit Queue

pkg/cloudprovider/providers/azure/azure_blobDiskController.go

@@ -59,11 +59,12 @@ type BlobDiskController struct {
 	accounts map[string]*storageAccountState
 }
 
-var defaultContainerName = ""
-var storageAccountNamePrefix = ""
-var storageAccountNameMatch = ""
-
-var accountsLock = &sync.Mutex{}
+var (
+	defaultContainerName     = ""
+	storageAccountNamePrefix = ""
+	storageAccountNameMatch  = ""
+	accountsLock             = &sync.Mutex{}
+)
 
 func newBlobDiskController(common *controllerCommon) (*BlobDiskController, error) {
 	c := BlobDiskController{common: common}
@@ -80,52 +81,49 @@ func newBlobDiskController(common *controllerCommon) (*BlobDiskController, error
 	return &c, nil
 }
 
-// CreateVolume creates a VHD blob in a given storage account, will create the given storage account if it does not exist in current resource group
-func (c *BlobDiskController) CreateVolume(name, storageAccount string, storageAccountType storage.SkuName, location string, requestGB int) (string, string, int, error) {
-	key, err := c.common.cloud.getStorageAccesskey(storageAccount)
-	if err != nil {
-		glog.V(2).Infof("azureDisk - no key found for storage account %s in resource group %s, begin to create a new storage account", storageAccount, c.common.resourceGroup)
-
-		cp := storage.AccountCreateParameters{
-			Sku:      &storage.Sku{Name: storageAccountType},
-			Tags:     &map[string]*string{"created-by": to.StringPtr("azure-dd")},
-			Location: &location}
-		cancel := make(chan struct{})
-
-		_, errchan := c.common.cloud.StorageAccountClient.Create(c.common.resourceGroup, storageAccount, cp, cancel)
-		err = <-errchan
-		if err != nil {
-			return "", "", 0, fmt.Errorf(fmt.Sprintf("Create Storage Account %s, error: %s", storageAccount, err))
-		}
-
-		key, err = c.common.cloud.getStorageAccesskey(storageAccount)
+// CreateVolume creates a VHD blob in a storage account that has storageType and location using the given storage account.
+// If no storage account is given, search all the storage accounts associated with the resource group and pick one that
+// fits storage type and location.
+func (c *BlobDiskController) CreateVolume(name, storageAccount, storageAccountType, location string, requestGB int) (string, string, int, error) {
+	var err error
+	accounts := []accountWithLocation{}
+	if len(storageAccount) > 0 {
+		accounts = append(accounts, accountWithLocation{Name: storageAccount})
+	} else {
+		// find a storage account
+		accounts, err = c.common.cloud.getStorageAccounts()
 		if err != nil {
-			return "", "", 0, fmt.Errorf("no key found for storage account %s even after creating a new storage account", storageAccount)
+			// TODO: create a storage account and container
+			return "", "", 0, err
 		}
-
-		glog.Errorf("no key found for storage account %s in resource group %s", storageAccount, c.common.resourceGroup)
-		return "", "", 0, err
 	}
+	for _, account := range accounts {
+		glog.V(4).Infof("account %s type %s location %s", account.Name, account.StorageType, account.Location)
+		if (storageAccountType == "" || account.StorageType == storageAccountType) && (location == "" || account.Location == location) || len(storageAccount) > 0 {
+			// find the access key with this account
+			key, err := c.common.cloud.getStorageAccesskey(account.Name)
+			if err != nil {
+				glog.V(2).Infof("no key found for storage account %s", account.Name)
+				continue
+			}
 
-	client, err := azstorage.NewBasicClientOnSovereignCloud(storageAccount, key, c.common.cloud.Environment)
-	if err != nil {
-		return "", "", 0, err
-	}
-	blobClient := client.GetBlobService()
+			client, err := azstorage.NewBasicClientOnSovereignCloud(account.Name, key, c.common.cloud.Environment)
+			if err != nil {
+				return "", "", 0, err
+			}
+			blobClient := client.GetBlobService()
 
-	container := blobClient.GetContainerReference(vhdContainerName)
-	_, err = container.CreateIfNotExists(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
-	if err != nil {
-		return "", "", 0, err
-	}
+			// create a page blob in this account's vhd container
+			diskName, diskURI, err := c.createVHDBlobDisk(blobClient, account.Name, name, vhdContainerName, int64(requestGB))
+			if err != nil {
+				return "", "", 0, err
+			}
 
-	diskName, diskURI, err := c.createVHDBlobDisk(blobClient, storageAccount, name, vhdContainerName, int64(requestGB))
-	if err != nil {
-		return "", "", 0, err
+			glog.V(4).Infof("azureDisk - created vhd blob uri: %s", diskURI)
+			return diskName, diskURI, requestGB, err
+		}
 	}
-
-	glog.V(4).Infof("azureDisk - created vhd blob uri: %s", diskURI)
-	return diskName, diskURI, requestGB, err
+	return "", "", 0, fmt.Errorf("failed to find a matching storage account")
 }
 
 // DeleteVolume deletes a VHD blob
@@ -173,11 +171,6 @@ func (c *BlobDiskController) getBlobNameAndAccountFromURI(diskURI string) (strin
 
 func (c *BlobDiskController) createVHDBlobDisk(blobClient azstorage.BlobStorageClient, accountName, vhdName, containerName string, sizeGB int64) (string, string, error) {
 	container := blobClient.GetContainerReference(containerName)
-	_, err := container.CreateIfNotExists(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
-	if err != nil {
-		return "", "", err
-	}
-
 	size := 1024 * 1024 * 1024 * sizeGB
 	vhdSize := size + vhd.VHD_HEADER_SIZE /* header size */
 	// Blob name in URL must end with '.vhd' extension.
@@ -190,7 +183,17 @@ func (c *BlobDiskController) createVHDBlobDisk(blobClient azstorage.BlobStorageC
 	blob := container.GetBlobReference(vhdName)
 	blob.Properties.ContentLength = vhdSize
 	blob.Metadata = tags
-	err = blob.PutPageBlob(nil)
+	err := blob.PutPageBlob(nil)
+	if err != nil {
+		// if container doesn't exist, create one and retry PutPageBlob
+		detail := err.Error()
+		if strings.Contains(detail, errContainerNotFound) {
+			err = container.Create(&azstorage.CreateContainerOptions{Access: azstorage.ContainerAccessTypePrivate})
+			if err == nil {
+				err = blob.PutPageBlob(nil)
+			}
+		}
+	}
 	if err != nil {
 		return "", "", fmt.Errorf("failed to put page blob %s in container %s: %v", vhdName, containerName, err)
 	}
@@ -236,24 +239,12 @@ func (c *BlobDiskController) deleteVhdBlob(accountName, accountKey, blobName str
 }
 
 //CreateBlobDisk : create a blob disk in a node
-func (c *BlobDiskController) CreateBlobDisk(dataDiskName string, storageAccountType storage.SkuName, sizeGB int, forceStandAlone bool) (string, error) {
-	glog.V(4).Infof("azureDisk - creating blob data disk named:%s on StorageAccountType:%s StandAlone:%v", dataDiskName, storageAccountType, forceStandAlone)
+func (c *BlobDiskController) CreateBlobDisk(dataDiskName string, storageAccountType storage.SkuName, sizeGB int) (string, error) {
+	glog.V(4).Infof("azureDisk - creating blob data disk named:%s on StorageAccountType:%s", dataDiskName, storageAccountType)
 
-	var storageAccountName = ""
-	var err error
-
-	if forceStandAlone {
-		// we have to wait until the storage account is is created
-		storageAccountName = "p" + MakeCRC32(c.common.subscriptionID+c.common.resourceGroup+dataDiskName)
-		err = c.createStorageAccount(storageAccountName, storageAccountType, c.common.location, false)
-		if err != nil {
-			return "", err
-		}
-	} else {
-		storageAccountName, err = c.findSANameForDisk(storageAccountType)
-		if err != nil {
-			return "", err
-		}
+	storageAccountName, err := c.findSANameForDisk(storageAccountType)
+	if err != nil {
+		return "", err
 	}
 
 	blobClient, err := c.getBlobSvcClient(storageAccountName)
@@ -266,15 +257,13 @@ func (c *BlobDiskController) CreateBlobDisk(dataDiskName string, storageAccountT
 		return "", err
 	}
 
-	if !forceStandAlone {
-		atomic.AddInt32(&c.accounts[storageAccountName].diskCount, 1)
-	}
+	atomic.AddInt32(&c.accounts[storageAccountName].diskCount, 1)
 
 	return diskURI, nil
 }
 
 //DeleteBlobDisk : delete a blob disk from a node
-func (c *BlobDiskController) DeleteBlobDisk(diskURI string, wasForced bool) error {
+func (c *BlobDiskController) DeleteBlobDisk(diskURI string) error {
 	storageAccountName, vhdName, err := diskNameandSANameFromURI(diskURI)
 	if err != nil {
 		return err
@@ -286,11 +275,6 @@ func (c *BlobDiskController) DeleteBlobDisk(diskURI string, wasForced bool) erro
 		glog.V(4).Infof("azureDisk - deleting volume %s", diskURI)
 		return c.DeleteVolume(diskURI)
 	}
-	// if forced (as in one disk = one storage account)
-	// delete the account completely
-	if wasForced {
-		return c.deleteStorageAccount(storageAccountName)
-	}
 
 	blobSvc, err := c.getBlobSvcClient(storageAccountName)
 	if err != nil {

pkg/volume/azure_dd/azure_dd.go

@@ -28,8 +28,8 @@ import (
 
 // interface exposed by the cloud provider implementing Disk functionlity
 type DiskController interface {
-	CreateBlobDisk(dataDiskName string, storageAccountType storage.SkuName, sizeGB int, forceStandAlone bool) (string, error)
-	DeleteBlobDisk(diskUri string, wasForced bool) error
+	CreateBlobDisk(dataDiskName string, storageAccountType storage.SkuName, sizeGB int) (string, error)
+	DeleteBlobDisk(diskUri string) error
 
 	CreateManagedDisk(diskName string, storageAccountType storage.SkuName, sizeGB int, tags map[string]string) (string, error)
 	DeleteManagedDisk(diskURI string) error
@@ -48,7 +48,7 @@ type DiskController interface {
 	GetNextDiskLun(nodeName types.NodeName) (int32, error)
 
 	// Create a VHD blob
-	CreateVolume(name, storageAccount string, storageAccountType storage.SkuName, location string, requestGB int) (string, string, int, error)
+	CreateVolume(name, storageAccount, storageAccountType, location string, requestGB int) (string, string, int, error)
 	// Delete a VHD blob
 	DeleteVolume(diskURI string) error
 }

pkg/volume/azure_dd/azure_provision.go

@@ -55,14 +55,13 @@ func (d *azureDiskDeleter) Delete() error {
 		return err
 	}
 
-	wasStandAlone := (*volumeSource.Kind != v1.AzureSharedBlobDisk)
 	managed := (*volumeSource.Kind == v1.AzureManagedDisk)
 
 	if managed {
 		return diskController.DeleteManagedDisk(volumeSource.DataDiskURI)
 	}
 
-	return diskController.DeleteBlobDisk(volumeSource.DataDiskURI, wasStandAlone)
+	return diskController.DeleteBlobDisk(volumeSource.DataDiskURI)
 }
 
 func (p *azureDiskProvisioner) Provision() (*v1.PersistentVolume, error) {
@@ -149,26 +148,13 @@ func (p *azureDiskProvisioner) Provision() (*v1.PersistentVolume, error) {
 			return nil, err
 		}
 	} else {
-		forceStandAlone := (kind == v1.AzureDedicatedBlobDisk)
 		if kind == v1.AzureDedicatedBlobDisk {
-			if location != "" && account != "" {
-				// use dedicated kind (by default) for compatibility
-				_, diskURI, _, err = diskController.CreateVolume(name, account, skuName, location, requestGB)
-				if err != nil {
-					return nil, err
-				}
-			} else {
-				if location != "" || account != "" {
-					return nil, fmt.Errorf("AzureDisk - location(%s) and account(%s) must be both empty or specified for dedicated kind, only one value specified is not allowed",
-						location, account)
-				}
-				diskURI, err = diskController.CreateBlobDisk(name, skuName, requestGB, forceStandAlone)
-				if err != nil {
-					return nil, err
-				}
+			_, diskURI, _, err = diskController.CreateVolume(name, account, storageAccountType, location, requestGB)
+			if err != nil {
+				return nil, err
 			}
 		} else {
-			diskURI, err = diskController.CreateBlobDisk(name, skuName, requestGB, forceStandAlone)
+			diskURI, err = diskController.CreateBlobDisk(name, skuName, requestGB)
 			if err != nil {
 				return nil, err
 			}

test/e2e/framework/pv_util.go

@@ -718,16 +718,11 @@ func createPD(zone string) (string, error) {
 			return "", err
 		}
 
-		if azureCloud.BlobDiskController == nil {
-			return "", fmt.Errorf("BlobDiskController is nil, it's not expected.")
-		}
-
-		diskUri, err := azureCloud.BlobDiskController.CreateBlobDisk(pdName, "standard_lrs", 1, false)
+		_, diskURI, _, err := azureCloud.CreateVolume(pdName, "" /* account */, "" /* sku */, "" /* location */, 1 /* sizeGb */)
 		if err != nil {
 			return "", err
 		}
-
-		return diskUri, nil
+		return diskURI, nil
 	} else {
 		return "", fmt.Errorf("provider does not support volume creation")
 	}
@@ -772,11 +767,7 @@ func deletePD(pdName string) error {
 		if err != nil {
 			return err
 		}
-		if azureCloud.BlobDiskController == nil {
-			return fmt.Errorf("BlobDiskController is nil, it's not expected.")
-		}
-		diskName := pdName[(strings.LastIndex(pdName, "/") + 1):]
-		err = azureCloud.BlobDiskController.DeleteBlobDisk(diskName, false)
+		err = azureCloud.DeleteVolume(pdName)
 		if err != nil {
 			Logf("failed to delete Azure volume %q: %v", pdName, err)
 			return err