Add Kubernetes Workload scaler (kedacore#2010)

Signed-off-by: jorturfer <jorge_turrado@hotmail.es>

Co-authored-by: Zbynek Roubalik <726523+zroubalik@users.noreply.github.com>
Signed-off-by: nilayasiktoprak <nilayasiktoprak@gmail.com>

.golangci.yml

@@ -64,6 +64,11 @@ issues:
  - path: scale_resolvers_test.go
  linters:
  - staticcheck
+ # Got "sigs.k8s.io/controller-runtime/pkg/client/fake is deprecated: please use pkg/envtest for testing"
+ # This might not be ideal, see: https://github.com/kubernetes-sigs/controller-runtime/issues/768
+ - path: kubernetes_workload_scaler_test.go
+ linters:
+ - staticcheck
  # https://github.com/go-critic/go-critic/issues/926
  - linters:
  - gocritic

CHANGELOG.md

@@ -32,6 +32,7 @@
 - Add new scaler for Selenium Grid ([#1971](https://github.com/kedacore/keda/pull/1971))
 - Support using regex to select the queues in RabbitMQ Scaler ([#1957](https://github.com/kedacore/keda/pull/1957))
 - Support custom metric name in RabbitMQ Scaler ([#1976](https://github.com/kedacore/keda/pull/1976))
+- Add new scaler for Kubernetes Workload ([#2010](https://github.com/kedacore/keda/pull/2010))
 
 ### Improvements
 

pkg/scalers/kubernetes_workload_scaler.go

@@ -0,0 +1,138 @@
+package scalers
+
+import (
+ "context"
+ "fmt"
+ "strconv"
+ "strings"
+
+ "k8s.io/api/autoscaling/v2beta2"
+ corev1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/api/resource"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/labels"
+ "k8s.io/metrics/pkg/apis/external_metrics"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+
+ kedautil "github.com/kedacore/keda/v2/pkg/util"
+)
+
+type kubernetesWorkloadScaler struct {
+ metadata *kubernetesWorkloadMetadata
+ kubeClient client.Client
+}
+
+const (
+ kubernetesWorkloadMetricType = "External"
+ podSelectorKey = "podSelector"
+ valueKey = "value"
+)
+
+type kubernetesWorkloadMetadata struct {
+ podSelector labels.Selector
+ namespace string
+ value int64
+}
+
+// NewKubernetesWorkloadScaler creates a new kubernetesWorkloadScaler
+func NewKubernetesWorkloadScaler(kubeClient client.Client, config *ScalerConfig) (Scaler, error) {
+ meta, parseErr := parseWorkloadMetadata(config)
+ if parseErr != nil {
+ return nil, fmt.Errorf("error parsing kubernetes workload metadata: %s", parseErr)
+ }
+
+ return &kubernetesWorkloadScaler{
+ metadata: meta,
+ kubeClient: kubeClient,
+ }, nil
+}
+
+func parseWorkloadMetadata(config *ScalerConfig) (*kubernetesWorkloadMetadata, error) {
+ meta := &kubernetesWorkloadMetadata{}
+ var err error
+ meta.namespace = config.Namespace
+ meta.podSelector, err = labels.Parse(config.TriggerMetadata[podSelectorKey])
+ if err != nil || meta.podSelector.String() == "" {
+ return nil, fmt.Errorf("invalid pod selector")
+ }
+ meta.value, err = strconv.ParseInt(config.TriggerMetadata[valueKey], 10, 64)
+ if err != nil || meta.value == 0 {
+ return nil, fmt.Errorf("value must be an integer greater than 0")
+ }
+ return meta, nil
+}
+
+// IsActive determines if we need to scale from zero
+func (s *kubernetesWorkloadScaler) IsActive(ctx context.Context) (bool, error) {
+ pods, err := s.getMetricValue(ctx)
+
+ if err != nil {
+ return false, err
+ }
+
+ return pods > 0, nil
+}
+
+// Close no need for kubernetes workload scaler
+func (s *kubernetesWorkloadScaler) Close() error {
+ return nil
+}
+
+// GetMetricSpecForScaling returns the metric spec for the HPA
+func (s *kubernetesWorkloadScaler) GetMetricSpecForScaling() []v2beta2.MetricSpec {
+ targetMetricValue := resource.NewQuantity(s.metadata.value, resource.DecimalSI)
+ externalMetric := &v2beta2.ExternalMetricSource{
+ Metric: v2beta2.MetricIdentifier{
+ Name: kedautil.NormalizeString(fmt.Sprintf("%s-%s-%s", "workload", s.metadata.namespace, normalizeSelectorString(s.metadata.podSelector))),
+ },
+ Target: v2beta2.MetricTarget{
+ Type: v2beta2.AverageValueMetricType,
+ AverageValue: targetMetricValue,
+ },
+ }
+ metricSpec := v2beta2.MetricSpec{External: externalMetric, Type: kubernetesWorkloadMetricType}
+ return []v2beta2.MetricSpec{metricSpec}
+}
+
+// GetMetrics returns value for a supported metric
+func (s *kubernetesWorkloadScaler) GetMetrics(ctx context.Context, metricName string, metricSelector labels.Selector) ([]external_metrics.ExternalMetricValue, error) {
+ pods, err := s.getMetricValue(ctx)
+ if err != nil {
+ return []external_metrics.ExternalMetricValue{}, fmt.Errorf("error inspecting kubernetes workload: %s", err)
+ }
+
+ metric := external_metrics.ExternalMetricValue{
+ MetricName: metricName,
+ Value: *resource.NewQuantity(int64(pods), resource.DecimalSI),
+ Timestamp: metav1.Now(),
+ }
+
+ return append([]external_metrics.ExternalMetricValue{}, metric), nil
+}
+
+func (s *kubernetesWorkloadScaler) getMetricValue(ctx context.Context) (int, error) {
+ podList := &corev1.PodList{}
+ listOptions := client.ListOptions{}
+ listOptions.LabelSelector = s.metadata.podSelector
+ listOptions.Namespace = s.metadata.namespace
+ opts := []client.ListOption{
+ &listOptions,
+ }
+
+ err := s.kubeClient.List(ctx, podList, opts...)
+ if err != nil {
+ return 0, err
+ }
+
+ return len(podList.Items), nil
+}
+
+func normalizeSelectorString(selector labels.Selector) string {
+ s := selector.String()
+ s = strings.ReplaceAll(s, " ", "")
+ s = strings.ReplaceAll(s, "(", "-")
+ s = strings.ReplaceAll(s, ")", "-")
+ s = strings.ReplaceAll(s, ",", "-")
+ s = strings.ReplaceAll(s, "!", "-")
+ return s
+}

pkg/scalers/kubernetes_workload_scaler_test.go

@@ -0,0 +1,138 @@
+package scalers
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ v1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+type workloadMetadataTestData struct {
+ metadata map[string]string
+ namespace string
+ isError bool
+}
+
+var parseWorkloadMetadataTestDataset = []workloadMetadataTestData{
+ {map[string]string{"value": "1", "podSelector": "app=demo"}, "test", false},
+ {map[string]string{"value": "1", "podSelector": "app=demo"}, "default", false},
+ {map[string]string{"value": "1", "podSelector": "app in (demo1, demo2)"}, "test", false},
+ {map[string]string{"value": "1", "podSelector": "app in (demo1, demo2),deploy in (deploy1, deploy2)"}, "test", false},
+ {map[string]string{"podSelector": "app=demo"}, "test", true},
+ {map[string]string{"podSelector": "app=demo"}, "default", true},
+ {map[string]string{"value": "1"}, "test", true},
+ {map[string]string{"value": "1"}, "default", true},
+ {map[string]string{"value": "a", "podSelector": "app=demo"}, "test", true},
+ {map[string]string{"value": "a", "podSelector": "app=demo"}, "default", true},
+ {map[string]string{"value": "0", "podSelector": "app=demo"}, "test", true},
+ {map[string]string{"value": "0", "podSelector": "app=demo"}, "default", true},
+}
+
+func TestParseWorkloadMetadata(t *testing.T) {
+ for _, testData := range parseWorkloadMetadataTestDataset {
+ _, err := parseWorkloadMetadata(&ScalerConfig{TriggerMetadata: testData.metadata, Namespace: testData.namespace})
+ 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")
+ }
+ }
+}
+
+type workloadIsActiveTestData struct {
+ metadata map[string]string
+ namespace string
+ podCount int
+ active bool
+}
+
+var isActiveWorkloadTestDataset = []workloadIsActiveTestData{
+ // "podSelector": "app=demo", "namespace": "test"
+ {parseWorkloadMetadataTestDataset[0].metadata, parseWorkloadMetadataTestDataset[0].namespace, 0, false},
+ {parseWorkloadMetadataTestDataset[0].metadata, parseWorkloadMetadataTestDataset[0].namespace, 1, false},
+ {parseWorkloadMetadataTestDataset[0].metadata, parseWorkloadMetadataTestDataset[0].namespace, 15, false},
+ // "podSelector": "app=demo", "namespace": "default"
+ {parseWorkloadMetadataTestDataset[1].metadata, parseWorkloadMetadataTestDataset[1].namespace, 0, false},
+ {parseWorkloadMetadataTestDataset[1].metadata, parseWorkloadMetadataTestDataset[1].namespace, 1, true},
+ {parseWorkloadMetadataTestDataset[1].metadata, parseWorkloadMetadataTestDataset[1].namespace, 15, true},
+}
+
+func TestWorkloadIsActive(t *testing.T) {
+ for _, testData := range isActiveWorkloadTestDataset {
+ s, _ := NewKubernetesWorkloadScaler(
+ fake.NewFakeClient(createPodlist(testData.podCount)),
+ &ScalerConfig{
+ TriggerMetadata: testData.metadata,
+ AuthParams: map[string]string{},
+ GlobalHTTPTimeout: 1000 * time.Millisecond,
+ Namespace: testData.namespace,
+ },
+ )
+ isActive, _ := s.IsActive(context.TODO())
+ if testData.active && !isActive {
+ t.Error("Expected active but got inactive")
+ }
+ if !testData.active && isActive {
+ t.Error("Expected inactive but got active")
+ }
+ }
+}
+
+type workloadGetMetricSpecForScalingTestData struct {
+ metadata map[string]string
+ namespace string
+ name string
+}
+
+var getMetricSpecForScalingTestDataset = []workloadGetMetricSpecForScalingTestData{
+ // "podSelector": "app=demo", "namespace": "test"
+ {parseWorkloadMetadataTestDataset[0].metadata, parseWorkloadMetadataTestDataset[0].namespace, "workload-test-app=demo"},
+ // "podSelector": "app=demo", "namespace": "default"
+ {parseWorkloadMetadataTestDataset[1].metadata, parseWorkloadMetadataTestDataset[1].namespace, "workload-default-app=demo"},
+ // "podSelector": "app in (demo1, demo2)", "namespace": "test"
+ {parseWorkloadMetadataTestDataset[2].metadata, parseWorkloadMetadataTestDataset[2].namespace, "workload-test-appin-demo1-demo2-"},
+ // "podSelector": "app in (demo1, demo2),deploy in (deploy1, deploy2)", "namespace": "test"
+ {parseWorkloadMetadataTestDataset[3].metadata, parseWorkloadMetadataTestDataset[3].namespace, "workload-test-appin-demo1-demo2--deployin-deploy1-deploy2-"},
+}
+
+func TestWorkloadGetMetricSpecForScaling(t *testing.T) {
+ for _, testData := range getMetricSpecForScalingTestDataset {
+ s, _ := NewKubernetesWorkloadScaler(
+ fake.NewFakeClient(),
+ &ScalerConfig{
+ TriggerMetadata: testData.metadata,
+ AuthParams: map[string]string{},
+ GlobalHTTPTimeout: 1000 * time.Millisecond,
+ Namespace: testData.namespace,
+ },
+ )
+ metric := s.GetMetricSpecForScaling()
+
+ if metric[0].External.Metric.Name != testData.name {
+ t.Errorf("Expected '%s' as metric name and got '%s'", testData.name, metric[0].External.Metric.Name)
+ }
+ }
+}
+
+func createPodlist(count int) *v1.PodList {
+ list := &v1.PodList{}
+ for i := 0; i < count; i++ {
+ pod := &v1.Pod{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: fmt.Sprintf("demo-pod-v%d", i),
+ Namespace: "default",
+ Annotations: map[string]string{},
+ Labels: map[string]string{
+ "app": "demo",
+ },
+ },
+ }
+ list.Items = append(list.Items, *pod)
+ }
+ return list
+}

pkg/scaling/scale_handler.go

@@ -373,7 +373,7 @@ func (h *scaleHandler) buildScalers(withTriggers *kedav1alpha1.WithTriggers, pod
  return []scalers.Scaler{}, err
  }
 
- scaler, err := buildScaler(trigger.Type, config)
+ scaler, err := buildScaler(h.client, trigger.Type, config)
  if err != nil {
  closeScalers(scalersRes)
  h.recorder.Event(withTriggers, corev1.EventTypeWarning, eventreason.KEDAScalerFailed, err.Error())
@@ -386,7 +386,7 @@ func (h *scaleHandler) buildScalers(withTriggers *kedav1alpha1.WithTriggers, pod
  return scalersRes, nil
 }
 
-func buildScaler(triggerType string, config *scalers.ScalerConfig) (scalers.Scaler, error) {
+func buildScaler(client client.Client, triggerType string, config *scalers.ScalerConfig) (scalers.Scaler, error) {
  // TRIGGERS-START
  switch triggerType {
  case "artemis-queue":
@@ -431,6 +431,8 @@ func buildScaler(triggerType string, config *scalers.ScalerConfig) (scalers.Scal
  return scalers.NewInfluxDBScaler(config)
  case "kafka":
  return scalers.NewKafkaScaler(config)
+ case "kubernetes-workload":
+ return scalers.NewKubernetesWorkloadScaler(client, config)
  case "liiklus":
  return scalers.NewLiiklusScaler(config)
  case "memory":