SRE-1763: Scale down ReplicaSet according to RolloutScaleDown

api/v1alpha1/rolloutscaledown_types.go

@@ -20,14 +20,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
-// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
-
 // RolloutScaleDownSpec defines the desired state of RolloutScaleDown
 type RolloutScaleDownSpec struct {
-	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
-	// Important: Run "make" to regenerate code after modifying this file
-
 	// Foo is an example field of RolloutScaleDown. Edit rolloutscaledown_types.go to remove/update
 	TargetRollout    string `json:"targetRollout,omitempty"`
 	TerminatePerOnce int    `json:"terminatePerOnce,1"`
@@ -36,8 +30,7 @@ type RolloutScaleDownSpec struct {
 
 // RolloutScaleDownStatus defines the observed state of RolloutScaleDown
 type RolloutScaleDownStatus struct {
-	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
-	// Important: Run "make" to regenerate code after modifying this file
+	LastScaleDownTime metav1.Time `json:"lastScaleDownTime,omitempty"`
 }
 
 //+kubebuilder:object:root=true

cmd/main.go

@@ -23,13 +23,13 @@ import (
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
+	"k8s.io/klog/v2"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
-	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	argorolloutsapiv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
@@ -60,14 +60,11 @@ func main() {
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
-	opts := zap.Options{
-		Development: true,
-	}
-	opts.BindFlags(flag.CommandLine)
-	flag.Parse()
-
-	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
+	klog.InitFlags(flag.CommandLine)
+	flag.Parse()
+	logger := klog.NewKlogr()
+	ctrl.SetLogger(logger)
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,
 		Metrics:                metricsserver.Options{BindAddress: metricsAddr},
@@ -94,6 +91,7 @@ func main() {
 	if err = (&controller.ArgoRolloutReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Recorder: mgr.GetEventRecorderFor("rollout-optimizer"),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ArgoRollout")
 		os.Exit(1)

config/crd/bases/rollout.ovice.com_rolloutscaledowns.yaml

@@ -3,7 +3,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.13.0
+    controller-gen.kubebuilder.io/version: v0.14.0
   name: rolloutscaledowns.rollout.ovice.com
 spec:
   group: rollout.ovice.com
@@ -20,14 +20,19 @@ spec:
         description: RolloutScaleDown is the Schema for the rolloutscaledowns API
         properties:
           apiVersion:
-            description: 'APIVersion defines the versioned schema of this representation
-              of an object. Servers should convert recognized schemas to the latest
-              internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
+            description: |-
+              APIVersion defines the versioned schema of this representation of an object.
+              Servers should convert recognized schemas to the latest internal value, and
+              may reject unrecognized values.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
             type: string
           kind:
-            description: 'Kind is a string value representing the REST resource this
-              object represents. Servers may infer this from the endpoint the client
-              submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+            description: |-
+              Kind is a string value representing the REST resource this object represents.
+              Servers may infer this from the endpoint the client submits requests to.
+              Cannot be updated.
+              In CamelCase.
+              More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
             type: string
           metadata:
             type: object
@@ -48,6 +53,10 @@ spec:
             type: object
           status:
             description: RolloutScaleDownStatus defines the observed state of RolloutScaleDown
+            properties:
+              lastScaleDownTime:
+                format: date-time
+                type: string
             type: object
         type: object
     served: true

config/rbac/role.yaml

@@ -4,6 +4,15 @@ kind: ClusterRole
 metadata:
   name: manager-role
 rules:
+- apiGroups:
+  - apps
+  resources:
+  - replicasets
+  verbs:
+  - get
+  - list
+  - update
+  - watch
 - apiGroups:
   - argoproj.io
   resources:

config/samples/argorollout.yaml

@@ -15,7 +15,7 @@ spec:
     spec:
       containers:
       - name: nginx
-        image: nginx:1.25.0
+        image: nginx:latest
         imagePullPolicy: Always
         ports:
         - containerPort: 80

config/samples/rollout_v1alpha1_rolloutscaledown.yaml

@@ -11,4 +11,4 @@ metadata:
 spec:
   targetRollout: "rollout-sample"
   terminatePerOnce: 1
-  coolTimeSeconds: 40
+  coolTimeSeconds: 300

internal/controller/argorollout_controller.go

@@ -2,35 +2,42 @@ package controller
 
 import (
 	"context"
+	"time"
 
 	argorolloutsapiv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/log"
+	klog "k8s.io/klog/v2"
+	"k8s.io/client-go/tools/record"
 
 	optimizerv1alpha1 "github.com/oviceinc/rollout-optimizer-controller/api/v1alpha1"
 )
 
 type ArgoRolloutReconciler struct {
 	client.Client
 	Scheme *runtime.Scheme
+	Recorder record.EventRecorder
 }
 
 //+kubebuilder:rbac:groups=argoproj.io,resources=rollouts,verbs=get;list;watch
 //+kubebuilder:rbac:groups=argoproj.io,resources=rollouts/status,verbs=get
 //+kubebuilder:rbac:groups=rollout.ovice.com,resources=rolloutscaledowns,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=rollout.ovice.com,resources=rolloutscaledowns/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=rollout.ovice.com,resources=rolloutscaledowns/finalizers,verbs=update
+//+kubebuilder:rbac:groups=apps,resources=replicasets,verbs=get;list;watch;update
 
 func (r *ArgoRolloutReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	logger := log.FromContext(ctx)
 	scaleDownList := &optimizerv1alpha1.RolloutScaleDownList{}
 	err := r.List(ctx, scaleDownList)
 	if err != nil {
-		logger.V(1).Error(err, "Faild to list RolloutScaleDown")
+		klog.Errorf("Failed to list RolloutScaleDown: %v", err)
 		return ctrl.Result{}, err
 	}
 	var targetScaleDown *optimizerv1alpha1.RolloutScaleDown
@@ -41,25 +48,118 @@ func (r *ArgoRolloutReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		}
 	}
 	if targetScaleDown == nil {
-		logger.V(2).Info("Not target of ScaleDown", req.Namespace, req.Name)
+		klog.V(1).Infof("No target of ScaleDown: %s/%s", req.Namespace, req.Name)
 		return ctrl.Result{}, nil
 	}
 	argoRollout := &argorolloutsapiv1alpha1.Rollout{}
 	err = r.Get(ctx, types.NamespacedName{Name: req.Name, Namespace: req.Namespace}, argoRollout)
 	if err != nil && errors.IsNotFound(err) {
-		logger.V(1).Info("Not found ArgoRollout", req.Namespace, req.Name)
+		klog.V(1).Infof("Not found ArgoRollout: %s/%s", req.Namespace, req.Name)
 		return ctrl.Result{}, nil
 	} else if err != nil {
-		logger.V(1).Error(err, "Failed to get ArgoRollout", req.Namespace, req.Name)
+		klog.Errorf("Failed to get ArgoRollout: %v", err)
 		return ctrl.Result{}, err
 	}
 
-	logger.V(1).Info("Target ArgoRollout", argoRollout.Namespace, argoRollout.Name)
-	// TODO: Check ArgoRollout status and scaledown if needed
+	klog.Infof("Target ArgoRollout: %s/%s", argoRollout.Namespace, argoRollout.Name)
+	if isCompleted(&argoRollout.Status) {
+		klog.V(1).Infof("Rollout is completed: %s/%s", argoRollout.Namespace, argoRollout.Name)
+		rsHash := argoRollout.Status.StableRS
+		oldRS := []*appsv1.ReplicaSet{}
+		RSList := &appsv1.ReplicaSetList{}
+		if err := r.List(ctx, RSList, &client.ListOptions{LabelSelector: labels.SelectorFromSet(map[string]string{"app": argoRollout.Name})}); err != nil {
+			klog.Errorf("Failed to list ReplicaSet: %v", err)
+			return ctrl.Result{}, err
+		}
+
+		for i := range RSList.Items {
+			rs := RSList.Items[i]
+			if rs.Name == argoRollout.Name + "-" + rsHash {
+				continue
+			}
+			if rs.Status.Replicas > 0 {
+				oldRS = append(oldRS, &rs)
+			}
+		}
+		if len(oldRS) == 0 {
+			klog.V(1).Infof("Old ReplicaSet is not found: %s/%s", argoRollout.Namespace, argoRollout.Name)
+			return ctrl.Result{}, nil
+		}
 
+		for i := range oldRS {
+			rs := oldRS[i]
+			if rs.Status.Replicas == 0 {
+				klog.V(1).Infof("Old ReplicaSet is already scaled down: %s/%s", rs.Namespace, rs.Name)
+				continue
+			}
+			klog.V(1).Infof("Old ReplicaSet: %s/%s", rs.Namespace, rs.Name)
+
+			retry, err := r.scaleDown(ctx, rs, targetScaleDown)
+			if err != nil {
+				klog.Errorf("Failed to scale down ReplicaSet: %v", err)
+				return ctrl.Result{}, err
+			}
+			if retry > 0 {
+				return ctrl.Result{RequeueAfter: retry, Requeue: true}, nil
+			}
+
+		}
+	}
 	return ctrl.Result{}, nil
 }
 
+
 func (r *ArgoRolloutReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).For(&argorolloutsapiv1alpha1.Rollout{}).Complete(r)
 }
+
+func isCompleted(status *argorolloutsapiv1alpha1.RolloutStatus) bool {
+	if status.Phase != argorolloutsapiv1alpha1.RolloutPhaseHealthy {
+		return false
+	}
+	completed := false
+	progressing := false
+	for _, cond := range status.Conditions {
+		if cond.Type == argorolloutsapiv1alpha1.RolloutCompleted && cond.Status == corev1.ConditionTrue {
+			completed = true
+		}
+		if cond.Type == argorolloutsapiv1alpha1.RolloutProgressing && cond.Status == corev1.ConditionTrue {
+			progressing = true
+		}
+	}
+	if completed == true && progressing == true && status.Replicas >= status.ReadyReplicas {
+		return true
+	}
+	return false
+}
+
+func (r *ArgoRolloutReconciler) scaleDown(ctx context.Context, rs *appsv1.ReplicaSet, scaleDown *optimizerv1alpha1.RolloutScaleDown) (time.Duration, error) {
+	lastUpdated := scaleDown.Status.LastScaleDownTime
+	if lastUpdated.Add(time.Duration(scaleDown.Spec.CoolTimeSeconds) * time.Second).After(time.Now()) {
+		return 30 * time.Second, nil
+	}
+
+	if *rs.Spec.Replicas == 0 {
+		return 0, nil
+	}
+	replicas := *rs.Spec.Replicas - int32(scaleDown.Spec.TerminatePerOnce)
+
+	newStatus := scaleDown.Status.DeepCopy()
+	newStatus.LastScaleDownTime = metav1.Now()
+	scaleDown.Status = *newStatus
+	if err := r.Status().Update(ctx, scaleDown); err != nil {
+		klog.Errorf("Failed to update RolloutScaleDown: %v", err)
+		return 0, err
+	}
+
+	r.Recorder.Eventf(scaleDown, corev1.EventTypeNormal, "ScaleDown", "Scaling down %s/%s replicas %d -> %d", rs.Namespace, rs.Name, *rs.Spec.Replicas, replicas)
+	klog.Infof("Scaling down ReplicaSet %s/%s replicas %d -> %d", rs.Namespace, rs.Name, *rs.Spec.Replicas, replicas)
+	rs.Spec.Replicas = &replicas
+	if err := r.Update(ctx, rs); err != nil {
+		klog.Errorf("Failed to update ReplicaSet: %v", err)
+		return 0, err
+	}
+
+
+	return 0, nil
+}