add Azure storage and blob service API to support Azure disk dynamic provisioning

pkg/cloudprovider/providers/azure/azure.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/arm/compute"
 	"github.com/Azure/azure-sdk-for-go/arm/network"
+	"github.com/Azure/azure-sdk-for-go/arm/storage"
 	"github.com/Azure/go-autorest/autorest/azure"
 	"github.com/ghodss/yaml"
 )
@@ -61,6 +62,7 @@ type Cloud struct {
 	PublicIPAddressesClient network.PublicIPAddressesClient
 	SecurityGroupsClient    network.SecurityGroupsClient
 	VirtualMachinesClient   compute.VirtualMachinesClient
+	StorageAccountClient    storage.AccountsClient
 }
 
 func init() {
@@ -135,6 +137,8 @@ func NewCloud(configReader io.Reader) (cloudprovider.Interface, error) {
 	az.SecurityGroupsClient.BaseURI = az.Environment.ResourceManagerEndpoint
 	az.SecurityGroupsClient.Authorizer = servicePrincipalToken
 
+	az.StorageAccountClient = storage.NewAccountsClientWithBaseURI(az.Environment.ResourceManagerEndpoint, az.SubscriptionID)
+	az.StorageAccountClient.Authorizer = servicePrincipalToken
 	return &az, nil
 }
 

pkg/cloudprovider/providers/azure/azure_blob.go

+/*
+Copyright 2016 The Kubernetes 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 azure
+
+import (
+	"fmt"
+	"regexp"
+
+	azs "github.com/Azure/azure-sdk-for-go/storage"
+)
+
+const (
+	vhdContainerName = "vhds"
+	useHTTPS         = true
+	blobServiceName  = "blob"
+)
+
+// create page blob
+func (az *Cloud) createVhdBlob(accountName, accountKey, name string, sizeGB int64, tags map[string]string) (string, string, error) {
+	blobClient, err := az.getBlobClient(accountName, accountKey)
+	if err != nil {
+		return "", "", err
+	}
+	size := 1024 * 1024 * 1024 * sizeGB
+	vhdSize := size + vhdHeaderSize /* header size */
+	// Blob name in URL must end with '.vhd' extension.
+	name = name + ".vhd"
+	err = blobClient.PutPageBlob(vhdContainerName, name, vhdSize, tags)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to put page blob: %v", err)
+	}
+	// add VHD signature to the blob
+	h, err := createVHDHeader(uint64(size))
+	if err != nil {
+		az.deleteVhdBlob(accountName, accountKey, name)
+		return "", "", fmt.Errorf("failed to create vhd header, err: %v", err)
+	}
+	if err = blobClient.PutPage(vhdContainerName, name, size, vhdSize-1, azs.PageWriteTypeUpdate, h[:vhdHeaderSize], nil); err != nil {
+		az.deleteVhdBlob(accountName, accountKey, name)
+		return "", "", fmt.Errorf("failed to update vhd header, err: %v", err)
+	}
+
+	scheme := "http"
+	if useHTTPS {
+		scheme = "https"
+	}
+	host := fmt.Sprintf("%s://%s.%s.%s", scheme, accountName, blobServiceName, az.Environment.StorageEndpointSuffix)
+	uri := fmt.Sprintf("%s/%s/%s", host, vhdContainerName, name)
+	return name, uri, nil
+
+}
+
+// delete a vhd blob
+func (az *Cloud) deleteVhdBlob(accountName, accountKey, blobName string) error {
+	blobClient, err := az.getBlobClient(accountName, accountKey)
+	if err == nil {
+		return blobClient.DeleteBlob(vhdContainerName, blobName, nil)
+	}
+	return err
+}
+
+func (az *Cloud) getBlobClient(accountName, accountKey string) (*azs.BlobStorageClient, error) {
+	client, err := azs.NewClient(accountName, accountKey, az.Environment.StorageEndpointSuffix, azs.DefaultAPIVersion, useHTTPS)
+	if err != nil {
+		return nil, fmt.Errorf("error creating azure client: %v", err)
+	}
+	b := client.GetBlobService()
+	return &b, nil
+}
+
+// get uri https://foo.blob.core.windows.net/vhds/bar.vhd and return foo (account) and bar.vhd (blob name)
+func (az *Cloud) getBlobNameAndAccountFromURI(uri string) (string, string, error) {
+	scheme := "http"
+	if useHTTPS {
+		scheme = "https"
+	}
+	host := fmt.Sprintf("%s://(.*).%s.%s", scheme, blobServiceName, az.Environment.StorageEndpointSuffix)
+	reStr := fmt.Sprintf("%s/%s/(.*)", host, vhdContainerName)
+	re := regexp.MustCompile(reStr)
+	res := re.FindSubmatch([]byte(uri))
+	if len(res) < 3 {
+		return "", "", fmt.Errorf("invalid vhd URI for regex %s: %s", reStr, uri)
+	}
+	return string(res[1]), string(res[2]), nil
+}

