Skip to content

Commit

Permalink
feat: Provide Azure Data Explorer Scaler (#1488)
Browse files Browse the repository at this point in the history
Signed-off-by: Yarden Siboni <yasiboni@microsoft.com>
  • Loading branch information
yasiboni committed Feb 13, 2022
1 parent 130bc93 commit c2e32af
Show file tree
Hide file tree
Showing 9 changed files with 832 additions and 1 deletion.
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@

### New

- TODO ([#XXX](https://github.com/kedacore/keda/pull/XXX))
- Add Azure Data Explorer Scaler ([#2627](https://github.com/kedacore/keda/pull/2627))

### Improvements

Expand Down
1 change: 1 addition & 0 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down Expand Up @@ -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=
Expand Down
130 changes: 130 additions & 0 deletions pkg/scalers/azure/azure_data_explorer.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
/*
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.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.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.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" {
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.Info("Query Result", "value", value, "dataType", dataType)
return int64(value), nil
}

return -1, fmt.Errorf("failed to extract metric value from Data Explorer request")
}
100 changes: 100 additions & 0 deletions pkg/scalers/azure/azure_data_explorer_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
}
}
Loading

0 comments on commit c2e32af

Please sign in to comment.