feat: Provide Azure Data Explorer Scaler (#1488)

CHANGELOG.md

@@ -32,6 +32,7 @@
 
 - **General:** Automatically release container image for ARM ([#2263]https://github.com/kedacore/keda/issues/2263))
 - **General:** Automatically run end-to-end tests on ARM ([#2262]https://github.com/kedacore/keda/issues/2262))
+- **Azure Data Explorer Scaler:** Add Azure Data Explorer Scaler ([#2627](https://github.com/kedacore/keda/pull/2627))
 
 ### Improvements
 

go.mod

@@ -7,6 +7,7 @@ require (
 	cloud.google.com/go/monitoring v1.2.0
 	github.com/Azure/azure-amqp-common-go/v3 v3.2.2
 	github.com/Azure/azure-event-hubs-go/v3 v3.3.16
+	github.com/Azure/azure-kusto-go v0.5.0
 	github.com/Azure/azure-sdk-for-go v61.4.0+incompatible
 	github.com/Azure/azure-service-bus-go v0.11.5
 	github.com/Azure/azure-storage-blob-go v0.14.0

go.sum

@@ -62,11 +62,14 @@ github.com/Azure/azure-amqp-common-go/v3 v3.2.2 h1:CJpxNAGxP7UBhDusRUoaOn0uOorQy
 github.com/Azure/azure-amqp-common-go/v3 v3.2.2/go.mod h1:O6X1iYHP7s2x7NjUKsXVhkwWrQhxrd+d8/3rRadj4CI=
 github.com/Azure/azure-event-hubs-go/v3 v3.3.16 h1:e3iHaU6Tgq5A8F313uIUWwwXtXAn75iIC6ekgzW6TG8=
 github.com/Azure/azure-event-hubs-go/v3 v3.3.16/go.mod h1:xgDvUi1+8/bb11WTEaU7VwZREYufzKzjWE4YiPZixb0=
+github.com/Azure/azure-kusto-go v0.5.0 h1:BUD6WK22gXlWp2x81uy8ztWyyAKoP8Mv2RU82XaiZhw=
+github.com/Azure/azure-kusto-go v0.5.0/go.mod h1:2xOhBxRcHyyNifFHmNMcqYL6AMdhyrUHCkEJkrZ+EI4=
 github.com/Azure/azure-pipeline-go v0.1.8/go.mod h1:XA1kFWRVhSK+KNFiOhfv83Fv8L9achrP7OxIzeTn1Yg=
 github.com/Azure/azure-pipeline-go v0.1.9/go.mod h1:XA1kFWRVhSK+KNFiOhfv83Fv8L9achrP7OxIzeTn1Yg=
 github.com/Azure/azure-pipeline-go v0.2.3 h1:7U9HBg1JFK3jHl5qmo4CTZKFTVgMwdFHMVtCdfBE21U=
 github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl2bqk7S3ta6S6u4k=
 github.com/Azure/azure-sdk-for-go v51.1.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
+github.com/Azure/azure-sdk-for-go v61.2.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go v61.4.0+incompatible h1:BF2Pm3aQWIa6q9KmxyF1JYKYXtVw67vtvu2Wd54NGuY=
 github.com/Azure/azure-sdk-for-go v61.4.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v0.19.0/go.mod h1:h6H6c8enJmmocHUbLiiGY6sx7f9i+X3m1CHdd5c6Rdw=
@@ -388,6 +391,7 @@ github.com/gobwas/ws v1.0.2/go.mod h1:szmBTxLgaFppYjEmNtny/v3w89xOydFnnZMcgRRu/E
 github.com/gocql/gocql v0.0.0-20211222173705-d73e6b1002a7 h1:jmIMM+nEO+vjz9xaRIg9sZNtNLq5nsSbsxwe1OtRwv4=
 github.com/gocql/gocql v0.0.0-20211222173705-d73e6b1002a7/go.mod h1:3gM2c4D3AnkISwBxGnMMsS8Oy4y2lhbPRsH4xnJrHG8=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gofrs/uuid v4.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -687,6 +691,7 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4FW1e6jwpg=
 github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=

pkg/scalers/azure/azure_data_explorer.go

@@ -0,0 +1,131 @@
+/*
+Copyright 2021 The KEDA 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 (
+	"context"
+	"fmt"
+	"io"
+	"strconv"
+
+	"github.com/Azure/azure-kusto-go/kusto"
+	"github.com/Azure/azure-kusto-go/kusto/data/table"
+	"github.com/Azure/azure-kusto-go/kusto/unsafe"
+	"github.com/Azure/go-autorest/autorest/azure/auth"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+type DataExplorerMetadata struct {
+	ClientID     string
+	ClientSecret string
+	DatabaseName string
+	Endpoint     string
+	MetricName   string
+	PodIdentity  string
+	Query        string
+	TenantID     string
+	Threshold    int64
+}
+
+var azureDataExplorerLogger = logf.Log.WithName("azure_data_explorer_scaler")
+
+func CreateAzureDataExplorerClient(metadata *DataExplorerMetadata) (*kusto.Client, error) {
+	authConfig, err := getDataExplorerAuthConfig(metadata)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get data explorer auth config: %v", err)
+	}
+
+	client, err := kusto.New(metadata.Endpoint, kusto.Authorization{Config: *authConfig})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create kusto client: %v", err)
+	}
+
+	return client, nil
+}
+
+func getDataExplorerAuthConfig(metadata *DataExplorerMetadata) (*auth.AuthorizerConfig, error) {
+	var authConfig auth.AuthorizerConfig
+
+	if metadata.PodIdentity != "" {
+		authConfig = auth.NewMSIConfig()
+		azureDataExplorerLogger.V(1).Info("Creating Azure Data Explorer Client using Pod Identity")
+		return &authConfig, nil
+	}
+
+	if metadata.ClientID != "" && metadata.ClientSecret != "" && metadata.TenantID != "" {
+		authConfig = auth.NewClientCredentialsConfig(metadata.ClientID, metadata.ClientSecret, metadata.TenantID)
+		azureDataExplorerLogger.V(1).Info("Creating Azure Data Explorer Client using clientID, clientSecret and tenantID")
+		return &authConfig, nil
+	}
+
+	return nil, fmt.Errorf("missing credentials. please reconfigure your scaled object metadata")
+}
+
+func GetAzureDataExplorerMetricValue(ctx context.Context, client *kusto.Client, db string, query string) (int64, error) {
+	azureDataExplorerLogger.V(1).Info("Querying Azure Data Explorer", "db", db, "query", query)
+
+	iter, err := client.Query(ctx, db, kusto.NewStmt("", kusto.UnsafeStmt(unsafe.Stmt{Add: true, SuppressWarning: false})).UnsafeAdd(query))
+	if err != nil {
+		return -1, fmt.Errorf("failed to get azure data explorer metric result from query %s: %v", query, err)
+	}
+	defer iter.Stop()
+
+	row, err := iter.Next()
+	if err != nil {
+		return -1, fmt.Errorf("failed to get query %s result: %v", query, err)
+	}
+
+	if !row.ColumnTypes[0].Type.Valid() {
+		return -1, fmt.Errorf("column type %s is not valid", row.ColumnTypes[0].Type)
+	}
+
+	// Return error if there is more than one row.
+	_, err = iter.Next()
+	if err != io.EOF {
+		return -1, fmt.Errorf("query %s result had more than a single result row", query)
+	}
+
+	metricValue, err := extractDataExplorerMetricValue(row)
+	if err != nil {
+		return -1, fmt.Errorf("failed to extract value from query %s: %v", query, err)
+	}
+
+	return metricValue, nil
+}
+
+func extractDataExplorerMetricValue(row *table.Row) (int64, error) {
+	if row == nil || len(row.ColumnTypes) == 0 {
+		return -1, fmt.Errorf("query has no results")
+	}
+
+	// Query result validation.
+	dataType := row.ColumnTypes[0].Type
+	if dataType != "real" && dataType != "int" && dataType != "long" {
+		return -1, fmt.Errorf("data type %s is not valid", dataType)
+	}
+
+	value, err := strconv.Atoi(row.Values[0].String())
+	if err != nil {
+		return -1, fmt.Errorf("failed to convert result %s to int", row.Values[0].String())
+	}
+	if value < 0 {
+		return -1, fmt.Errorf("query result must be >= 0 but received: %d", value)
+	}
+
+	azureDataExplorerLogger.V(1).Info("Query Result", "value", value, "dataType", dataType)
+	return int64(value), nil
+}

pkg/scalers/azure/azure_data_explorer_test.go

@@ -0,0 +1,100 @@
+/*
+Copyright 2021 The KEDA 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 (
+	"testing"
+
+	"github.com/Azure/azure-kusto-go/kusto/data/errors"
+	"github.com/Azure/azure-kusto-go/kusto/data/table"
+	"github.com/Azure/azure-kusto-go/kusto/data/types"
+	"github.com/Azure/azure-kusto-go/kusto/data/value"
+)
+
+type testExtractDataExplorerMetricValue struct {
+	testRow *table.Row
+	isError bool
+}
+
+type testGetDataExplorerAuthConfig struct {
+	testMetadata *DataExplorerMetadata
+	isError      bool
+}
+
+var (
+	clientID                 = "test_client_id"
+	rowName                  = "result"
+	rowType     types.Column = "long"
+	rowValue    int64        = 3
+	podIdentity              = "Azure"
+	secret                   = "test_secret"
+	tenantID                 = "test_tenant_id"
+)
+
+var testExtractDataExplorerMetricValues = []testExtractDataExplorerMetricValue{
+	// pass
+	{testRow: &table.Row{ColumnTypes: table.Columns{{Name: rowName, Type: rowType}}, Values: value.Values{value.Long{Value: rowValue, Valid: true}}, Op: errors.OpQuery}, isError: false},
+	// nil row - fail
+	{testRow: nil, isError: true},
+	// Empty row - fail
+	{testRow: &table.Row{}, isError: true},
+	// Metric value is not bigger than 0 - fail
+	{testRow: &table.Row{ColumnTypes: table.Columns{{Name: rowName, Type: rowType}}, Values: value.Values{value.Long{Value: -1, Valid: true}}, Op: errors.OpQuery}, isError: true},
+	// Metric result is invalid - fail
+	{testRow: &table.Row{ColumnTypes: table.Columns{{Name: rowName, Type: rowType}}, Values: value.Values{value.String{Value: "invalid", Valid: true}}, Op: errors.OpQuery}, isError: true},
+	// Metric Type is not valid - fail
+	{testRow: &table.Row{ColumnTypes: table.Columns{{Name: rowName, Type: "String"}}, Values: value.Values{value.Long{Value: rowValue, Valid: true}}, Op: errors.OpQuery}, isError: true},
+}
+
+var testGetDataExplorerAuthConfigs = []testGetDataExplorerAuthConfig{
+	// Auth with aad app - pass
+	{testMetadata: &DataExplorerMetadata{ClientID: clientID, ClientSecret: secret, TenantID: tenantID}, isError: false},
+	// Auth with podIdentity - pass
+	{testMetadata: &DataExplorerMetadata{PodIdentity: podIdentity}, isError: false},
+	// Empty metadata - fail
+	{testMetadata: &DataExplorerMetadata{}, isError: true},
+	// Empty tenantID - fail
+	{testMetadata: &DataExplorerMetadata{ClientID: clientID, ClientSecret: secret}, isError: true},
+	// Empty clientID - fail
+	{testMetadata: &DataExplorerMetadata{ClientSecret: secret, TenantID: tenantID}, isError: true},
+	// Empty clientSecret - fail
+	{testMetadata: &DataExplorerMetadata{ClientID: clientID, TenantID: tenantID}, isError: true},
+}
+
+func TestExtractDataExplorerMetricValue(t *testing.T) {
+	for _, testData := range testExtractDataExplorerMetricValues {
+		_, err := extractDataExplorerMetricValue(testData.testRow)
+		if err != nil && !testData.isError {
+			t.Error("Expected success but got error", err)
+		}
+		if testData.isError && err == nil {
+			t.Error("Expected error but got success")
+		}
+	}
+}
+
+func TestGetDataExplorerAuthConfig(t *testing.T) {
+	for _, testData := range testGetDataExplorerAuthConfigs {
+		_, err := getDataExplorerAuthConfig(testData.testMetadata)
+		if err != nil && !testData.isError {
+			t.Error("Expected success but got error", err)
+		}
+		if testData.isError && err == nil {
+			t.Error("Expected error but got success")
+		}
+	}
+}