Refactor Azure Storage handling.

.gitignore

@@ -343,4 +343,5 @@ config
 .vscode
 
 # GO Vendor
-vendor
+vendor
+cover.out

Makefile

@@ -27,7 +27,7 @@ all: test build
 ##################################################
 .PHONY: test
 test:
-	go test ./...
+	go test ./... -covermode=atomic -coverprofile cover.out
 
 .PHONY: e2e-test
 e2e-test:

cmd/manager/main.go

@@ -12,8 +12,8 @@ import (
 	"k8s.io/client-go/rest"
 
 	"github.com/kedacore/keda/pkg/apis"
-	"github.com/kedacore/keda/version"
 	"github.com/kedacore/keda/pkg/controller"
+	"github.com/kedacore/keda/version"
 
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
 	kubemetrics "github.com/operator-framework/operator-sdk/pkg/kube-metrics"

pkg/apis/keda/v1alpha1/scaledobject_types.go

@@ -79,7 +79,7 @@ type ScaleTriggers struct {
 // +optional
 type ScaledObjectStatus struct {
 	// +optional
-	LastActiveTime  *metav1.Time `json:"lastActiveTime,omitempty"`
+	LastActiveTime *metav1.Time `json:"lastActiveTime,omitempty"`
 	// +optional
 	// +listType
 	ExternalMetricNames []string `json:"externalMetricNames,omitempty"`

pkg/handler/scale_jobs.go

@@ -61,11 +61,11 @@ func (h *ScaleHandler) createJobs(scaledObject *kedav1alpha1.ScaledObject, scale
 				GenerateName: scaledObject.GetName() + "-",
 				Namespace:    scaledObject.GetNamespace(),
 				Labels: map[string]string{
-					"app.kubernetes.io/name": scaledObject.GetName(),
-					"app.kubernetes.io/version": version.Version,
-					"app.kubernetes.io/part-of": scaledObject.GetName(),
+					"app.kubernetes.io/name":       scaledObject.GetName(),
+					"app.kubernetes.io/version":    version.Version,
+					"app.kubernetes.io/part-of":    scaledObject.GetName(),
 					"app.kubernetes.io/managed-by": "keda-operator",
-					"scaledobject": scaledObject.GetName(),
+					"scaledobject":                 scaledObject.GetName(),
 				},
 			},
 			Spec: *scaledObject.Spec.JobTargetRef.DeepCopy(),

pkg/scalers/azure_aad_podidentity.go → pkg/scalers/azure/azure_aad_podidentity.go

@@ -1,4 +1,4 @@
-package scalers
+package azure
 
 import (
 	"encoding/json"
@@ -13,11 +13,11 @@ const (
 	msiURL = "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=%s"
 )
 
-func getAzureADPodIdentityToken(uri string) (AADToken, error) {
+func GetAzureADPodIdentityToken(audience string) (AADToken, error) {
 
 	var token AADToken
 
-	resp, err := http.Get(fmt.Sprintf(msiURL, url.QueryEscape(uri)))
+	resp, err := http.Get(fmt.Sprintf(msiURL, url.QueryEscape(audience)))
 	if err != nil {
 		return token, err
 	}

pkg/scalers/azure/azure_blob.go

@@ -0,0 +1,28 @@
+package azure
+
+import (
+	"context"
+	"github.com/Azure/azure-storage-blob-go/azblob"
+)
+
+// GetAzureBlobListLength returns the count of the blobs in blob container in int
+func GetAzureBlobListLength(ctx context.Context, podIdentity string, connectionString, blobContainerName string, accountName string, blobDelimiter string, blobPrefix string) (int, error) {
+	credential, endpoint, err := ParseAzureStorageBlobConnection(podIdentity, connectionString, accountName)
+	if err != nil {
+		return -1, err
+	}
+
+	listBlobsSegmentOptions := azblob.ListBlobsSegmentOptions{
+		Prefix: blobPrefix,
+	}
+	p := azblob.NewPipeline(credential, azblob.PipelineOptions{})
+	serviceURL := azblob.NewServiceURL(*endpoint, p)
+	containerURL := serviceURL.NewContainerURL(blobContainerName)
+
+	props, err := containerURL.ListBlobsHierarchySegment(ctx, azblob.Marker{}, blobDelimiter, listBlobsSegmentOptions)
+	if err != nil {
+		return -1, err
+	}
+
+	return len(props.Segment.BlobItems), nil
+}

pkg/scalers/azure/azure_blob_test.go

@@ -0,0 +1,36 @@
+package azure
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func TestGetBlobLength(t *testing.T) {
+	length, err := GetAzureBlobListLength(context.TODO(), "", "", "blobContainerName", "", "", "")
+	if length != -1 {
+		t.Error("Expected length to be -1, but got", length)
+	}
+
+	if err == nil {
+		t.Error("Expected error for empty connection string, but got nil")
+	}
+
+	if !strings.Contains(err.Error(), "parse storage connection string") {
+		t.Error("Expected error to contain parsing error message, but got", err.Error())
+	}
+
+	length, err = GetAzureBlobListLength(context.TODO(), "", "DefaultEndpointsProtocol=https;AccountName=name;AccountKey=key==;EndpointSuffix=core.windows.net", "blobContainerName", "", "", "")
+
+	if length != -1 {
+		t.Error("Expected length to be -1, but got", length)
+	}
+
+	if err == nil {
+		t.Error("Expected error for empty connection string, but got nil")
+	}
+
+	if !strings.Contains(err.Error(), "illegal base64") {
+		t.Error("Expected error to contain base64 error message, but got", err.Error())
+	}
+}

pkg/scalers/azure_eventhub.go → pkg/scalers/azure/azure_eventhub.go

@@ -1,4 +1,4 @@
-package scalers
+package azure
 
 import (
 	"bytes"
@@ -13,11 +13,6 @@ import (
 
 	eventhub "github.com/Azure/azure-event-hubs-go"
 	"github.com/Azure/azure-storage-blob-go/azblob"
-	"github.com/Azure/go-autorest/autorest/azure"
-)
-
-const (
-	environmentName = "AzurePublicCloud"
 )
 
 type baseCheckpoint struct {
@@ -43,29 +38,16 @@ type pythonCheckpoint struct {
 	SequenceNumber int64  `json:"sequence_number"`
 }
 
-// GetStorageCredentials returns azure env and storage credentials
-func GetStorageCredentials(storageConnection string) (azure.Environment, *azblob.SharedKeyCredential, error) {
-	_, storageAccountName, storageAccountKey, _, err := ParseAzureStorageConnectionString(storageConnection)
-	if err != nil {
-		return azure.Environment{}, &azblob.SharedKeyCredential{}, fmt.Errorf("unable to parse connection string: %s", storageConnection)
-	}
-
-	azureEnv, err := azure.EnvironmentFromName(environmentName)
-	if err != nil {
-		return azureEnv, nil, fmt.Errorf("could not get azure.Environment struct: %s", err)
-	}
-
-	cred, err := azblob.NewSharedKeyCredential(storageAccountName, storageAccountKey)
-	if err != nil {
-		return azureEnv, nil, fmt.Errorf("could not prepare a blob storage credential: %s", err)
-	}
-
-	return azureEnv, cred, nil
+type EventHubInfo struct {
+	EventHubConnection    string
+	EventHubConsumerGroup string
+	StorageConnection     string
+	BlobContainer         string
 }
 
 // GetEventHubClient returns eventhub client
-func GetEventHubClient(connectionString string) (*eventhub.Hub, error) {
-	hub, err := eventhub.NewHubFromConnectionString(connectionString)
+func GetEventHubClient(info EventHubInfo) (*eventhub.Hub, error) {
+	hub, err := eventhub.NewHubFromConnectionString(info.EventHubConnection)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create hub client: %s", err)
 	}
@@ -74,39 +56,34 @@ func GetEventHubClient(connectionString string) (*eventhub.Hub, error) {
 }
 
 // GetCheckpointFromBlobStorage accesses Blob storage and gets checkpoint information of a partition
-func GetCheckpointFromBlobStorage(ctx context.Context, partitionID string, eventHubMetadata EventHubMetadata) (Checkpoint, error) {
-	endpointProtocol, storageAccountName, _, endpointSuffix, err := ParseAzureStorageConnectionString(eventHubMetadata.storageConnection)
+func GetCheckpointFromBlobStorage(ctx context.Context, info EventHubInfo, partitionID string) (Checkpoint, error) {
+
+	blobCreds, storageEndpoint, err := ParseAzureStorageBlobConnection("none", info.StorageConnection, "")
 	if err != nil {
-		return Checkpoint{}, fmt.Errorf("unable to parse storage connection string: %s", err)
+		return Checkpoint{}, err
 	}
 
-	// Remove trailing spaces from endpointSuffix
-	endpointSuffix = strings.TrimSpace(endpointSuffix)
-
-	eventHubNamespace, eventHubName, err := ParseAzureEventHubConnectionString(eventHubMetadata.eventHubConnection)
+	eventHubNamespace, eventHubName, err := ParseAzureEventHubConnectionString(info.EventHubConnection)
 	if err != nil {
-		return Checkpoint{}, fmt.Errorf("unable to parse event hub connection string: %s", err)
+		return Checkpoint{}, err
 	}
 
 	// TODO: add more ways to read from different types of storage and read checkpoints/leases written in different JSON formats
-	var u *url.URL
+	var baseURL *url.URL
 	// Checking blob store for C# and Java applications
-	if eventHubMetadata.blobContainer != "" {
-		// URL format - <endpointProtocol>://<storageAccountName>.blob.<endpointSuffix>/<blobContainer>/<eventHubConsumerGroup>/<partitionID>
-		u, _ = url.Parse(fmt.Sprintf("%s://%s.blob.%s/%s/%s/%s", endpointProtocol, storageAccountName, endpointSuffix, eventHubMetadata.blobContainer, eventHubMetadata.eventHubConsumerGroup, partitionID))
+	if info.BlobContainer != "" {
+		// URL format - <storageEndpoint>/<blobContainer>/<eventHubConsumerGroup>/<partitionID>
+		path, _ := url.Parse(fmt.Sprintf("/%s/%s/%s", info.BlobContainer, info.EventHubConsumerGroup, partitionID))
+		baseURL = storageEndpoint.ResolveReference(path)
 	} else {
 		// Checking blob store for Azure functions
-		// URL format - <endpointProtocol>://<storageAccountName>.blob.<endpointSuffix>/azure-webjobs-eventhub/<eventHubNamespace>/<eventHubName>/<eventHubConsumerGroup>/<partitionID>
-		u, _ = url.Parse(fmt.Sprintf("%s://%s.blob.%s/azure-webjobs-eventhub/%s/%s/%s/%s", endpointProtocol, storageAccountName, endpointSuffix, eventHubNamespace, eventHubName, eventHubMetadata.eventHubConsumerGroup, partitionID))
-	}
-
-	_, cred, err := GetStorageCredentials(eventHubMetadata.storageConnection)
-	if err != nil {
-		return Checkpoint{}, fmt.Errorf("unable to get storage credentials: %s", err)
+		// URL format - <storageEndpoint>/azure-webjobs-eventhub/<eventHubNamespace>/<eventHubName>/<eventHubConsumerGroup>/<partitionID>
+		path, _ := url.Parse(fmt.Sprintf("/azure-webjobs-eventhub/%s/%s/%s/%s", eventHubNamespace, eventHubName, info.EventHubConsumerGroup, partitionID))
+		baseURL = storageEndpoint.ResolveReference(path)
 	}
 
 	// Create a BlockBlobURL object to a blob in the container.
-	blobURL := azblob.NewBlockBlobURL(*u, azblob.NewPipeline(cred, azblob.PipelineOptions{}))
+	blobURL := azblob.NewBlockBlobURL(*baseURL, azblob.NewPipeline(blobCreds, azblob.PipelineOptions{}))
 
 	get, err := blobURL.Download(ctx, 0, 0, azblob.BlobAccessConditions{}, false)
 	if err != nil {
@@ -164,7 +141,7 @@ func ParseAzureEventHubConnectionString(connectionString string) (string, string
 	}
 
 	if eventHubNamespace == "" || eventHubName == "" {
-		return "", "", errors.New("Can't parse event hub connection string")
+		return "", "", errors.New("can't parse event hub connection string. Missing eventHubNamespace or eventHubName")
 	}
 
 	return eventHubNamespace, eventHubName, nil

pkg/scalers/azure/azure_eventhub_test.go

@@ -0,0 +1,37 @@
+package azure
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const csharpSdkCheckpoint = `{
+		"Epoch": 123456,
+		"Offset": "test offset",
+		"Owner": "test owner",
+		"PartitionId": "test partitionId",
+		"SequenceNumber": 12345
+	}`
+
+const pythonSdkCheckpoint = `{
+		"epoch": 123456,
+		"offset": "test offset",
+		"owner": "test owner",
+		"partition_id": "test partitionId",
+		"sequence_number": 12345
+	}`
+
+func TestGetCheckpoint(t *testing.T) {
+	cckp, err := getCheckpoint([]byte(csharpSdkCheckpoint))
+	if err != nil {
+		t.Error(err)
+	}
+
+	pckp, err := getCheckpoint([]byte(pythonSdkCheckpoint))
+	if err != nil {
+		t.Error(err)
+	}
+
+	assert.Equal(t, cckp, pckp)
+}