From 60046b5e673a249cb35d74499ad069520b5d3fb3 Mon Sep 17 00:00:00 2001 From: Omer Aplatony Date: Sat, 7 Dec 2024 14:45:15 +0200 Subject: [PATCH] refactor gcp cloud tasks scaler Signed-off-by: Omer Aplatony --- pkg/scalers/gcp_cloud_tasks_scaler.go | 75 +++---------- pkg/scalers/gcp_cloud_tasks_scaler_test.go | 125 ++++++++++++++++----- 2 files changed, 114 insertions(+), 86 deletions(-) diff --git a/pkg/scalers/gcp_cloud_tasks_scaler.go b/pkg/scalers/gcp_cloud_tasks_scaler.go index 3c92633d1e7..90a250693de 100644 --- a/pkg/scalers/gcp_cloud_tasks_scaler.go +++ b/pkg/scalers/gcp_cloud_tasks_scaler.go @@ -3,7 +3,6 @@ package scalers import ( "context" "fmt" - "strconv" "github.com/go-logr/logr" v2 "k8s.io/api/autoscaling/v2" @@ -16,8 +15,6 @@ import ( const ( cloudTasksStackDriverQueueSize = "cloudtasks.googleapis.com/queue/depth" - - cloudTaskDefaultValue = 100 ) type gcpCloudTasksScaler struct { @@ -28,12 +25,12 @@ type gcpCloudTasksScaler struct { } type gcpCloudTaskMetadata struct { - value float64 - activationValue float64 - filterDuration int64 + Value float64 `keda:"name=value, order=triggerMetadata, optional, default=100"` + ActivationValue float64 `keda:"name=activationValue, order=triggerMetadata, optional, default=0"` + FilterDuration int64 `keda:"name=filterDuration, order=triggerMetadata, optional"` - queueName string - projectID string + QueueName string `keda:"name=queueName, order=triggerMetadata"` + ProjectID string `keda:"name=projectID, order=triggerMetadata"` gcpAuthorization *gcp.AuthorizationMetadata triggerIndex int } @@ -60,61 +57,19 @@ func NewGcpCloudTasksScaler(config *scalersconfig.ScalerConfig) (Scaler, error) } func parseGcpCloudTasksMetadata(config *scalersconfig.ScalerConfig) (*gcpCloudTaskMetadata, error) { - meta := gcpCloudTaskMetadata{value: cloudTaskDefaultValue} - - value, valuePresent := config.TriggerMetadata["value"] - - if valuePresent { - triggerValue, err := strconv.ParseFloat(value, 64) - if err != nil { - return nil, fmt.Errorf("value parsing error %w", err) - } - meta.value = triggerValue - } - - if val, ok := config.TriggerMetadata["queueName"]; ok { - if val == "" { - return nil, fmt.Errorf("no queue name given") - } - meta.queueName = val - } else { - return nil, fmt.Errorf("no queue name given") - } - - if val, ok := config.TriggerMetadata["filterDuration"]; ok { - filterDuration, err := strconv.ParseInt(val, 10, 64) - if err != nil { - return nil, fmt.Errorf("filterDuration parsing error %w", err) - } - meta.filterDuration = filterDuration - } - - meta.activationValue = 0 - if val, ok := config.TriggerMetadata["activationValue"]; ok { - activationValue, err := strconv.ParseFloat(val, 64) - if err != nil { - return nil, fmt.Errorf("activationValue parsing error %w", err) - } - meta.activationValue = activationValue - } - - if val, ok := config.TriggerMetadata["projectID"]; ok { - if val == "" { - return nil, fmt.Errorf("no project id given") - } - - meta.projectID = val - } else { - return nil, fmt.Errorf("no project id given") + meta := &gcpCloudTaskMetadata{} + if err := config.TypedConfig(meta); err != nil { + return nil, fmt.Errorf("error parsing Gcp cloud task metadata: %w", err) } auth, err := gcp.GetGCPAuthorization(config) if err != nil { return nil, err } + meta.gcpAuthorization = auth meta.triggerIndex = config.TriggerIndex - return &meta, nil + return meta, nil } func (s *gcpCloudTasksScaler) Close(context.Context) error { @@ -132,9 +87,9 @@ func (s *gcpCloudTasksScaler) Close(context.Context) error { func (s *gcpCloudTasksScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec { externalMetric := &v2.ExternalMetricSource{ Metric: v2.MetricIdentifier{ - Name: GenerateMetricNameWithIndex(s.metadata.triggerIndex, kedautil.NormalizeString(fmt.Sprintf("gcp-ct-%s", s.metadata.queueName))), + Name: GenerateMetricNameWithIndex(s.metadata.triggerIndex, kedautil.NormalizeString(fmt.Sprintf("gcp-ct-%s", s.metadata.QueueName))), }, - Target: GetMetricTargetMili(s.metricType, s.metadata.value), + Target: GetMetricTargetMili(s.metricType, s.metadata.Value), } // Create the metric spec for the HPA @@ -158,7 +113,7 @@ func (s *gcpCloudTasksScaler) GetMetricsAndActivity(ctx context.Context, metricN metric := GenerateMetricInMili(metricName, value) - return []external_metrics.ExternalMetricValue{metric}, value > s.metadata.activationValue, nil + return []external_metrics.ExternalMetricValue{metric}, value > s.metadata.ActivationValue, nil } func (s *gcpCloudTasksScaler) setStackdriverClient(ctx context.Context) error { @@ -185,9 +140,9 @@ func (s *gcpCloudTasksScaler) getMetrics(ctx context.Context, metricType string) return -1, err } } - filter := `metric.type="` + metricType + `" AND resource.labels.queue_id="` + s.metadata.queueName + `"` + filter := `metric.type="` + metricType + `" AND resource.labels.queue_id="` + s.metadata.QueueName + `"` // Cloud Tasks metrics are collected every 60 seconds so no need to aggregate them. // See: https://cloud.google.com/monitoring/api/metrics_gcp#gcp-cloudtasks - return s.client.GetMetrics(ctx, filter, s.metadata.projectID, nil, nil, s.metadata.filterDuration) + return s.client.GetMetrics(ctx, filter, s.metadata.ProjectID, nil, nil, s.metadata.FilterDuration) } diff --git a/pkg/scalers/gcp_cloud_tasks_scaler_test.go b/pkg/scalers/gcp_cloud_tasks_scaler_test.go index 61772272e0a..4f3b6d84b62 100644 --- a/pkg/scalers/gcp_cloud_tasks_scaler_test.go +++ b/pkg/scalers/gcp_cloud_tasks_scaler_test.go @@ -2,10 +2,12 @@ package scalers import ( "context" + "reflect" "testing" "github.com/go-logr/logr" + "github.com/kedacore/keda/v2/pkg/scalers/gcp" "github.com/kedacore/keda/v2/pkg/scalers/scalersconfig" ) @@ -17,6 +19,8 @@ type parseGcpCloudTasksMetadataTestData struct { authParams map[string]string metadata map[string]string isError bool + expected *gcpCloudTaskMetadata + comment string } type gcpCloudTasksMetricIdentifier struct { @@ -26,25 +30,82 @@ type gcpCloudTasksMetricIdentifier struct { } var testGcpCloudTasksMetadata = []parseGcpCloudTasksMetadataTestData{ - {map[string]string{}, map[string]string{}, true}, - // all properly formed - {nil, map[string]string{"queueName": "myQueue", "value": "7", "credentialsFromEnv": "SAMPLE_CREDS", "projectID": "myproject", "activationValue": "5"}, false}, - // missing subscriptionName - {nil, map[string]string{"queueName": "", "value": "7", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true}, - // missing credentials - {nil, map[string]string{"queueName": "myQueue", "value": "7", "projectID": "myproject", "credentialsFromEnv": ""}, true}, - // malformed subscriptionSize - {nil, map[string]string{"queueName": "myQueue", "value": "AA", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true}, - // malformed mode - {nil, map[string]string{"queueName": "", "mode": "AA", "value": "7", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true}, - // malformed activationTargetValue - {nil, map[string]string{"queueName": "myQueue", "value": "7", "credentialsFromEnv": "SAMPLE_CREDS", "projectID": "myproject", "activationValue": "AA"}, true}, - // Credentials from AuthParams - {map[string]string{"GoogleApplicationCredentials": "Creds"}, map[string]string{"queueName": "myQueue", "value": "7", "projectID": "myproject"}, false}, - // Credentials from AuthParams with empty creds - {map[string]string{"GoogleApplicationCredentials": ""}, map[string]string{"queueName": "myQueue", "subscriptionSize": "7", "projectID": "myproject"}, true}, - // properly formed float value and activationTargetValue - {nil, map[string]string{"queueName": "mysubscription", "value": "7.1", "credentialsFromEnv": "SAMPLE_CREDS", "activationValue": "2.1", "projectID": "myproject"}, false}, + + {map[string]string{}, map[string]string{}, true, nil, "erro case"}, + + {nil, map[string]string{"queueName": "myQueue", "value": "7", "credentialsFromEnv": "SAMPLE_CREDS", "projectID": "myproject", "activationValue": "5"}, false, &gcpCloudTaskMetadata{ + Value: 7, + ActivationValue: 5, + FilterDuration: 0, + QueueName: "myQueue", + ProjectID: "myproject", + gcpAuthorization: &gcp.AuthorizationMetadata{ + GoogleApplicationCredentials: "{}", + PodIdentityProviderEnabled: false, + }, + triggerIndex: 0}, "all properly formed"}, + + {nil, map[string]string{"queueName": "", "value": "7", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true, nil, "missing subscriptionName"}, + + {nil, map[string]string{"queueName": "myQueue", "value": "7", "projectID": "myproject", "credentialsFromEnv": ""}, true, nil, "missing credentials"}, + + {nil, map[string]string{"queueName": "myQueue", "value": "AA", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true, nil, "malformed subscriptionSize"}, + + {nil, map[string]string{"queueName": "", "mode": "AA", "value": "7", "projectID": "myproject", "credentialsFromEnv": "SAMPLE_CREDS"}, true, nil, "malformed mode"}, + + {nil, map[string]string{"queueName": "myQueue", "value": "7", "credentialsFromEnv": "SAMPLE_CREDS", "projectID": "myproject", "activationValue": "AA"}, true, nil, "malformed activationTargetValue"}, + + {map[string]string{"GoogleApplicationCredentials": "Creds"}, map[string]string{"queueName": "myQueue", "value": "7", "projectID": "myproject"}, false, &gcpCloudTaskMetadata{ + Value: 7, + ActivationValue: 0, + FilterDuration: 0, + QueueName: "myQueue", + ProjectID: "myproject", + gcpAuthorization: &gcp.AuthorizationMetadata{ + GoogleApplicationCredentials: "Creds", + PodIdentityProviderEnabled: false, + }, + triggerIndex: 0}, "Credentials from AuthParams"}, + + {map[string]string{"GoogleApplicationCredentials": ""}, map[string]string{"queueName": "myQueue", "subscriptionSize": "7", "projectID": "myproject"}, true, nil, "Credentials from AuthParams with empty creds"}, + + {nil, map[string]string{"queueName": "mysubscription", "value": "7.1", "credentialsFromEnv": "SAMPLE_CREDS", "activationValue": "2.1", "projectID": "myproject"}, false, &gcpCloudTaskMetadata{ + Value: 7.1, + ActivationValue: 2.1, + FilterDuration: 0, + QueueName: "mysubscription", + ProjectID: "myproject", + gcpAuthorization: &gcp.AuthorizationMetadata{ + GoogleApplicationCredentials: "{}", + PodIdentityProviderEnabled: false, + }, + triggerIndex: 0}, "properly formed float value and activationTargetValue"}, + + {nil, map[string]string{"queueName": "myQueue", "projectID": "myProject", "credentialsFromEnv": "SAMPLE_CREDS"}, false, &gcpCloudTaskMetadata{ + Value: 100, + ActivationValue: 0, + FilterDuration: 0, + QueueName: "myQueue", + ProjectID: "myProject", + gcpAuthorization: &gcp.AuthorizationMetadata{ + GoogleApplicationCredentials: "{}", + PodIdentityProviderEnabled: false, + }, + triggerIndex: 0}, "test default value (100) when value is not provided"}, + + {nil, map[string]string{"queueName": "myQueue", "projectID": "myProject", "credentialsFromEnv": "SAMPLE_CREDS", "activationValue": "5"}, false, &gcpCloudTaskMetadata{ + Value: 100, + ActivationValue: 5, + FilterDuration: 0, + QueueName: "myQueue", + ProjectID: "myProject", + gcpAuthorization: &gcp.AuthorizationMetadata{ + GoogleApplicationCredentials: "{}", + PodIdentityProviderEnabled: false, + }, + triggerIndex: 0}, "test default value with specified activationVal"}, + + {nil, map[string]string{"queueName": "myQueue", "projectID": "myProject", "credentialsFromEnv": "SAMPLE_CREDS", "filterDuration": "invalid"}, true, nil, "test invalid filterDuration with default values"}, } var gcpCloudTasksMetricIdentifiers = []gcpCloudTasksMetricIdentifier{ @@ -54,13 +115,25 @@ var gcpCloudTasksMetricIdentifiers = []gcpCloudTasksMetricIdentifier{ func TestGcpCloudTasksParseMetadata(t *testing.T) { for _, testData := range testGcpCloudTasksMetadata { - _, err := parseGcpCloudTasksMetadata(&scalersconfig.ScalerConfig{AuthParams: testData.authParams, TriggerMetadata: testData.metadata, ResolvedEnv: testGcpCloudTasksResolvedEnv}) - 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") - } + t.Run(testData.comment, func(t *testing.T) { + metadata, err := parseGcpCloudTasksMetadata(&scalersconfig.ScalerConfig{ + AuthParams: testData.authParams, + TriggerMetadata: testData.metadata, + ResolvedEnv: testGcpCloudTasksResolvedEnv, + }) + + if err != nil && !testData.isError { + t.Errorf("Expected success but got error") + } + + if testData.isError && err == nil { + t.Errorf("Expected error but got success") + } + + if !testData.isError && !reflect.DeepEqual(testData.expected, metadata) { + t.Fatalf("Expected %#v but got %+#v", testData.expected, metadata) + } + }) } }