Skip to content

Commit

Permalink
scale to zero with cpu/mem scaler (kedacore#4346)
Browse files Browse the repository at this point in the history
Signed-off-by: gauron99 <fridrich.david19@gmail.com>
  • Loading branch information
gauron99 authored Mar 17, 2023
1 parent 6ed564e commit b1b1be1
Show file tree
Hide file tree
Showing 8 changed files with 378 additions and 34 deletions.
15 changes: 15 additions & 0 deletions .golangci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,21 @@ issues:
- path: mongo_scaler.go
linters:
- dupl
# Exclude gci check for //+kubebuilder:scaffold:imports comments. Waiting to
# resolve https://github.com/kedacore/keda/issues/4379
- path: cmd/operator/main.go
linters:
- gci
- path: cmd/webhooks/main.go
linters:
- gci
- path: controllers/keda/suite_test.go
linters:
- gci
- path: apis/keda/v1alpha1/scaledobject_webhook_test.go
linters:
- gci


linters-settings:
funlen:
Expand Down
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio

### New

- **CPU/Memory scaler**: Add support for scale to zero if there are multiple triggers([#4269](https://github.com/kedacore/keda/issues/4269))
- TODO ([#XXX](https://github.com/kedacore/keda/issue/XXX))

### Improvements
Expand Down
27 changes: 24 additions & 3 deletions apis/keda/v1alpha1/scaledobject_webhook.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@ var scaledobjectlog = logf.Log.WithName("scaledobject-validation-webhook")
var kc client.Client
var restMapper meta.RESTMapper

var memoryString = "memory"
var cpuString = "cpu"

func (so *ScaledObject) SetupWebhookWithManager(mgr ctrl.Manager) error {
kc = mgr.GetClient()
restMapper = mgr.GetRESTMapper()
Expand Down Expand Up @@ -182,7 +185,7 @@ func verifyScaledObjects(incomingSo *ScaledObject, action string) error {
func verifyCPUMemoryScalers(incomingSo *ScaledObject, action string) error {
var podSpec *corev1.PodSpec
for _, trigger := range incomingSo.Spec.Triggers {
if trigger.Type == "cpu" || trigger.Type == "memory" {
if trigger.Type == cpuString || trigger.Type == memoryString {
if podSpec == nil {
key := types.NamespacedName{
Namespace: incomingSo.Namespace,
Expand Down Expand Up @@ -218,7 +221,7 @@ func verifyCPUMemoryScalers(incomingSo *ScaledObject, action string) error {
if conainerName != "" && container.Name != conainerName {
continue
}
if trigger.Type == "cpu" {
if trigger.Type == cpuString {
if container.Resources.Requests == nil ||
container.Resources.Requests.Cpu() == nil ||
container.Resources.Requests.Cpu().AsApproximateFloat64() == 0 {
Expand All @@ -227,7 +230,7 @@ func verifyCPUMemoryScalers(incomingSo *ScaledObject, action string) error {
prommetrics.RecordScaledObjectValidatingErrors(incomingSo.Namespace, action, "missing-requests")
return err
}
} else if trigger.Type == "memory" {
} else if trigger.Type == memoryString {
if container.Resources.Requests == nil ||
container.Resources.Requests.Memory() == nil ||
container.Resources.Requests.Memory().AsApproximateFloat64() == 0 {
Expand All @@ -238,6 +241,24 @@ func verifyCPUMemoryScalers(incomingSo *ScaledObject, action string) error {
}
}
}

// validate scaledObject with cpu/mem triggers:
// If scaled object has only cpu/mem triggers AND has minReplicaCount 0
// return an error because it will never scale to zero
scaleToZeroErr := true
for _, trig := range incomingSo.Spec.Triggers {
if trig.Type != cpuString && trig.Type != memoryString {
scaleToZeroErr = false
break
}
}

if (scaleToZeroErr && incomingSo.Spec.MinReplicaCount == nil) || (scaleToZeroErr && *incomingSo.Spec.MinReplicaCount == 0) {
err := fmt.Errorf("scaledobject has only cpu/memory triggers AND minReplica is 0 (scale to zero doesn't work in this case)")
scaledobjectlog.Error(err, "validation error")
prommetrics.RecordScaledObjectValidatingErrors(incomingSo.Namespace, action, "scale-to-zero-requirements-not-met")
return err
}
}
}
return nil
Expand Down
91 changes: 91 additions & 0 deletions apis/keda/v1alpha1/scaledobject_webhook_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -401,6 +401,54 @@ var _ = It("should validate the so creation without cpu and memory when custom r
Expect(err).ToNot(HaveOccurred())
})

var _ = It("should validate so creation when all requirements are met for scaling to zero with cpu scaler", func() {
namespaceName := "scale-to-zero-good"
namespace := createNamespace(namespaceName)
workload := createDeployment(namespaceName, true, false)

scaledobject := createScaledObjectSTZ(soName, namespaceName, workloadName, 0, 5, true)

err := k8sClient.Create(context.Background(), namespace)
Expect(err).ToNot(HaveOccurred())

err = k8sClient.Create(context.Background(), workload)
Expect(err).ToNot(HaveOccurred())

err = k8sClient.Create(context.Background(), scaledobject)
Expect(err).ToNot(HaveOccurred())
})

var _ = It("shouldn't validate so creation with cpu scaler requirements not being met for scaling to 0", func() {
namespaceName := "scale-to-zero-min-replicas-bad"
namespace := createNamespace(namespaceName)
workload := createDeployment(namespaceName, true, false)

scaledobject := createScaledObjectSTZ(soName, namespaceName, workloadName, 0, 5, false)

err := k8sClient.Create(context.Background(), namespace)
Expect(err).ToNot(HaveOccurred())
err = k8sClient.Create(context.Background(), workload)
Expect(err).ToNot(HaveOccurred())
err = k8sClient.Create(context.Background(), scaledobject)
Expect(err).To(HaveOccurred())
})

var _ = It("should validate so creation when min replicas is > 0 with only cpu scaler given", func() {
namespaceName := "scale-to-zero-no-external-trigger-good"
namespace := createNamespace(namespaceName)
workload := createDeployment(namespaceName, true, false)

scaledobject := createScaledObjectSTZ(soName, namespaceName, workloadName, 1, 5, false)

err := k8sClient.Create(context.Background(), namespace)
Expect(err).ToNot(HaveOccurred())
err = k8sClient.Create(context.Background(), workload)
Expect(err).ToNot(HaveOccurred())
err = k8sClient.Create(context.Background(), scaledobject)
Expect(err).ToNot(HaveOccurred())

})

var _ = AfterSuite(func() {
cancel()
By("tearing down the test environment")
Expand Down Expand Up @@ -595,3 +643,46 @@ func createStatefulSet(namespace string, hasCPU, hasMemory bool) *appsv1.Statefu
},
}
}

func createScaledObjectSTZ(name string, namespace string, targetName string, minReplicas int32, maxReplicas int32, hasExternalTrigger bool) *ScaledObject {
triggers := []ScaleTriggers{
{
Type: "cpu",
Metadata: map[string]string{
"value": "10",
},
},
}

if hasExternalTrigger {
kubeWorkloadTrigger := ScaleTriggers{
Type: "kubernetes-workload",
Metadata: map[string]string{
"podSelector": "pod=workload-test",
"value": "1",
},
}
triggers = append(triggers, kubeWorkloadTrigger)
}

return &ScaledObject{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
UID: types.UID(name),
},
TypeMeta: metav1.TypeMeta{
Kind: "ScaledObject",
APIVersion: "keda.sh",
},
Spec: ScaledObjectSpec{
ScaleTargetRef: &ScaleTarget{
Name: targetName,
},
MinReplicaCount: pointer.Int32(minReplicas),
MaxReplicaCount: pointer.Int32(maxReplicas),
CooldownPeriod: pointer.Int32(1),
Triggers: triggers,
},
}
}
2 changes: 1 addition & 1 deletion controllers/keda/suite_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ import (
kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
"github.com/kedacore/keda/v2/pkg/k8s"
"github.com/kedacore/keda/v2/pkg/scaling"
// +kubebuilder:scaffold:imports
//+kubebuilder:scaffold:imports
)

// These tests use Ginkgo (BDD-style Go testing framework). Refer to
Expand Down
18 changes: 16 additions & 2 deletions pkg/scaling/scale_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -550,6 +550,17 @@ func (h *scaleHandler) getScaledObjectState(ctx context.Context, scaledObject *k
return false, true, map[string]metricscache.MetricsRecord{}, fmt.Errorf("error getting scalers cache %w", err)
}

// count the number of non-external triggers (cpu/mem) in order to check for
// scale to zero requirements if atleast one cpu/mem trigger is given.
// This is calculated here because of algorithm complexity but
// evaluated in the loop below.
cpuMemCount := 0
for _, trigger := range scaledObject.Spec.Triggers {
if trigger.Type == "cpu" || trigger.Type == "memory" {
cpuMemCount++
}
}

// Let's collect status of all scalers, no matter if any scaler raises error or is active
scalers, scalerConfigs := cache.GetScalers()
for scalerIndex := 0; scalerIndex < len(scalers); scalerIndex++ {
Expand All @@ -566,9 +577,12 @@ func (h *scaleHandler) getScaledObjectState(ctx context.Context, scaledObject *k
}

for _, spec := range metricSpecs {
// skip cpu/memory resource scaler, these scalers are also always Active
// if cpu/memory resource scaler has minReplicas==0 & at least one external
// trigger exists -> object can be scaled to zero
if spec.External == nil {
isScaledObjectActive = true
if len(scaledObject.Spec.Triggers) <= cpuMemCount {
isScaledObjectActive = true
}
continue
}

Expand Down
Loading

0 comments on commit b1b1be1

Please sign in to comment.