diff --git a/go.mod b/go.mod index 24d4e81..86605e3 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3f31474..c3e71b3 100644 --- a/go.sum +++ b/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= diff --git a/pkg/controllers/initializers/init_rollout.go b/pkg/controllers/initializers/init_rollout.go index 04332f8..2977017 100644 --- a/pkg/controllers/initializers/init_rollout.go +++ b/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)) } diff --git a/pkg/controllers/rolloutrun/control/control.go b/pkg/controllers/rolloutrun/control/control.go index ffa8e2e..9e1c4e3 100644 --- a/pkg/controllers/rolloutrun/control/control.go +++ b/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) -} diff --git a/pkg/controllers/rolloutrun/executor/batch.go b/pkg/controllers/rolloutrun/executor/batch.go index e619a02..275fd43 100644 --- a/pkg/controllers/rolloutrun/executor/batch.go +++ b/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()), diff --git a/pkg/controllers/rolloutrun/executor/step_lifecycle.go b/pkg/controllers/rolloutrun/executor/step_lifecycle.go index 588b8e8..f0e497f 100644 --- a/pkg/controllers/rolloutrun/executor/step_lifecycle.go +++ b/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) diff --git a/pkg/controllers/scalerun/control/control.go b/pkg/controllers/scalerun/control/control.go new file mode 100644 index 0000000..a66b14d --- /dev/null +++ b/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 +} diff --git a/pkg/controllers/scalerun/executor/batch.go b/pkg/controllers/scalerun/executor/batch.go new file mode 100644 index 0000000..d7bd02a --- /dev/null +++ b/pkg/controllers/scalerun/executor/batch.go @@ -0,0 +1,278 @@ +/** + * 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 + * + * http://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 executor + +import ( + "fmt" + "time" + + utilerrors "k8s.io/apimachinery/pkg/util/errors" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + ctrl "sigs.k8s.io/controller-runtime" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/controllers/scalerun/control" + "kusionstack.io/rollout/pkg/utils" + "kusionstack.io/rollout/pkg/workload" +) + +type batchExecutor struct { + webhook webhookExecutor + stateEngine *stepStateEngine +} + +func newBatchExecutor(webhook webhookExecutor) *batchExecutor { + e := &batchExecutor{ + webhook: webhook, + stateEngine: newStepStateEngine(), + } + + e.stateEngine.add(rorexecutor.StepNone, rorexecutor.StepPending, e.doPausing, e.release) + e.stateEngine.add(rorexecutor.StepPending, rorexecutor.StepPreBatchStepHook, skipStep, e.release) + e.stateEngine.add(rorexecutor.StepPreBatchStepHook, rorexecutor.StepRunning, e.doPreStepHook, e.release) + e.stateEngine.add(rorexecutor.StepRunning, rorexecutor.StepPostBatchStepHook, e.doBatchUpgrading, e.release) + e.stateEngine.add(rorexecutor.StepPostBatchStepHook, rorexecutor.StepResourceRecycling, e.doPostStepHook, e.release) + e.stateEngine.add(rorexecutor.StepResourceRecycling, rorexecutor.StepSucceeded, e.doRecycle, e.release) + e.stateEngine.add(rorexecutor.StepSucceeded, "", skipStep, skipStep) + return e +} + +func (e *batchExecutor) init(ctx *ExecutorContext) bool { + logger := ctx.GetBatchLogger() + if !e.isSupported(ctx) { + // skip batch scale if workload accessor don't support it. + logger.Info("workload accessor don't support batch scale, skip it") + ctx.SkipCurrentRelease() + return true + } + return false +} + +func (e *batchExecutor) Do(ctx *ExecutorContext) (done bool, result ctrl.Result, err error) { + if e.init(ctx) { + return true, ctrl.Result{Requeue: true}, nil + } + newStatus := ctx.NewStatus + currentBatchIndex := newStatus.Batches.CurrentBatchIndex + currentState := newStatus.Batches.CurrentBatchState + + stepDone, result, err := e.stateEngine.do(ctx, currentState) + if err != nil { + return false, result, err + } + if !stepDone { + return false, result, nil + } + + if int(currentBatchIndex+1) < len(ctx.ScaleRun.Spec.Batch.Batches) { + // move to next batch + newStatus.Batches.CurrentBatchState = rorexecutor.StepNone + newStatus.Batches.CurrentBatchIndex = currentBatchIndex + 1 + return false, result, nil + } + + return true, result, nil +} + +func (e *batchExecutor) Cancel(ctx *ExecutorContext) (done bool, result ctrl.Result, err error) { + done = e.init(ctx) + if done { + return true, ctrl.Result{Requeue: true}, nil + } + return e.stateEngine.cancel(ctx, ctx.NewStatus.Batches.CurrentBatchState) +} + +func (e *batchExecutor) isSupported(ctx *ExecutorContext) bool { + _, ok := ctx.Accessor.(workload.ScaleControl) + return ok +} + +func (e *batchExecutor) release(ctx *ExecutorContext) (bool, time.Duration, error) { + // frstly try to stop webhook + e.webhook.Cancel(ctx) + + // try to finalize all workloads + allTargets := map[rolloutv1alpha1.CrossClusterObjectNameReference]bool{} + // finalize batch release + batchControl := control.NewBatchScaleControl(ctx.Accessor, ctx.Client) + + for _, item := range ctx.ScaleRun.Spec.Batch.Batches { + for _, target := range item.Targets { + allTargets[target.CrossClusterObjectNameReference] = true + } + } + + var finalizeErrs []error + + for target := range allTargets { + wi := ctx.Workloads.Get(target.Cluster, target.Name) + if wi == nil { + // ignore not found workload + continue + } + err := batchControl.Finalize(ctx, wi) + if err != nil { + // try our best to finalize all workloasd + finalizeErrs = append(finalizeErrs, err) + continue + } + } + + if len(finalizeErrs) > 0 { + return false, retryDefault, utilerrors.NewAggregate(finalizeErrs) + } + + return true, retryImmediately, nil +} + +func (e *batchExecutor) doRecycle(ctx *ExecutorContext) (bool, time.Duration, error) { + // recycling only work on last batch now + if int(ctx.NewStatus.Batches.CurrentBatchIndex+1) < len(ctx.ScaleRun.Spec.Batch.Batches) { + return true, retryImmediately, nil + } + return e.release(ctx) +} + +func (e *batchExecutor) doPausing(ctx *ExecutorContext) (bool, time.Duration, error) { + scaleRunName := ctx.ScaleRun.Name + newStatus := ctx.NewStatus + currentBatchIndex := newStatus.Batches.CurrentBatchIndex + currentBatch := ctx.ScaleRun.Spec.Batch.Batches[currentBatchIndex] + + batchControl := control.NewBatchScaleControl(ctx.Accessor, ctx.Client) + + for _, item := range currentBatch.Targets { + wi := ctx.Workloads.Get(item.Cluster, item.Name) + if wi == nil { + return false, retryStop, newWorkloadNotFoundError(item.CrossClusterObjectNameReference) + } + err := batchControl.Initialize(ctx, wi, scaleRunName, currentBatchIndex) + if err != nil { + return false, retryStop, err + } + } + + if ctx.ScaleRun.Spec.Batch.Batches[currentBatchIndex].Breakpoint { + ctx.Pause() + } + return true, retryImmediately, nil +} + +func (e *batchExecutor) doPreStepHook(ctx *ExecutorContext) (bool, time.Duration, error) { + return e.webhook.Do(ctx, rolloutv1alpha1.PreBatchStepHook) +} + +func (e *batchExecutor) doPostStepHook(ctx *ExecutorContext) (bool, time.Duration, error) { + return e.webhook.Do(ctx, rolloutv1alpha1.PostBatchStepHook) +} + +func newWorkloadNotFoundError(ref rolloutv1alpha1.CrossClusterObjectNameReference) error { + return utils.TerminalError(&rolloutv1alpha1.CodeReasonMessage{ + Code: "WorkloadNotFound", + Reason: "WorkloadNotFound", + Message: fmt.Sprintf("workload (%s) not found ", ref.String()), + }) +} + +// doBatchUpgrading process upgrading state +func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Duration, error) { + scaleRun := ctx.ScaleRun + newStatus := ctx.NewStatus + currentBatchIndex := newStatus.Batches.CurrentBatchIndex + currentBatch := scaleRun.Spec.Batch.Batches[currentBatchIndex] + currentBatchWorkloadStatus := newStatus.Batches.Records[currentBatchIndex].Targets + + logger := ctx.GetBatchLogger() + + batchControl := control.NewBatchScaleControl(ctx.Accessor, ctx.Client) + + batchTargetStatuses := make([]rolloutv1alpha1.ScaleWorkloadStatus, 0) + + allWorkloadReady := true + for _, item := range currentBatch.Targets { + info := ctx.Workloads.Get(item.Cluster, item.Name) + if info == nil { + // If the target workload does not exist, the retries will stop. + return false, retryStop, newWorkloadNotFoundError(item.CrossClusterObjectNameReference) + } + + needApplyReplicas := false + + workloadStatus := info.ScaleWorkloadStatus() + currentWorkloadStatus := e.findCurrentWorkloadStatus(currentBatchWorkloadStatus, item.Cluster, item.Name) + if currentWorkloadStatus == nil { + workloadStatus.ScaleFrom = info.Status.Replicas + workloadStatus.ScaleTo = item.Replicas + needApplyReplicas = true + } else { + workloadStatus.ScaleFrom = currentWorkloadStatus.ScaleFrom + workloadStatus.ScaleTo = currentWorkloadStatus.ScaleTo + } + batchTargetStatuses = append(batchTargetStatuses, workloadStatus) + + if e.checkScaledReady(info, workloadStatus.ScaleFrom, workloadStatus.ScaleTo) && !needApplyReplicas { + // if the target is ready, we will not change replicas + continue + } + + allWorkloadReady = false + if !needApplyReplicas { + // if the target's replicas has been updated, we will not change replicas + continue + } + + logger.Info("need to apply target replicas", "target", item.CrossClusterObjectNameReference) + changed, err := batchControl.Scale(ctx, info, item.Replicas) + if err != nil { + return false, retryStop, err + } + if changed { + logger.V(2).Info("upgrade target replicas", "target", item.CrossClusterObjectNameReference, "replicas", item.Replicas) + } + } + + // update target status in batch + newStatus.Batches.Records[currentBatchIndex].Targets = batchTargetStatuses + + if allWorkloadReady { + return true, retryImmediately, nil + } + + // wait for next reconcile + return false, retryDefault, nil +} + +func (e *batchExecutor) checkScaledReady(info *workload.Info, scaledFrom, scaledTo int32) bool { + if info.Status.ObservedGeneration != info.Generation { + return false + } + + if scaledFrom > scaledTo { + return info.Status.CurrentReplicas <= info.Status.Replicas + } else { + return info.Status.AvailableReplicas >= info.Status.Replicas + } +} + +func (e *batchExecutor) findCurrentWorkloadStatus(status []rolloutv1alpha1.ScaleWorkloadStatus, cluster, name string) *rolloutv1alpha1.ScaleWorkloadStatus { + for _, workloadStatus := range status { + if workloadStatus.Cluster == cluster && workloadStatus.Name == name { + return &workloadStatus + } + } + return nil +} diff --git a/pkg/controllers/scalerun/executor/context.go b/pkg/controllers/scalerun/executor/context.go new file mode 100644 index 0000000..5f5c27e --- /dev/null +++ b/pkg/controllers/scalerun/executor/context.go @@ -0,0 +1,273 @@ +/** + * 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 + * + * http://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 executor + +import ( + "context" + "slices" + "sync" + + "github.com/go-logr/logr" + "github.com/samber/lo" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/tools/record" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + "sigs.k8s.io/controller-runtime/pkg/client" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/workload" +) + +// ExecutorContext context of scaleRun +type ExecutorContext struct { + context.Context + once sync.Once + Client client.Client + Recorder record.EventRecorder + + Accessor workload.Accessor + ScaleRun *rolloutv1alpha1.ScaleRun + NewStatus *rolloutv1alpha1.ScaleRunStatus + Workloads *workload.Set +} + +func (c *ExecutorContext) Initialize() { + c.once.Do(func() { + if c.NewStatus == nil { + c.NewStatus = c.ScaleRun.Status.DeepCopy() + } + newStatus := c.NewStatus + newStatus.ObservedGeneration = c.ScaleRun.Generation + + if len(newStatus.Phase) == 0 { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseInitial + } + + // init BatchStatus + if c.ScaleRun.Spec.Batch != nil { + if newStatus.Batches == nil { + newStatus.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{} + } + // resize records + specBatchSize := len(c.ScaleRun.Spec.Batch.Batches) + statusBatchSize := len(newStatus.Batches.Records) + if specBatchSize > statusBatchSize { + for i := 0; i < specBatchSize-statusBatchSize; i++ { + newStatus.Batches.Records = append(newStatus.Batches.Records, + rolloutv1alpha1.ScaleRunStepStatus{ + Index: ptr.To(int32(statusBatchSize + i)), + State: rorexecutor.StepNone, + }, + ) + } + } else if specBatchSize < statusBatchSize { + newStatus.Batches.Records = newStatus.Batches.Records[:specBatchSize] + } + } + }) +} + +// filterWebhooks return webhooks met hookType +func filterWebhooks(hookType rolloutv1alpha1.HookType, scaleRun *rolloutv1alpha1.ScaleRun) []rolloutv1alpha1.RolloutWebhook { + return lo.Filter(scaleRun.Spec.Webhooks, func(w rolloutv1alpha1.RolloutWebhook, _ int) bool { + return slices.Contains(w.HookTypes, hookType) + }) +} + +func (c *ExecutorContext) GetWebhooksAndLatestStatusBy(hookType rolloutv1alpha1.HookType) ([]rolloutv1alpha1.RolloutWebhook, *rolloutv1alpha1.RolloutWebhookStatus) { + c.Initialize() + + run := c.ScaleRun + newStatus := c.NewStatus + webhooks := filterWebhooks(hookType, run) + if len(webhooks) == 0 { + // no webhooks + return nil, nil + } + var webhookStatuses []rolloutv1alpha1.RolloutWebhookStatus + index := newStatus.Batches.CurrentBatchIndex + webhookStatuses = newStatus.Batches.Records[index].Webhooks + var status *rolloutv1alpha1.RolloutWebhookStatus + if len(webhookStatuses) > 0 { + latestStatus := &webhookStatuses[len(webhookStatuses)-1] + if latestStatus.HookType == hookType { + status = latestStatus + } + } + return webhooks, status +} + +func (c *ExecutorContext) SetWebhookStatus(status rolloutv1alpha1.RolloutWebhookStatus) { + c.Initialize() + + newStatus := c.NewStatus + index := newStatus.Batches.CurrentBatchIndex + newStatus.Batches.Records[index].Webhooks = appendWebhookStatus(newStatus.Batches.Records[index].Webhooks, status) +} + +func isFinalStepState(state rolloutv1alpha1.RolloutStepState) bool { + return state == rorexecutor.StepSucceeded +} + +func (c *ExecutorContext) GetCurrentState() (string, rolloutv1alpha1.RolloutStepState) { + return "batch", c.NewStatus.Batches.CurrentBatchState +} + +func (c *ExecutorContext) MoveToNextState(nextState rolloutv1alpha1.RolloutStepState) { + c.Initialize() + + newStatus := c.NewStatus + index := newStatus.Batches.CurrentBatchIndex + newStatus.Batches.CurrentBatchState = nextState + newStatus.Batches.Records[index].State = nextState + if nextState == rorexecutor.StepPreBatchStepHook { + newStatus.Batches.Records[index].StartTime = ptr.To(metav1.Now()) + } else if isFinalStepState(nextState) { + newStatus.Batches.Records[index].FinishTime = ptr.To(metav1.Now()) + } +} + +func (c *ExecutorContext) SkipCurrentRelease() { + c.Initialize() + + newStatus := c.NewStatus + newStatus.Batches.CurrentBatchIndex = int32(len(c.ScaleRun.Spec.Batch.Batches) - 1) + newStatus.Batches.CurrentBatchState = rorexecutor.StepSucceeded + for i := range newStatus.Batches.Records { + if newStatus.Batches.Records[i].State == rorexecutor.StepNone || + newStatus.Batches.Records[i].State == rorexecutor.StepPending { + newStatus.Batches.Records[i].State = rorexecutor.StepSucceeded + } + if newStatus.Batches.Records[i].StartTime == nil { + newStatus.Batches.Records[i].StartTime = ptr.To(metav1.Now()) + } + if newStatus.Batches.Records[i].FinishTime == nil { + newStatus.Batches.Records[i].FinishTime = ptr.To(metav1.Now()) + } + } +} + +func (c *ExecutorContext) Pause() { + c.Initialize() + c.NewStatus.Phase = rolloutv1alpha1.RolloutRunPhasePaused +} + +func (c *ExecutorContext) Fail(err error) { + c.Initialize() + //nolint:errorlint + crm, ok := err.(*rolloutv1alpha1.CodeReasonMessage) + if ok { + c.NewStatus.Error = crm + } else { + c.NewStatus.Error = &rolloutv1alpha1.CodeReasonMessage{ + Code: "Error", + Reason: "ExecutorFailed", + Message: err.Error(), + } + } +} + +func (c *ExecutorContext) MoveToNextStateIfMatch(curState, nextState rolloutv1alpha1.RolloutStepState) { + c.Initialize() + _, state := c.GetCurrentState() + if state == curState { + c.MoveToNextState(nextState) + } +} + +func appendWebhookStatus(origin []rolloutv1alpha1.RolloutWebhookStatus, input rolloutv1alpha1.RolloutWebhookStatus) []rolloutv1alpha1.RolloutWebhookStatus { + length := len(origin) + if length == 0 { + return []rolloutv1alpha1.RolloutWebhookStatus{input} + } + + if origin[length-1].HookType == input.HookType && origin[length-1].Name == input.Name { + origin[length-1] = input + } else { + origin = append(origin, input) + } + return origin +} + +func (r *ExecutorContext) makeRolloutWebhookReview(hookType rolloutv1alpha1.HookType, webhook rolloutv1alpha1.RolloutWebhook) rolloutv1alpha1.RolloutWebhookReview { + r.Initialize() + + scaleRun := r.ScaleRun + newStatus := r.NewStatus + + review := rolloutv1alpha1.RolloutWebhookReview{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: scaleRun.Namespace, + Name: webhook.Name, + }, + Spec: rolloutv1alpha1.RolloutWebhookReviewSpec{ + RolloutID: scaleRun.Name, + HookType: hookType, + Properties: webhook.Properties, + TargetType: scaleRun.Spec.TargetType, + }, + } + + scaleTargets := scaleRun.Spec.Batch.Batches[newStatus.Batches.CurrentBatchIndex].Targets + targets := make([]rolloutv1alpha1.RolloutRunStepTarget, len(scaleTargets)) + for _, target := range scaleTargets { + targets = append(targets, rolloutv1alpha1.RolloutRunStepTarget{ + CrossClusterObjectNameReference: rolloutv1alpha1.CrossClusterObjectNameReference{ + Cluster: target.Cluster, + Name: target.Name, + }, + Replicas: intstr.FromInt(int(target.Replicas)), + }) + } + + review.Spec.Batch = &rolloutv1alpha1.RolloutWebhookReviewBatch{ + BatchIndex: newStatus.Batches.CurrentBatchIndex, + Targets: targets, + Properties: scaleRun.Spec.Batch.Batches[newStatus.Batches.CurrentBatchIndex].Properties, + } + + return review +} + +func (e *ExecutorContext) WithLogger(logger logr.Logger) logr.Logger { + l := logger.WithValues( + "namespace", e.ScaleRun.Namespace, + "scaleRun", e.ScaleRun.Name, + ) + + e.Context = logr.NewContext(e.Context, l) + return l +} + +func (e *ExecutorContext) GetLogger() logr.Logger { + return logr.FromContextOrDiscard(e.Context) +} + +func (e *ExecutorContext) GetBatchLogger() logr.Logger { + e.Initialize() + l := e.GetLogger().WithValues("step", "batch") + if e.NewStatus != nil && e.NewStatus.Batches != nil { + l = l.WithValues("batchIndex", e.NewStatus.Batches.CurrentBatchIndex) + } + return l +} + +func (e *ExecutorContext) GetCanaryLogger() logr.Logger { + return e.GetLogger().WithValues("step", "canary") +} diff --git a/pkg/controllers/scalerun/executor/default.go b/pkg/controllers/scalerun/executor/default.go new file mode 100644 index 0000000..e010c89 --- /dev/null +++ b/pkg/controllers/scalerun/executor/default.go @@ -0,0 +1,148 @@ +package executor + +import ( + "time" + + "github.com/go-logr/logr" + rolloutapis "kusionstack.io/kube-api/rollout" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + ctrl "sigs.k8s.io/controller-runtime" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/utils" +) + +type Executor struct { + logger logr.Logger + batch *batchExecutor +} + +func NewDefaultExecutor(logger logr.Logger) *Executor { + webhookExec := newWebhookExecutor(time.Second) + batchExec := newBatchExecutor(webhookExec) + e := &Executor{ + logger: logger, + batch: batchExec, + } + return e +} + +// Do execute the lifecycle for rollout run, and will return new status +func (r *Executor) Do(ctx *ExecutorContext) (bool, ctrl.Result, error) { + // init NewStatus + ctx.Initialize() + + logger := ctx.WithLogger(r.logger) + + newStatus := ctx.NewStatus + scaleRun := ctx.ScaleRun + prePhase := newStatus.Phase + + defer func() { + if prePhase != newStatus.Phase { + logger.Info("scaleRun status phase transition", "phase.from", prePhase, "phase.to", newStatus.Phase) + } + }() + + // if command exist, do command + if _, exist := utils.GetMapValue(scaleRun.Annotations, rolloutapis.AnnoManualCommandKey); exist { + return false, r.doCommand(ctx), nil + } + + return r.lifecycle(ctx) +} + +// lifecycle +func (r *Executor) lifecycle(executorContext *ExecutorContext) (done bool, result ctrl.Result, err error) { + newStatus := executorContext.NewStatus + result = ctrl.Result{Requeue: true} + scaleRun := executorContext.ScaleRun + + // treat deletion as canceling and requeue + if !scaleRun.DeletionTimestamp.IsZero() && newStatus.Phase != rolloutv1alpha1.RolloutRunPhaseCanceling { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseCanceling + // leave a phase transition log + return false, result, nil + } + + switch newStatus.Phase { + case rolloutv1alpha1.RolloutRunPhaseInitial: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePreRollout + case rolloutv1alpha1.RolloutRunPhasePausing: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePaused + case rolloutv1alpha1.RolloutRunPhaseCanceling: + var canceled bool + canceled, result, err = r.doCanceling(executorContext) + if canceled { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseCanceled + } + case rolloutv1alpha1.RolloutRunPhasePreRollout: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + case rolloutv1alpha1.RolloutRunPhaseProgressing: + var processingDone bool + processingDone, result, err = r.doProcessing(executorContext) + if processingDone { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePostRollout + } + case rolloutv1alpha1.RolloutRunPhasePostRollout: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseSucceeded + case rolloutv1alpha1.RolloutRunPhasePaused: + // rolloutRun is paused, do not requeue + result.Requeue = false + case rolloutv1alpha1.RolloutRunPhaseSucceeded, rolloutv1alpha1.RolloutRunPhaseCanceled: + done = true + result.Requeue = false + } + return done, result, err +} + +// doProcessing process canary and batch one-by-one +func (r *Executor) doProcessing(ctx *ExecutorContext) (bool, ctrl.Result, error) { + scaleRun := ctx.ScaleRun + newStatus := ctx.NewStatus + + logger := ctx.GetLogger() + + if newStatus.Error != nil { + // if error occurred, do nothing + return false, ctrl.Result{Requeue: true}, nil + } + + if scaleRun.Spec.Batch != nil && len(scaleRun.Spec.Batch.Batches) > 0 { + // init BatchStatus + if len(newStatus.Batches.CurrentBatchState) == 0 { + newStatus.Batches.CurrentBatchState = rorexecutor.StepNone + } + preCurrentBatchIndex := newStatus.Batches.CurrentBatchIndex + preCurrentBatchState := newStatus.Batches.CurrentBatchState + defer func() { + if preCurrentBatchIndex != newStatus.Batches.CurrentBatchIndex || + preCurrentBatchState != newStatus.Batches.CurrentBatchState { + logger.Info("scalerun batch state trasition", + "current.index", preCurrentBatchIndex, + "current.state", preCurrentBatchState, + "next.index", newStatus.Batches.CurrentBatchIndex, + "next.state", newStatus.Batches.CurrentBatchState, + ) + } + }() + return r.batch.Do(ctx) + } + + return true, ctrl.Result{Requeue: true}, nil +} + +func (r *Executor) doCanceling(ctx *ExecutorContext) (bool, ctrl.Result, error) { + scaleRun := ctx.ScaleRun + newStatus := ctx.NewStatus + + if scaleRun.Spec.Batch != nil && len(scaleRun.Spec.Batch.Batches) > 0 { + // init BatchStatus + if len(newStatus.Batches.CurrentBatchState) == 0 { + newStatus.Batches.CurrentBatchState = rorexecutor.StepNone + } + return r.batch.Cancel(ctx) + } + + return true, ctrl.Result{Requeue: true}, nil +} diff --git a/pkg/controllers/scalerun/executor/do_command.go b/pkg/controllers/scalerun/executor/do_command.go new file mode 100644 index 0000000..793d17a --- /dev/null +++ b/pkg/controllers/scalerun/executor/do_command.go @@ -0,0 +1,62 @@ +package executor + +import ( + rolloutapis "kusionstack.io/kube-api/rollout" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + ctrl "sigs.k8s.io/controller-runtime" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" +) + +// doCommand +func (r *Executor) doCommand(ctx *ExecutorContext) ctrl.Result { + scaleRun := ctx.ScaleRun + cmd := scaleRun.Annotations[rolloutapis.AnnoManualCommandKey] + logger := ctx.WithLogger(r.logger) + logger.Info("processing manual command", "command", cmd) + + newStatus := ctx.NewStatus + newBatchStatus := ctx.NewStatus.Batches + + batchError := newStatus.Error + currentBatchIndex := newBatchStatus.CurrentBatchIndex + switch cmd { + case rolloutapis.AnnoManualCommandPause: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePausing + case rolloutapis.AnnoManualCommandResume, rolloutapis.AnnoManualCommandContinue: // nolint + if newStatus.Phase == rolloutv1alpha1.RolloutRunPhasePaused { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + } + case rolloutapis.AnnoManualCommandRetry: + if batchError != nil { + newStatus.Error = nil + } + case rolloutapis.AnnoManualCommandSkip: + if batchError != nil { + newStatus.Error = nil + + if int(currentBatchIndex) < (len(scaleRun.Spec.Batch.Batches) - 1) { + currentBatchIndex++ + newBatchStatus.CurrentBatchIndex = currentBatchIndex + newBatchStatus.CurrentBatchState = rorexecutor.StepNone + } else { + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePostRollout + } + } + case rolloutapis.AnnoManualCommandCancel: + newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseCanceling + case rolloutapis.AnnoManualCommandForceSkipCurrentBatch: + if batchError != nil { + newStatus.Error = nil + } + if int(currentBatchIndex) < (len(scaleRun.Spec.Batch.Batches) - 1) { + currentBatchIndex++ + newBatchStatus.CurrentBatchIndex = currentBatchIndex + newBatchStatus.CurrentBatchState = rorexecutor.StepNone + } else { + newBatchStatus.CurrentBatchState = rorexecutor.StepPostBatchStepHook + } + } + + return ctrl.Result{Requeue: true} +} diff --git a/pkg/controllers/scalerun/executor/do_hook.go b/pkg/controllers/scalerun/executor/do_hook.go new file mode 100644 index 0000000..0559672 --- /dev/null +++ b/pkg/controllers/scalerun/executor/do_hook.go @@ -0,0 +1,184 @@ +package executor + +import ( + "time" + + "github.com/samber/lo" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + + "kusionstack.io/rollout/pkg/controllers/rolloutrun/webhook" + "kusionstack.io/rollout/pkg/utils" +) + +const ( + ReasonWebhookNotExist = "WebhookNotExist" + ReasonWebhookExecuteError = "WebhookExecuteError" + ReasonWebhookFailurePolicyInvalid = "WebhookFailurePolicyInvalid" + ReasonWebhookReviewStatusCodeUnknown = "WebhookReviewStatusCodeUnknown" + ReasonWebhookFailureThresholdExceeded = "WebhookFailureThresholdExceeded" +) + +type webhookExecutor interface { + Do(ctx *ExecutorContext, hookType rolloutv1alpha1.HookType) (bool, time.Duration, error) + Cancel(ctx *ExecutorContext) +} + +type webhookExecutorImpl struct { + webhookManager webhook.Manager + webhookInitTime time.Duration +} + +func newWebhookExecutor(webhookInitTime time.Duration) webhookExecutor { + return &webhookExecutorImpl{ + webhookManager: webhook.NewManager(), + webhookInitTime: webhookInitTime, + } +} + +func (r *webhookExecutorImpl) Do(ctx *ExecutorContext, hookType rolloutv1alpha1.HookType) (bool, time.Duration, error) { + curWebhook, nextWebhook := r.findCurrentAndNextWebhook(ctx, hookType) + if curWebhook == nil { + return true, retryImmediately, nil + } + + logger := ctx.GetLogger() + logger.Info("processing webhook", "hookType", hookType, "webhook", curWebhook.Name) + + hookResult, _, err := r.startOrGetWebhookWorker(ctx, hookType, *curWebhook.RolloutWebhook, curWebhook.status) + if err != nil { + logger.Error(err, "failed to get webhook result") + return false, retryImmediately, err + } + + logger.V(2).Info("get webhook result", "hookType", hookType, "webhook", curWebhook.Name, "result", hookResult) + + // shorten long message + hookResult.Message = utils.Abbreviate(hookResult.Message, 1024) + + ctx.SetWebhookStatus(rolloutv1alpha1.RolloutWebhookStatus(*hookResult)) + + if hookResult.State == rolloutv1alpha1.WebhookOnHold && + hookResult.Code == rolloutv1alpha1.WebhookReviewCodeError && + ctx.NewStatus.Error == nil { + // set error if possible + ctx.NewStatus.Error = &hookResult.CodeReasonMessage + } + if hookResult.State != rolloutv1alpha1.WebhookCompleted { + // the webhook sill running, requeue after defaultRequeueAfter duration + return false, retryDefault, nil + } + + if nextWebhook != nil { + // add empty status to start next webhook + ctx.SetWebhookStatus(rolloutv1alpha1.RolloutWebhookStatus{ + HookType: hookType, + Name: nextWebhook.Name, + }) + return false, retryImmediately, nil + } + + // NOTE: + // The code up to this point indicates that the webhooks have all been completed, and we can safely clean up the results. + // However, there is still one scenario where, if the current webhook status is not updated successfully, the executor will come back + // and execute the last webhook again. Because the webhook is idempotent, it is safe to re-execute it. + logger.Info("clean up final webhook", "hookType", hookType) + r.webhookManager.Stop(ctx.ScaleRun.UID) + + return true, retryImmediately, nil +} + +func (r *webhookExecutorImpl) Cancel(ctx *ExecutorContext) { + logger := ctx.GetLogger() + logger.Info("cancel webhook", "scaleRun", ctx.ScaleRun.Name) + r.webhookManager.Stop(ctx.ScaleRun.UID) +} + +type webhookWithStatus struct { + *rolloutv1alpha1.RolloutWebhook + status *rolloutv1alpha1.RolloutWebhookStatus +} + +func (r *webhookExecutorImpl) findCurrentAndNextWebhook(executorContext *ExecutorContext, hookType rolloutv1alpha1.HookType) (*webhookWithStatus, *rolloutv1alpha1.RolloutWebhook) { + webhooks, latestStatus := executorContext.GetWebhooksAndLatestStatusBy(hookType) + if len(webhooks) == 0 { + // no webhooks + return nil, nil + } + + index := 0 + var currentWebhookStatus *rolloutv1alpha1.RolloutWebhookStatus + + if latestStatus != nil { + _, tempI, found := lo.FindIndexOf(webhooks, func(rw rolloutv1alpha1.RolloutWebhook) bool { + return rw.Name == latestStatus.Name + }) + if found { + // last status found in webhooks, it is current webhook + currentWebhookStatus = latestStatus + index = tempI + } + } + + current, next := getCurrentAndNext(webhooks, index) + if current == nil { + return nil, nil + } + + currentWebhook := &webhookWithStatus{ + RolloutWebhook: current, + status: currentWebhookStatus, + } + + return currentWebhook, next +} + +func (r *webhookExecutorImpl) startOrGetWebhookWorker(ctx *ExecutorContext, hookType rolloutv1alpha1.HookType, webhookCfg rolloutv1alpha1.RolloutWebhook, lastStatus *rolloutv1alpha1.RolloutWebhookStatus) (*webhook.Result, bool, error) { + run := ctx.ScaleRun + key := run.UID + logger := ctx.GetLogger() + worker, ok := r.webhookManager.Get(key) + if ok { + // webhook already started + curResult := worker.Result() + if curResult.Name == webhookCfg.Name && curResult.HookType == hookType { + if lastStatus != nil && lastStatus.State == rolloutv1alpha1.WebhookOnHold { + // lastStatus is onHold, that means it should be retry + worker.Retry() + // return a temporary result + curResult.State = rolloutv1alpha1.WebhookRunning + } + return &curResult, false, nil + } + + // webhook name or type not match, stop it and start a new one + logger.Info("stop the old webhook worker", "webhook", curResult.Name, "type", curResult.HookType) + worker.Stop() + } + + logger.Info("start a new webhook worker and wait for the result for a brief period.", "webhook", webhookCfg.Name, "type", hookType) + + review := ctx.makeRolloutWebhookReview(hookType, webhookCfg) + worker, err := r.webhookManager.Start(key, webhookCfg, review) + if err != nil { + return nil, false, err + } + + // Delay briefly and attempt to retrieve the webhook result immediately. + time.Sleep(r.webhookInitTime) + + return ptr.To(worker.Result()), true, nil +} + +func getCurrentAndNext[T any](input []T, index int) (*T, *T) { + length := len(input) + var current, next *T + + if index < length { + current = &input[index] + } + if index < length-1 { + next = &input[index+1] + } + return current, next +} diff --git a/pkg/controllers/scalerun/executor/step_lifecycle.go b/pkg/controllers/scalerun/executor/step_lifecycle.go new file mode 100644 index 0000000..e116cb5 --- /dev/null +++ b/pkg/controllers/scalerun/executor/step_lifecycle.go @@ -0,0 +1,141 @@ +/** + * 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 + * + * http://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 executor + +import ( + "errors" + "fmt" + "time" + + "github.com/samber/lo" + corev1 "k8s.io/api/core/v1" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + ctrl "sigs.k8s.io/controller-runtime" + + "kusionstack.io/rollout/pkg/utils" +) + +const ( + retryStop = time.Duration(-1) + retryImmediately = time.Duration(0) + retryDefault = 5 * time.Second +) + +func newUnknownStepStateError(state rolloutv1alpha1.RolloutStepState) *rolloutv1alpha1.CodeReasonMessage { + return &rolloutv1alpha1.CodeReasonMessage{ + Code: "Error", + Reason: "UnknownStepState", + Message: fmt.Sprintf("Unknown step state %s in state machine", state), + } +} + +type stateProcess func(*ExecutorContext) (done bool, retry time.Duration, err error) + +func skipStep(*ExecutorContext) (bool, time.Duration, error) { + return true, retryImmediately, nil +} + +type stepLifecycle struct { + current rolloutv1alpha1.RolloutStepState + next rolloutv1alpha1.RolloutStepState + do stateProcess + cancel stateProcess +} + +type stepStateEngine struct { + lifecycle []stepLifecycle +} + +func newStepStateEngine() *stepStateEngine { + return &stepStateEngine{ + lifecycle: make([]stepLifecycle, 0), + } +} + +func (e *stepStateEngine) add(state, nextState rolloutv1alpha1.RolloutStepState, do, cancel stateProcess) { + if do == nil { + do = skipStep + } + if cancel == nil { + do = skipStep + } + e.lifecycle = append(e.lifecycle, stepLifecycle{ + current: state, + next: nextState, + do: do, + cancel: cancel, + }) +} + +func (e *stepStateEngine) cancel(ctx *ExecutorContext, currentState rolloutv1alpha1.RolloutStepState) (done bool, result ctrl.Result, err error) { + return e.process(ctx, currentState, true) +} + +func (e *stepStateEngine) do(ctx *ExecutorContext, currentState rolloutv1alpha1.RolloutStepState) (done bool, result ctrl.Result, err error) { + return e.process(ctx, currentState, false) +} + +func (e *stepStateEngine) process(ctx *ExecutorContext, currentState rolloutv1alpha1.RolloutStepState, cancel bool) (done bool, result ctrl.Result, err error) { + lifecycle, found := lo.Find(e.lifecycle, func(step stepLifecycle) bool { + return step.current == currentState + }) + + if !found { + ctx.Fail(newUnknownStepStateError(currentState)) + return false, ctrl.Result{}, nil + } + + fn := lifecycle.do + if cancel { + fn = lifecycle.cancel + } + stateDone, retry, err := fn(ctx) + if err != nil { + ctx.Recorder.Eventf(ctx.ScaleRun, corev1.EventTypeWarning, "FailedRunStep", "step failed, currentState %s, err: %v", currentState, err) + 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) + } + return false, ctrl.Result{}, err + } + + if stateDone { + if cancel { + // if cancel is true, we stop at this state + ctx.Recorder.Eventf(ctx.ScaleRun, corev1.EventTypeNormal, "StepCanceled", "step canceled, currentState %s", currentState) + done = true + } else if len(lifecycle.next) == 0 { + // final state + done = true + } else { + done = false + ctx.MoveToNextState(lifecycle.next) + } + } + + switch retry { + case retryStop: + result = ctrl.Result{} + case retryImmediately: + result = ctrl.Result{Requeue: true} + default: + result = ctrl.Result{RequeueAfter: retry} + } + + return done, result, nil +} diff --git a/pkg/controllers/scalerun/initializer.go b/pkg/controllers/scalerun/initializer.go new file mode 100644 index 0000000..850f828 --- /dev/null +++ b/pkg/controllers/scalerun/initializer.go @@ -0,0 +1,40 @@ +// Copyright 2023 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 +// +// http://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 scalerun + +import ( + "kusionstack.io/kube-utils/controller/initializer" + "sigs.k8s.io/controller-runtime/pkg/manager" + + "kusionstack.io/rollout/pkg/controllers/registry" +) + +func InitFunc(mgr manager.Manager) (bool, error) { + return initFunc(mgr, registry.Workloads) +} + +func InitFuncWith(r registry.WorkloadRegistry) initializer.InitFunc { + return func(m manager.Manager) (enabled bool, err error) { + return initFunc(m, r) + } +} + +func initFunc(mgr manager.Manager, r registry.WorkloadRegistry) (bool, error) { + err := NewReconciler(mgr, r).SetupWithManager(mgr) + if err != nil { + return false, err + } + return true, nil +} diff --git a/pkg/controllers/scalerun/scalerun_controller.go b/pkg/controllers/scalerun/scalerun_controller.go new file mode 100644 index 0000000..dfba0e9 --- /dev/null +++ b/pkg/controllers/scalerun/scalerun_controller.go @@ -0,0 +1,302 @@ +// Copyright 2023 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 +// +// http://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 scalerun + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/equality" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "kusionstack.io/kube-api/rollout" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + "kusionstack.io/kube-api/rollout/v1alpha1/condition" + kubeutilclient "kusionstack.io/kube-utils/client" + "kusionstack.io/kube-utils/controller/mixin" + "kusionstack.io/kube-utils/multicluster/clusterinfo" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "kusionstack.io/rollout/pkg/controllers/registry" + "kusionstack.io/rollout/pkg/controllers/scalerun/executor" + "kusionstack.io/rollout/pkg/features/rolloutclasspredicate" + "kusionstack.io/rollout/pkg/utils" + "kusionstack.io/rollout/pkg/utils/expectations" + "kusionstack.io/rollout/pkg/workload" +) + +const ( + ControllerName = "scalerun" +) + +// ScaleRunReconciler reconciles a Rollout object +type ScaleRunReconciler struct { + *mixin.ReconcilerMixin + + workloadRegistry registry.WorkloadRegistry + + rvExpectation expectations.ResourceVersionExpectationInterface + + executor *executor.Executor +} + +func NewReconciler(mgr manager.Manager, workloadRegistry registry.WorkloadRegistry) *ScaleRunReconciler { + r := &ScaleRunReconciler{ + ReconcilerMixin: mixin.NewReconcilerMixin(ControllerName, mgr), + workloadRegistry: workloadRegistry, + rvExpectation: expectations.NewResourceVersionExpectation(), + } + + r.executor = executor.NewDefaultExecutor(r.Logger) + return r +} + +// SetupWithManager sets up the controller with the Manager. +func (r *ScaleRunReconciler) SetupWithManager(mgr ctrl.Manager) error { + if r.workloadRegistry == nil { + return fmt.Errorf("workload manager must be set") + } + + b := ctrl.NewControllerManagedBy(mgr). + For(&rolloutv1alpha1.ScaleRun{}, + builder.WithPredicates( + predicate.ResourceVersionChangedPredicate{}, + // NOTE: This controller only watches one kind of resource, + // so we can use predicate to filter events by rollout-class + rolloutclasspredicate.RolloutClassMatchesPredicate, + )) + + _, err := b.Build(r) + return err +} + +//+kubebuilder:rbac:groups=rollout.kusionstack.io,resources=scaleruns,verbs=get;list;watch;create;update;patch;delete +//+kubebuilder:rbac:groups=rollout.kusionstack.io,resources=scaleruns/status,verbs=get;update;patch +//+kubebuilder:rbac:groups=rollout.kusionstack.io,resources=scaleruns/finalizers,verbs=update + +// Reconcile is part of the main kubernetes reconciliation loop which aims to +// move the current state of the cluster closer to the desired state. +func (r *ScaleRunReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + logger := r.Logger.WithValues("scalerun", req.String()) + logger.V(4).Info("started reconciling scalerun") + defer logger.V(4).Info("finished reconciling scalerun") + + obj := &rolloutv1alpha1.ScaleRun{} + err := r.Client.Get(clusterinfo.WithCluster(ctx, clusterinfo.Fed), req.NamespacedName, obj) + if err != nil { + if errors.IsNotFound(err) { + r.rvExpectation.DeleteExpectations(req.String()) + return reconcile.Result{}, nil + } + return reconcile.Result{}, err + } + + if !r.satisfiedExpectations(obj) { + return reconcile.Result{}, nil + } + + if err = r.handleFinalizer(ctx, obj); err != nil { + logger.Error(err, "scalerun handleFinalizer failed") + return ctrl.Result{}, nil + } + + if obj.IsCompleted() { + // scaleRun is completed, skip syncing + return reconcile.Result{}, nil + } + + newStatus := obj.Status.DeepCopy() + + accessor, workloads, err := r.findWorkloadsCrossCluster(ctx, obj) + if err != nil { + return reconcile.Result{}, err + } + + var result ctrl.Result + result, err = r.syncScaleRun(ctx, obj, newStatus, accessor, workloads) + + if tempErr := r.cleanupAnnotation(ctx, obj); tempErr != nil { + logger.Error(tempErr, "failed to clean up scalerun annotation") + } + + updateStatus := r.updateStatusOnly(ctx, obj, newStatus) + if updateStatus != nil { + logger.Error(updateStatus, "failed to update scalerun status") + return reconcile.Result{}, updateStatus + } + + if err != nil { + return reconcile.Result{}, err + } + + return result, nil +} + +func (r *ScaleRunReconciler) satisfiedExpectations(instance *rolloutv1alpha1.ScaleRun) bool { + key := utils.ObjectKeyString(instance) + logger := r.Logger.WithValues("scalerun", key) + + if !r.rvExpectation.SatisfiedExpectations(key, instance.ResourceVersion) { + logger.Info("scalerun does not statisfy resourceVersion expectation, skip reconciling") + return false + } + + return true +} + +func (r *ScaleRunReconciler) handleFinalizer(ctx context.Context, scaleRun *rolloutv1alpha1.ScaleRun) error { + if scaleRun.IsCompleted() { + // remove finalizer when scaleRun is completed + if err := kubeutilclient.RemoveFinalizerAndUpdate(ctx, r.Client, scaleRun, rollout.FinalizerRolloutProtection); err != nil { + return err + } + } else if scaleRun.DeletionTimestamp.IsZero() { + // add finalizer when scaleRun is not completed and not deleted + if err := kubeutilclient.AddFinalizerAndUpdate(ctx, r.Client, scaleRun, rollout.FinalizerRolloutProtection); err != nil { + return err + } + } + + return nil +} + +func (r *ScaleRunReconciler) cleanupAnnotation(ctx context.Context, obj *rolloutv1alpha1.ScaleRun) error { + // delete manual command annotations from rollout + _, err := utils.UpdateOnConflict(clusterinfo.WithCluster(ctx, clusterinfo.Fed), r.Client, r.Client, obj, func() error { + delete(obj.Annotations, rollout.AnnoManualCommandKey) + return nil + }) + if err != nil { + return err + } + key := utils.ObjectKeyString(obj) + r.rvExpectation.ExpectUpdate(key, obj.ResourceVersion) // nolint + return nil +} + +func (r *ScaleRunReconciler) syncScaleRun( + ctx context.Context, + obj *rolloutv1alpha1.ScaleRun, + newStatus *rolloutv1alpha1.ScaleRunStatus, + accesor workload.Accessor, + workloads *workload.Set, +) (ctrl.Result, error) { + var ( + done bool + result ctrl.Result + err error + ) + + executorCtx := &executor.ExecutorContext{ + Context: ctx, + Client: r.Client, + Recorder: r.Recorder, + Accessor: accesor, + ScaleRun: obj, + NewStatus: newStatus, + Workloads: workloads, + } + if done, result, err = r.executor.Do(executorCtx); err != nil { + return ctrl.Result{}, err + } + if done || newStatus.Phase == rolloutv1alpha1.RolloutRunPhaseSucceeded { + newCondition := condition.NewCondition( + rolloutv1alpha1.RolloutConditionProgressing, + metav1.ConditionFalse, + rolloutv1alpha1.RolloutReasonProgressingCompleted, + "scaleRun is completed", + ) + newStatus.Conditions = condition.SetCondition(newStatus.Conditions, *newCondition) + } else if newStatus.Phase == rolloutv1alpha1.RolloutRunPhaseCanceled { + newCondition := condition.NewCondition( + rolloutv1alpha1.RolloutConditionProgressing, + metav1.ConditionFalse, + rolloutv1alpha1.RolloutReasonProgressingCanceled, + "scaleRun is canceled", + ) + newStatus.Conditions = condition.SetCondition(newStatus.Conditions, *newCondition) + } else if newStatus.Error != nil { + newCondition := condition.NewCondition( + rolloutv1alpha1.RolloutConditionProgressing, + metav1.ConditionFalse, + rolloutv1alpha1.RolloutReasonProgressingError, + "scaleRun stop rolling since error exist", + ) + newStatus.Conditions = condition.SetCondition(newStatus.Conditions, *newCondition) + } else { + newCondition := condition.NewCondition( + rolloutv1alpha1.RolloutConditionProgressing, + metav1.ConditionTrue, + rolloutv1alpha1.RolloutReasonProgressingRunning, + "scaleRun is running", + ) + newStatus.Conditions = condition.SetCondition(newStatus.Conditions, *newCondition) + } + return result, nil +} + +func (r *ScaleRunReconciler) findWorkloadsCrossCluster(ctx context.Context, obj *rolloutv1alpha1.ScaleRun) (workload.Accessor, *workload.Set, error) { + all := make([]rolloutv1alpha1.CrossClusterObjectNameReference, 0) + + for _, b := range obj.Spec.Batch.Batches { + for _, t := range b.Targets { + all = append(all, t.CrossClusterObjectNameReference) + } + } + match := rolloutv1alpha1.ResourceMatch{ + Names: all, + } + + gvk := schema.FromAPIVersionAndKind(obj.Spec.TargetType.APIVersion, obj.Spec.TargetType.Kind) + accessor, err := r.workloadRegistry.Get(gvk) + if err != nil { + return nil, nil, err + } + + list, _, err := workload.List(ctx, r.Client, accessor, obj.Namespace, match) + if err != nil { + return nil, nil, err + } + return accessor, workload.NewSet(list...), nil +} + +func (r *ScaleRunReconciler) updateStatusOnly(ctx context.Context, obj *rolloutv1alpha1.ScaleRun, newStatus *rolloutv1alpha1.ScaleRunStatus) error { + if equality.Semantic.DeepEqual(obj.Status, *newStatus) { + // no change + return nil + } + key := utils.ObjectKeyString(obj) + now := metav1.Now() + newStatus.LastUpdateTime = &now + _, err := utils.UpdateOnConflict(clusterinfo.WithCluster(ctx, clusterinfo.Fed), r.Client, r.Client.Status(), obj, func() error { + obj.Status = *newStatus + obj.Status.ObservedGeneration = obj.Generation + return nil + }) + if err != nil { + r.Recorder.Eventf(obj, corev1.EventTypeWarning, "FailedUpdateStatus", "failed to update scaleRun %q status: %v", key, err) + r.Logger.Error(err, "failed to update scaleRun status", "scaleRun", key) + return err + } + + r.rvExpectation.ExpectUpdate(key, obj.ResourceVersion) // nolint + return nil +} diff --git a/pkg/utils/error.go b/pkg/utils/error.go new file mode 100644 index 0000000..9ca58db --- /dev/null +++ b/pkg/utils/error.go @@ -0,0 +1,32 @@ +package utils + +import "errors" + +// 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) +} diff --git a/pkg/utils/patch.go b/pkg/utils/patch.go new file mode 100644 index 0000000..d92869e --- /dev/null +++ b/pkg/utils/patch.go @@ -0,0 +1,80 @@ +/** + * Copyright 2023 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 + * + * http://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 utils + +import ( + "context" + "fmt" + + "k8s.io/apimachinery/pkg/api/equality" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" +) + +type PatchWriter interface { + // Patch updates the fields corresponding to the status subresource for the + // given obj. obj must be a struct pointer so that obj can be updated + // with the content returned by the Server. + Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error +} + +// PatchOnConflict attempts to update a resource while avoiding conflicts that may arise from concurrent modifications. +// It utilizes the mutateFn function to apply changes to the original obj and then attempts an update using the writer, +// which can be either client.Writer or client.StatusWriter. +// In case of an update failure due to a conflict, UpdateOnConflict will retrieve the latest version of the object using +// the reader and attempt the update again. +// The retry mechanism adheres to the retry.DefaultBackoff policy. +func PatchOnConflict(ctx context.Context, reader client.Reader, writer PatchWriter, obj client.Object, mutateFn controllerutil.MutateFn) (patched bool, err error) { + key := client.ObjectKeyFromObject(obj) + first := true + + err = retry.RetryOnConflict(retry.DefaultBackoff, func() error { + if !first { + // refresh object + if innerErr := reader.Get(ctx, key, obj); innerErr != nil { + return innerErr + } + } else { + first = false + } + + var existing client.Object + var ok bool + if existing, ok = obj.DeepCopyObject().(client.Object); !ok { + return fmt.Errorf("object %s does not implement client.Object", key) + } + if innerErr := mutate(mutateFn, key, obj); innerErr != nil { + return innerErr + } + + if equality.Semantic.DeepEqual(existing, obj) { + // nothing changed, skip update + return nil + } + + patch := client.MergeFrom(existing) + if innerErr := writer.Patch(ctx, obj, patch); innerErr != nil { + return innerErr + } + + patched = true + return nil + }) + + return patched, err +} diff --git a/pkg/workload/info.go b/pkg/workload/info.go index df5f939..a92b1ef 100644 --- a/pkg/workload/info.go +++ b/pkg/workload/info.go @@ -54,6 +54,10 @@ type InfoStatus struct { UpdatedRevision string // Replicas is the desired number of pods targeted by workload Replicas int32 + // CurrentReplicas is the current number of existed pods targeted by workload + CurrentReplicas int32 + // AvailableReplicas is the number of service available pods targeted by workload. + AvailableReplicas int32 // UpdatedReplicas is the number of pods targeted by workload that have the updated template spec. UpdatedReplicas int32 // UpdatedReadyReplicas is the number of ready pods targeted by workload that have the updated template spec. @@ -112,6 +116,16 @@ func (o *Info) APIStatus() rolloutv1alpha1.RolloutWorkloadStatus { } } +func (o *Info) ScaleWorkloadStatus() rolloutv1alpha1.ScaleWorkloadStatus { + return rolloutv1alpha1.ScaleWorkloadStatus{ + Cluster: o.ClusterName, + Name: o.Name, + Replicas: o.Status.Replicas, + CurrentReplicas: o.Status.CurrentReplicas, + AvailableReplicas: o.Status.AvailableReplicas, + } +} + func (o *Info) UpdateOnConflict(ctx context.Context, c client.Client, mutateFn func(client.Object) error) (bool, error) { ctx = clusterinfo.WithCluster(ctx, o.ClusterName) obj := o.Object diff --git a/pkg/workload/interface.go b/pkg/workload/interface.go index 5bead8b..b225bb3 100644 --- a/pkg/workload/interface.go +++ b/pkg/workload/interface.go @@ -49,10 +49,9 @@ type BatchReleaseControl interface { // CanaryReleaseControl defines the control functions for workload canary release type CanaryReleaseControl interface { + ScaleControl // CanaryPreCheck checks object before canary release. CanaryPreCheck(obj client.Object) error - // Scale scales the workload replicas. - Scale(obj client.Object, replicas int32) error // ApplyCanaryPatch applies canary to the workload. ApplyCanaryPatch(canary client.Object, podTemplatePatch *v1alpha1.MetadataPatch) error } @@ -65,3 +64,9 @@ type ReplicaObjectControl interface { // GetReplicObjects gets the pod selector of the workload GetReplicObjects(ctx context.Context, reader client.Reader, workload client.Object) ([]client.Object, error) } + +// ScaleControl defines the control functions for workload scale +type ScaleControl interface { + // Scale use replicas to update replicas of the workload. + Scale(obj client.Object, replicas int32) error +}