feat: scalerun support

go.mod

@@ -23,7 +23,7 @@ require (
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kubernetes v1.22.2
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
-	kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919
+	kusionstack.io/kube-api v0.7.4-0.20250922083401-278352ec5aab
 	kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6
 	kusionstack.io/resourceconsist v0.0.4
 	sigs.k8s.io/controller-runtime v0.21.0

go.sum

@@ -1021,8 +1021,8 @@ k8s.io/sample-apiserver v0.22.2/go.mod h1:h+/DIV5EmuNq4vfPr5TSXy9mIBVXXlPAKQMPbj
 k8s.io/system-validators v1.5.0/go.mod h1:bPldcLgkIUK22ALflnsXk8pvkTEndYdNuaHH6gRrl0Q=
 k8s.io/utils v0.0.0-20240102154912-e7106e64919e h1:eQ/4ljkx21sObifjzXwlPKpdGLrCfRziVtos3ofG/sQ=
 k8s.io/utils v0.0.0-20240102154912-e7106e64919e/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919 h1:EMcsFMNMZO3oW7pqGOZd1nJC6YyQSt2RrcGdpC1J1gU=
-kusionstack.io/kube-api v0.7.4-0.20250727122744-2399b387a919/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk=
+kusionstack.io/kube-api v0.7.4-0.20250922083401-278352ec5aab h1:7l7Y3YezVPBP65JtlH57IwH9XsJdV2QhZcJEcTEg3cQ=
+kusionstack.io/kube-api v0.7.4-0.20250922083401-278352ec5aab/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk=
 kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6 h1:HYE6Wa8EzSlA6UmaTLtNKUgkB2mmasp6Ul69d3/SpK0=
 kusionstack.io/kube-utils v0.2.1-0.20250613035327-11e9cdaec9d6/go.mod h1:5Uy3GCJ1JEGqZw/Sp/uVnHBJN1t9wjY6USPSZ9s4idk=
 kusionstack.io/resourceconsist v0.0.4 h1:wRqLJuNh8O4TT6p0uOklFpHUKiRdRxcAH71Sw/q9LhE=

pkg/controllers/initializers/init_rollout.go

@@ -21,6 +21,7 @@ import (
 
 	"kusionstack.io/rollout/pkg/controllers/rollout"
 	"kusionstack.io/rollout/pkg/controllers/rolloutrun"
+	"kusionstack.io/rollout/pkg/controllers/scalerun"
 )
 
 func init() {
@@ -29,4 +30,7 @@ func init() {
 
 	// init rolloutRun controller
 	utilruntime.Must(Controllers.Add(rolloutrun.ControllerName, rolloutrun.InitFunc))
+
+	// init scaleRun controller
+	utilruntime.Must(Controllers.Add(scalerun.ControllerName, scalerun.InitFunc))
 }

pkg/controllers/rolloutrun/control/control.go

@@ -19,7 +19,6 @@ package control
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strings"
 
@@ -56,7 +55,7 @@ func NewBatchReleaseControl(impl workload.Accessor, c client.Client) *BatchRelea
 func (c *BatchReleaseControl) Initialize(ctx context.Context, info *workload.Info, ownerKind, ownerName, rolloutRun string, batchIndex int32) error {
 	// pre-check
 	if err := c.control.BatchPreCheck(info.Object); err != nil {
-		return TerminalError(err)
+		return utils.TerminalError(err)
 	}
 
 	// add progressing annotation
@@ -120,7 +119,7 @@ func NewCanaryReleaseControl(impl workload.Accessor, c client.Client) *CanaryRel
 func (c *CanaryReleaseControl) Initialize(ctx context.Context, stable *workload.Info, ownerKind, ownerName, rolloutRun string) error {
 	// pre check
 	if err := c.control.CanaryPreCheck(stable.Object); err != nil {
-		return TerminalError(err)
+		return utils.TerminalError(err)
 	}
 
 	// add progressing annotation
@@ -275,32 +274,3 @@ func (c *CanaryReleaseControl) applyCanaryDefaults(canaryObj client.Object) {
 		labels[rolloutapi.CanaryResourceLabelKey] = "true"
 	})
 }
-
-// TerminalError is an error that will not be retried but still be logged
-// and recorded in metrics.
-//
-// TODO: delete this error when controller-runtime version is grather than v0.15
-func TerminalError(wrapped error) error {
-	return &terminalError{err: wrapped}
-}
-
-type terminalError struct {
-	err error
-}
-
-// This function will return nil if te.err is nil.
-func (te *terminalError) Unwrap() error {
-	return te.err
-}
-
-func (te *terminalError) Error() string {
-	if te.err == nil {
-		return "nil terminal error"
-	}
-	return "terminal error: " + te.err.Error()
-}
-
-func (te *terminalError) Is(target error) bool {
-	tp := &terminalError{}
-	return errors.As(target, &tp)
-}

pkg/controllers/rolloutrun/executor/batch.go

@@ -26,6 +26,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	"kusionstack.io/rollout/pkg/controllers/rolloutrun/control"
+	"kusionstack.io/rollout/pkg/utils"
 	"kusionstack.io/rollout/pkg/workload"
 )
 
@@ -180,7 +181,7 @@ func (e *batchExecutor) doPostStepHook(ctx *ExecutorContext) (bool, time.Duratio
 }
 
 func newWorkloadNotFoundError(ref rolloutv1alpha1.CrossClusterObjectNameReference) error {
-	return control.TerminalError(&rolloutv1alpha1.CodeReasonMessage{
+	return utils.TerminalError(&rolloutv1alpha1.CodeReasonMessage{
 		Code:    "WorkloadNotFound",
 		Reason:  "WorkloadNotFound",
 		Message: fmt.Sprintf("workload (%s) not found ", ref.String()),

pkg/controllers/rolloutrun/executor/step_lifecycle.go

@@ -26,7 +26,7 @@ import (
 	rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1"
 	ctrl "sigs.k8s.io/controller-runtime"
 
-	"kusionstack.io/rollout/pkg/controllers/rolloutrun/control"
+	"kusionstack.io/rollout/pkg/utils"
 )
 
 const (
@@ -106,7 +106,7 @@ func (e *stepStateEngine) process(ctx *ExecutorContext, currentState rolloutv1al
 	stateDone, retry, err := fn(ctx)
 	if err != nil {
 		ctx.Recorder.Eventf(ctx.RolloutRun, corev1.EventTypeWarning, "FailedRunStep", "step failed, currentState %s, err: %v", currentState, err)
-		if errors.Is(err, control.TerminalError(nil)) {
+		if errors.Is(err, utils.TerminalError(nil)) {
 			// we will stop retry if err is CodeReasonMessage
 			// TODO: change err to reconcile.TerminalError when controller-runtime supports it
 			ctx.Fail(err)

pkg/controllers/scalerun/control/control.go

@@ -0,0 +1,89 @@
+/**
+ * Copyright 2024 The KusionStack Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package control
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/go-logr/logr"
+	rolloutapi "kusionstack.io/kube-api/rollout"
+	rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1"
+	"kusionstack.io/kube-utils/multicluster/clusterinfo"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"kusionstack.io/rollout/pkg/utils"
+	"kusionstack.io/rollout/pkg/workload"
+)
+
+type BatchScaleControl struct {
+	workload workload.Accessor
+	control  workload.ScaleControl
+	client   client.Client
+}
+
+func NewBatchScaleControl(impl workload.Accessor, c client.Client) *BatchScaleControl {
+	return &BatchScaleControl{
+		workload: impl,
+		control:  impl.(workload.ScaleControl),
+		client:   c,
+	}
+}
+
+func (c *BatchScaleControl) Initialize(ctx context.Context, info *workload.Info, scaleRun string, batchIndex int32) error {
+	// add progressing annotation
+	pInfo := rolloutv1alpha1.ProgressingInfo{
+		Kind:      "RollingScale",
+		RolloutID: scaleRun,
+		Batch: &rolloutv1alpha1.BatchProgressingInfo{
+			CurrentBatchIndex: batchIndex,
+		},
+	}
+	progress, _ := json.Marshal(pInfo)
+
+	_, err := info.UpdateOnConflict(ctx, c.client, func(obj client.Object) error {
+		utils.MutateAnnotations(obj, func(annotations map[string]string) {
+			annotations[rolloutapi.AnnoRolloutProgressingInfo] = string(progress)
+		})
+		return nil
+	})
+	return err
+}
+
+func (c *BatchScaleControl) Scale(ctx context.Context, info *workload.Info, updatedReplicas int32) (bool, error) {
+	ctx = clusterinfo.WithCluster(ctx, info.ClusterName)
+	obj := info.Object
+	return utils.PatchOnConflict(ctx, c.client, c.client, obj, func() error {
+		return c.control.Scale(obj, updatedReplicas)
+	})
+}
+
+func (c *BatchScaleControl) Finalize(ctx context.Context, info *workload.Info) error {
+	// delete progressing annotation
+	changed, err := info.UpdateOnConflict(ctx, c.client, func(obj client.Object) error {
+		utils.MutateAnnotations(obj, func(annotations map[string]string) {
+			delete(annotations, rolloutapi.AnnoRolloutProgressingInfo)
+		})
+		return nil
+	})
+
+	if changed {
+		logger := logr.FromContextOrDiscard(ctx)
+		logger.Info("delete progressing info on workload", "name", info.Name, "gvk", info.GroupVersionKind.String())
+	}
+	return err
+}