Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

EE-114: Supprting Keda 2.7 #1

Open
wants to merge 1 commit into
base: keda_support
Choose a base branch
from
+26 −26
Open

EE-114: Supprting Keda 2.7 #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion engine/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ const (

// annotation used on resources (deployments, statefulsets...)
originalReplicas = "originalReplicas"

originalScaleTargetRefName = "originalScaleTargetRefName"
kedaAutoscalingPaused = "autoscaling.keda.sh/paused-replicas"
)

type Engine struct {
Expand Down
50 changes: 25 additions & 25 deletions engine/scaledobjects.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,8 @@ package engine

import (
"context"
"fmt"


"github.com/rs/zerolog"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand All @@ -13,26 +13,22 @@ import (
kedaclientset "github.com/kedacore/keda/v2/pkg/generated/clientset/versioned"
)

// checkRunningDeploymentsConformity verifies that all deployments within the namespace are
// currently running

func checkRunningScaledobjectsConformity(ctx context.Context, l zerolog.Logger, scaledobjects []kedav1alpha1.ScaledObject, cs *kedaclientset.Clientset, ns, prefix string) (bool, error) {
hasBeenPatched := false
for _, so := range scaledobjects {
// Check if the deployment is annotated and deployment name
origName := so.Annotations[prefix+originalScaleTargetRefName]
if origName != "" {
patchedName := fmt.Sprintf("%s-suspend", origName)
if so.Spec.ScaleTargetRef.Name != patchedName {
continue
}
l.Info().Str("scaledobject", so.Name).Msgf("changing scaleTargetRef name from %s to %s", so.Spec.ScaleTargetRef.Name, origName)
// patch the deployment
if err := patchScaledobjects(ctx, cs, ns, so.Name, prefix, origName); err != nil {
// Check if the ScaledObject is annotated
if _, exits := so.Annotations[kedaAutoscalingPaused]; exits {
l.Info().Str("scaledobjects", so.Name).Msgf("Removing annotations %s from the ScaledObject", kedaAutoscalingPaused)
origName := so.Annotations[prefix+originalScaleTargetRefName]
if err := patchScaledobjects(ctx, cs, ns, so.Name, prefix, origName, false); err != nil {
return hasBeenPatched, err
}

hasBeenPatched = true
}
}

return hasBeenPatched, nil
}

Expand All @@ -42,32 +38,36 @@ func checkSuspendedScaledobejctsConformity(ctx context.Context, l zerolog.Logger
}

for _, so := range scaledobjects {
origName := so.Annotations[prefix+originalScaleTargetRefName]
if origName == "" || so.Spec.ScaleTargetRef.Name == origName {
// TODO: what about fixing the annotation original Replicas here ?
patchedName := fmt.Sprintf("%s-suspend", so.Spec.ScaleTargetRef.Name)
l.Info().Str("scaledobjects", so.Name).Msgf("changing scaleTargetRef name from %s to %s", origName, patchedName)
// patch the deployment
if err := patchScaledobjects(ctx, cs, ns, so.Name, prefix, patchedName); err != nil {
if _, exists := so.Annotations[kedaAutoscalingPaused]; exists {
continue
} else {
l.Info().Str("scaledobject", so.Name).Msgf("Patching the ScaledObject with %s", kedaAutoscalingPaused)
if err := patchScaledobjects(ctx, cs, ns, so.Name, prefix, "", true); err != nil {
return err
}
}
}

return nil
}

// patchScaledobjects updates the minmum and maximum deployments
func patchScaledobjects(ctx context.Context, cs *kedaclientset.Clientset, ns, so, prefix string, patchedName string) error {
func patchScaledobjects(ctx context.Context, cs *kedaclientset.Clientset, ns, so, prefix string, patchedName string, suspend bool) error {
return retry.RetryOnConflict(retry.DefaultRetry, func() error {
result, err := cs.KedaV1alpha1().ScaledObjects(ns).Get(ctx, so, metav1.GetOptions{})
if err != nil {
return err
}
// If there's no annotation, set it up
if result.Annotations[prefix+originalScaleTargetRefName] == "" {
result.Annotations[prefix+originalScaleTargetRefName] = result.Spec.ScaleTargetRef.Name
if suspend {
result.Annotations[kedaAutoscalingPaused] = "0"
} else {
delete(result.Annotations, kedaAutoscalingPaused)
delete(result.Annotations, prefix+originalScaleTargetRefName)
if patchedName != "" {
result.Spec.ScaleTargetRef.Name = patchedName
}
}
result.Spec.ScaleTargetRef.Name = patchedName

_, err = cs.KedaV1alpha1().ScaledObjects(ns).Update(ctx, result, v1.UpdateOptions{})
return err
})
Expand Down