feat: Provide Azure Application Insights Scaler

.golangci.yml

@@ -76,7 +76,11 @@ issues:
     - linters:
        - stylecheck
       text: "ST1000:"
-
+  # The call to autorest.Send() in scalers/azure_app_insights.go is marked as not closing the response body. However, autorest.DoCloseIfError()
+  # and autorest.ByClosing() should ensure that the response body is closed.
+    - path: azure/azure_app_insights.go
+      linters:
+        - bodyclose
 linters-settings:
   funlen:
     lines: 80

CHANGELOG.md

@@ -27,6 +27,7 @@
 - Add ActiveMQ Scaler ([#2305](https://github.com/kedacore/keda/pull/2305))
 - Add New Datadog Scaler ([#2354](https://github.com/kedacore/keda/pull/2354))
 - Add PredictKube Scaler ([#2418](https://github.com/kedacore/keda/pull/2418))
+- Add Azure Application Insights Scaler ([2506](https://github.com/kedacore/keda/pull/2506))
 
 ### Improvements
 

pkg/scalers/azure/azure_app_insights.go

@@ -0,0 +1,145 @@
+package azure
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/azure/auth"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+)
+
+const (
+	appInsightsResource = "https://api.applicationinsights.io"
+)
+
+type AppInsightsInfo struct {
+	ApplicationInsightsID string
+	TenantID              string
+	MetricID              string
+	AggregationTimespan   string
+	AggregationType       string
+	Filter                string
+	ClientID              string
+	ClientPassword        string
+}
+
+type ApplicationInsightsMetric struct {
+	Value map[string]interface{}
+}
+
+var azureAppInsightsLog = logf.Log.WithName("azure_app_insights_scaler")
+
+func toISO8601(time string) (string, error) {
+	timeSegments := strings.Split(time, ":")
+	if len(timeSegments) != 2 {
+		return "", fmt.Errorf("invalid interval %s", time)
+	}
+
+	hours, herr := strconv.Atoi(timeSegments[0])
+	minutes, merr := strconv.Atoi(timeSegments[1])
+
+	if herr != nil || merr != nil {
+		return "", fmt.Errorf("errors parsing time: %v, %v", herr, merr)
+	}
+
+	return fmt.Sprintf("PT%02dH%02dM", hours, minutes), nil
+}
+
+func getAuthConfig(info AppInsightsInfo, podIdentity kedav1alpha1.PodIdentityProvider) auth.AuthorizerConfig {
+	if podIdentity == "" || podIdentity == kedav1alpha1.PodIdentityProviderNone {
+		config := auth.NewClientCredentialsConfig(info.ClientID, info.ClientPassword, info.TenantID)
+		config.Resource = appInsightsResource
+		return config
+	}
+
+	config := auth.NewMSIConfig()
+	config.Resource = appInsightsResource
+	return config
+}
+
+func extractAppInsightValue(info AppInsightsInfo, metric ApplicationInsightsMetric) (int32, error) {
+	if _, ok := metric.Value[info.MetricID]; !ok {
+		return -1, fmt.Errorf("metric named %s not found in app insights response", info.MetricID)
+	}
+
+	floatVal := 0.0
+	if val, ok := metric.Value[info.MetricID].(map[string]interface{})[info.AggregationType]; ok {
+		if val == nil {
+			return -1, fmt.Errorf("metric %s was nil for aggregation type %s", info.MetricID, info.AggregationType)
+		}
+		floatVal = val.(float64)
+	} else {
+		return -1, fmt.Errorf("metric %s did not containe aggregation type %s", info.MetricID, info.AggregationType)
+	}
+
+	azureAppInsightsLog.V(2).Info("value extracted from metric request", "metric type", info.AggregationType, "metric value", floatVal)
+
+	return int32(math.Round(floatVal)), nil
+}
+
+func queryParamsForAppInsightsRequest(info AppInsightsInfo) (map[string]interface{}, error) {
+	timespan, err := toISO8601(info.AggregationTimespan)
+	if err != nil {
+		return nil, err
+	}
+
+	queryParams := map[string]interface{}{
+		"aggregation": info.AggregationType,
+		"timespan":    timespan,
+	}
+	if info.Filter != "" {
+		queryParams["filter"] = info.Filter
+	}
+
+	return queryParams, nil
+}
+
+// GetAzureAppInsightsMetricValue returns the value of an Azure App Insights metric, rounded to the nearest int
+func GetAzureAppInsightsMetricValue(ctx context.Context, info AppInsightsInfo, podIdentity kedav1alpha1.PodIdentityProvider) (int32, error) {
+	config := getAuthConfig(info, podIdentity)
+	authorizer, err := config.Authorizer()
+	if err != nil {
+		return -1, err
+	}
+
+	queryParams, err := queryParamsForAppInsightsRequest(info)
+	if err != nil {
+		return -1, err
+	}
+
+	req, err := autorest.Prepare(&http.Request{},
+		autorest.WithBaseURL(appInsightsResource),
+		autorest.WithPath("v1/apps"),
+		autorest.WithPath(info.ApplicationInsightsID),
+		autorest.WithPath("metrics"),
+		autorest.WithPath(info.MetricID),
+		autorest.WithQueryParameters(queryParams),
+		authorizer.WithAuthorization())
+	if err != nil {
+		return -1, err
+	}
+
+	resp, err := autorest.Send(req,
+		autorest.DoErrorUnlessStatusCode(http.StatusOK),
+		autorest.DoCloseIfError())
+	if err != nil {
+		return -1, err
+	}
+
+	metric := &ApplicationInsightsMetric{}
+	err = autorest.Respond(resp,
+		autorest.ByUnmarshallingJSON(metric),
+		autorest.ByClosing())
+	if err != nil {
+		return -1, err
+	}
+
+	return extractAppInsightValue(info, *metric)
+}

pkg/scalers/azure/azure_app_insights_test.go

@@ -0,0 +1,176 @@
+package azure
+
+import (
+	"testing"
+
+	"github.com/Azure/go-autorest/autorest/azure/auth"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+)
+
+type testExtractAzAppInsightsTestData struct {
+	testName      string
+	isError       bool
+	expectedValue int32
+	info          AppInsightsInfo
+	metricResult  ApplicationInsightsMetric
+}
+
+func mockAppInsightsInfo(aggregationType string) AppInsightsInfo {
+	return AppInsightsInfo{
+		MetricID:        "testns/test",
+		AggregationType: aggregationType,
+	}
+}
+
+func mockAppInsightsMetric(metricName, aggregationType string, value *float64) ApplicationInsightsMetric {
+	metric := ApplicationInsightsMetric{
+		Value: map[string]interface{}{
+			metricName: map[string]interface{}{},
+		},
+	}
+
+	if value == nil {
+		metric.Value[metricName].(map[string]interface{})[aggregationType] = nil
+	} else {
+		metric.Value[metricName].(map[string]interface{})[aggregationType] = *value
+	}
+
+	return metric
+}
+
+func newMetricValue(f float64) *float64 {
+	return &f
+}
+
+var testExtractAzAppInsightsData = []testExtractAzAppInsightsTestData{
+	{"metric not found", true, -1, mockAppInsightsInfo("avg"), mockAppInsightsMetric("test/test", "avg", newMetricValue(0.0))},
+	{"metric is nil", true, -1, mockAppInsightsInfo("avg"), mockAppInsightsMetric("testns/test", "avg", nil)},
+	{"incorrect aggregation type", true, -1, mockAppInsightsInfo("avg"), mockAppInsightsMetric("testns/test", "max", newMetricValue(0.0))},
+	{"success round down value", false, 5, mockAppInsightsInfo("max"), mockAppInsightsMetric("testns/test", "max", newMetricValue(5.2))},
+	{"success round up value", false, 6, mockAppInsightsInfo("max"), mockAppInsightsMetric("testns/test", "max", newMetricValue(5.5))},
+}
+
+func TestAzGetAzureAppInsightsMetricValue(t *testing.T) {
+	for _, testData := range testExtractAzAppInsightsData {
+		value, err := extractAppInsightValue(testData.info, testData.metricResult)
+		if testData.isError {
+			if err == nil {
+				t.Errorf("Test: %v; Expected error but got success. testData: %v", testData.testName, testData)
+			}
+		} else {
+			if err == nil {
+				if testData.expectedValue != value {
+					t.Errorf("Test: %v; Expected value %v but got %v testData: %v", testData.testName, testData.expectedValue, value, testData)
+				}
+			} else {
+				t.Errorf("Test: %v; Expected success but got error: %v", testData.testName, err)
+			}
+		}
+	}
+}
+
+type testAppInsightsAuthConfigTestData struct {
+	testName    string
+	expectMSI   bool
+	info        AppInsightsInfo
+	podIdentity kedav1alpha1.PodIdentityProvider
+}
+
+var testAppInsightsAuthConfigData = []testAppInsightsAuthConfigTestData{
+	{"client credentials", false, AppInsightsInfo{ClientID: "1234", ClientPassword: "pw", TenantID: "5678"}, ""},
+	{"client credentials - pod id none", false, AppInsightsInfo{ClientID: "1234", ClientPassword: "pw", TenantID: "5678"}, kedav1alpha1.PodIdentityProviderNone},
+	{"azure pod identity", true, AppInsightsInfo{}, kedav1alpha1.PodIdentityProviderAzure},
+}
+
+func TestAzAppInfoGetAuthConfig(t *testing.T) {
+	for _, testData := range testAppInsightsAuthConfigData {
+		authConfig := getAuthConfig(testData.info, testData.podIdentity)
+		if testData.expectMSI {
+			if _, ok := authConfig.(auth.MSIConfig); !ok {
+				t.Errorf("Test %v; incorrect auth config. expected MSI config", testData.testName)
+			}
+		} else {
+			if _, ok := authConfig.(auth.ClientCredentialsConfig); !ok {
+				t.Errorf("Test: %v; incorrect auth config. expected client credentials config", testData.testName)
+			}
+		}
+	}
+}
+
+type toISO8601TestData struct {
+	testName      string
+	isError       bool
+	time          string
+	expectedValue string
+}
+
+var toISO8601Data = []toISO8601TestData{
+	{testName: "time with no colons", isError: true, time: "00", expectedValue: "doesnotmatter"},
+	{testName: "time with too many colons", isError: true, time: "00:00:00", expectedValue: "doesnotmatter"},
+	{testName: "time is not a number", isError: true, time: "12a:55", expectedValue: "doesnotmatter"},
+	{testName: "valid time", isError: false, time: "12:55", expectedValue: "PT12H55M"},
+}
+
+func TestToISO8601(t *testing.T) {
+	for _, testData := range toISO8601Data {
+		value, err := toISO8601(testData.time)
+		if testData.isError {
+			if err == nil {
+				t.Errorf("Test: %v; Expected error but got success. testData: %v", testData.testName, testData)
+			}
+		} else {
+			if err == nil {
+				if testData.expectedValue != value {
+					t.Errorf("Test: %v; Expected value %v but got %v testData: %v", testData.testName, testData.expectedValue, value, testData)
+				}
+			} else {
+				t.Errorf("Test: %v; Expected success but got error: %v", testData.testName, err)
+			}
+		}
+	}
+}
+
+type queryParameterTestData struct {
+	testName         string
+	isError          bool
+	info             AppInsightsInfo
+	expectedTimespan string
+}
+
+var queryParameterData = []queryParameterTestData{
+	{testName: "invalid timespace", isError: true, info: AppInsightsInfo{AggregationType: "avg", AggregationTimespan: "00:00:00", Filter: "cloud/roleName eq 'role'"}},
+	{testName: "empty filter", isError: false, expectedTimespan: "PT01H02M", info: AppInsightsInfo{AggregationType: "min", AggregationTimespan: "01:02", Filter: ""}},
+	{testName: "filter specified", isError: false, expectedTimespan: "PT01H02M", info: AppInsightsInfo{AggregationType: "min", AggregationTimespan: "01:02", Filter: "cloud/roleName eq 'role'"}},
+}
+
+func TestQueryParamsForAppInsightsRequest(t *testing.T) {
+	for _, testData := range queryParameterData {
+		params, err := queryParamsForAppInsightsRequest(testData.info)
+		if testData.isError {
+			if err == nil {
+				t.Errorf("Test: %v; Expected error but got success. testData: %v", testData.testName, testData)
+			}
+		} else {
+			if err == nil {
+				if testData.info.AggregationType != params["aggregation"] {
+					t.Errorf("Test: %v; Expected aggregation %v actual %v", testData.testName, testData.info.AggregationType, params["aggregation"])
+				}
+				if testData.expectedTimespan != params["timespan"] {
+					t.Errorf("Test: %v; Expected timespan %v actual %v", testData.testName, testData.expectedTimespan, params["timespan"])
+				}
+				if testData.info.Filter == "" {
+					if params["filter"] != nil {
+						t.Errorf("Test: %v; Filter should not be included in params", testData.testName)
+					}
+				} else {
+					if params["filter"] != testData.info.Filter {
+						t.Errorf("Test: %v; Expected filter %v actual %v", testData.testName, testData.info.Filter, params["filter"])
+					}
+				}
+			} else {
+				t.Errorf("Test: %v; Expected success but got error: %v", testData.testName, err)
+			}
+		}
+	}
+}