pkg/cloudprovider/providers/azure/azure_storage.go

@@ -154,3 +154,62 @@ func (az *Cloud) GetNextDiskLun(nodeName types.NodeName) (int32, error) {
 	}
 	return -1, fmt.Errorf("All Luns are used")
 }
+
+// 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 (az *Cloud) CreateVolume(name, storageAccount, storageType, 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 = az.getStorageAccounts()
+		if err != nil {
+			// TODO: create a storage account and container
+			return "", "", 0, err
+		}
+	}
+	for _, account := range accounts {
+		glog.V(4).Infof("account %s type %s location %s", account.Name, account.StorageType, account.Location)
+		if ((storageType == "" || account.StorageType == storageType) && (location == "" || account.Location == location)) || len(storageAccount) > 0 {
+			// find the access key with this account
+			key, err := az.getStorageAccesskey(account.Name)
+			if err != nil {
+				glog.V(2).Infof("no key found for storage account %s", account.Name)
+				continue
+			}
+
+			// create a page blob in this account's vhd container
+			name, uri, err := az.createVhdBlob(account.Name, key, name, int64(requestGB), nil)
+			if err != nil {
+				glog.V(2).Infof("failed to create vhd in account %s: %v", account.Name, err)
+				continue
+			}
+			glog.V(4).Infof("created vhd blob uri: %s", uri)
+			return name, uri, requestGB, err
+		}
+	}
+	return "", "", 0, fmt.Errorf("failed to find a matching storage account")
+}
+
+// DeleteVolume deletes a VHD blob
+func (az *Cloud) DeleteVolume(name, uri string) error {
+	accountName, blob, err := az.getBlobNameAndAccountFromURI(uri)
+	if err != nil {
+		return fmt.Errorf("failed to parse vhd URI %v", err)
+	}
+	key, err := az.getStorageAccesskey(accountName)
+	if err != nil {
+		return fmt.Errorf("no key for storage account %s, err %v", accountName, err)
+	}
+	err = az.deleteVhdBlob(accountName, key, blob)
+	if err != nil {
+		glog.Warningf("failed to delete blob %s err: %v", uri, err)
+		return fmt.Errorf("failed to delete vhd %v, account %s, blob %s, err: %v", uri, accountName, blob, err)
+	}
+	glog.V(4).Infof("blob %s deleted", uri)
+	return nil
+
+}

pkg/cloudprovider/providers/azure/azure_storageaccount.go

+/*
+Copyright 2016 The Kubernetes 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 azure
+
+import (
+	"fmt"
+	"strings"
+)
+
+type accountWithLocation struct {
+	Name, StorageType, Location string
+}
+
+// getStorageAccounts gets the storage accounts' name, type, location in a resource group
+func (az *Cloud) getStorageAccounts() ([]accountWithLocation, error) {
+	result, err := az.StorageAccountClient.ListByResourceGroup(az.ResourceGroup)
+	if err != nil {
+		return nil, err
+	}
+	if result.Value == nil {
+		return nil, fmt.Errorf("no storage accounts from resource group %s", az.ResourceGroup)
+	}
+
+	accounts := []accountWithLocation{}
+	for _, acct := range *result.Value {
+		if acct.Name != nil {
+			name := *acct.Name
+			loc := ""
+			if acct.Location != nil {
+				loc = *acct.Location
+			}
+			storageType := ""
+			if acct.Sku != nil {
+				storageType = string((*acct.Sku).Name)
+			}
+			accounts = append(accounts, accountWithLocation{Name: name, StorageType: storageType, Location: loc})
+		}
+	}
+
+	return accounts, nil
+}
+
+// getStorageAccesskey gets the storage account access key
+func (az *Cloud) getStorageAccesskey(account string) (string, error) {
+	result, err := az.StorageAccountClient.ListKeys(az.ResourceGroup, account)
+	if err != nil {
+		return "", err
+	}
+	if result.Keys == nil {
+		return "", fmt.Errorf("empty keys")
+	}
+
+	for _, k := range *result.Keys {
+		if k.Value != nil && *k.Value != "" {
+			v := *k.Value
+			if ind := strings.LastIndex(v, " "); ind >= 0 {
+				v = v[(ind + 1):]
+			}
+			return v, nil
+		}
+	}
+	return "", fmt.Errorf("no valid keys")
+}

pkg/cloudprovider/providers/azure/vhd.go

+/*
+Copyright 2016 The Kubernetes 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 azure
+
+import (
+	"bytes"
+	"encoding/binary"
+
+	"github.com/rubiojr/go-vhd/vhd"
+)
+
+const (
+	vhdHeaderSize = vhd.VHD_HEADER_SIZE
+)
+
+func createVHDHeader(size uint64) ([]byte, error) {
+	h := vhd.CreateFixedHeader(size, &vhd.VHDOptions{})
+	b := new(bytes.Buffer)
+	err := binary.Write(b, binary.BigEndian, h)
+	if err != nil {
+		return nil, err
+	}
+	return b.Bytes(), nil
+}