fix: ops always is pending if the dependent ops is pending but behind… (#10118)

(cherry picked from commit 70fac03)

Author: wangyelei

pkg/operations/ops_util_test.go

@@ -20,6 +20,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package operations
 
 import (
+	"fmt"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -350,5 +351,91 @@ var _ = Describe("OpsUtil functions", func() {
 			_, _ = GetOpsManager().Do(reqCtx, k8sClient, opsRes)
 			Eventually(testops.GetOpsRequestPhase(&testCtx, client.ObjectKeyFromObject(opsRes.OpsRequest))).Should(Equal(opsv1alpha1.OpsCreatingPhase))
 		})
+
+		It("Test swapOpsWithDependentBefore when force ops enqueued before dependent ops", func() {
+			By("init operations resources ")
+			opsRes, _, _ := initOperationsResources(compDefName, clusterName)
+			testapps.MockInstanceSetComponent(&testCtx, clusterName, defaultCompName)
+
+			By("create a force opsRequest that depends on another ops")
+			ops1 := createHorizontalScaling(clusterName, opsv1alpha1.HorizontalScaling{
+				ComponentOps: opsv1alpha1.ComponentOps{ComponentName: defaultCompName},
+				ScaleIn: &opsv1alpha1.ScaleIn{
+					ReplicaChanger: opsv1alpha1.ReplicaChanger{ReplicaChanges: pointer.Int32(1)},
+				},
+			}, false)
+			ops2 := createHorizontalScaling(clusterName, opsv1alpha1.HorizontalScaling{
+				ComponentOps: opsv1alpha1.ComponentOps{ComponentName: defaultCompName},
+				ScaleOut: &opsv1alpha1.ScaleOut{
+					ReplicaChanger: opsv1alpha1.ReplicaChanger{ReplicaChanges: pointer.Int32(1)},
+				},
+			}, false)
+			ops3 := createHorizontalScaling(clusterName, opsv1alpha1.HorizontalScaling{
+				ComponentOps: opsv1alpha1.ComponentOps{ComponentName: defaultCompName},
+				ScaleOut: &opsv1alpha1.ScaleOut{
+					ReplicaChanger: opsv1alpha1.ReplicaChanger{ReplicaChanges: pointer.Int32(1)},
+				},
+			}, false)
+			ops3.Annotations = map[string]string{constant.OpsDependentOnSuccessfulOpsAnnoKey: fmt.Sprintf("%s,%s", ops1.Name, ops2.Name)}
+			ops3.Spec.Force = true
+
+			By("manually add ops3 to cluster annotation first with InQueue=false")
+			ops3Recorder := opsv1alpha1.OpsRecorder{
+				Name:        ops3.Name,
+				Type:        opsv1alpha1.HorizontalScalingType,
+				InQueue:     false,
+				QueueBySelf: false,
+			}
+			Expect(testapps.ChangeObj(&testCtx, opsRes.Cluster, func(cluster *appsv1.Cluster) {
+				opsutil.SetOpsRequestToCluster(cluster, []opsv1alpha1.OpsRecorder{ops3Recorder})
+			})).Should(Succeed())
+
+			By("ops2 should be in queue")
+			opsRequestSlice, _ := opsutil.GetOpsRequestSliceFromCluster(opsRes.Cluster)
+			Expect(len(opsRequestSlice)).Should(Equal(1))
+			Expect(opsRequestSlice[0].Name).Should(Equal(ops3.Name))
+			Expect(opsRequestSlice[0].InQueue).Should(BeFalse())
+
+			By("now create ops1 and add it to cluster annotation after ops2")
+			ops1Recorder := opsv1alpha1.OpsRecorder{
+				Name:        ops1.Name,
+				Type:        opsv1alpha1.HorizontalScalingType,
+				InQueue:     true,
+				QueueBySelf: false,
+			}
+			ops2Recorder := opsv1alpha1.OpsRecorder{
+				Name:        ops2.Name,
+				Type:        opsv1alpha1.HorizontalScalingType,
+				InQueue:     true,
+				QueueBySelf: false,
+			}
+			Expect(testapps.ChangeObj(&testCtx, opsRes.Cluster, func(cluster *appsv1.Cluster) {
+				opsutil.SetOpsRequestToCluster(cluster, []opsv1alpha1.OpsRecorder{ops3Recorder, ops1Recorder, ops2Recorder})
+			})).Should(Succeed())
+
+			By("verify ops1 is after ops2 in the slice")
+			opsRequestSlice, _ = opsutil.GetOpsRequestSliceFromCluster(opsRes.Cluster)
+			Expect(len(opsRequestSlice)).Should(Equal(3))
+			Expect(opsRequestSlice[0].Name).Should(Equal(ops3.Name))
+			Expect(opsRequestSlice[1].Name).Should(Equal(ops1.Name))
+			Expect(opsRequestSlice[2].Name).Should(Equal(ops2.Name))
+
+			By("simulate ops3 reconciliation that triggers swap")
+			opsRes.OpsRequest = ops3
+			ops3.Status.Phase = opsv1alpha1.OpsPendingPhase
+
+			reqCtx := intctrlutil.RequestCtx{Ctx: testCtx.Ctx}
+			opsBehaviour := GetOpsManager().OpsMap[opsv1alpha1.HorizontalScalingType]
+			opsRecorder, err := enqueueOpsRequestToClusterAnnotation(reqCtx.Ctx, k8sClient, opsRes, opsBehaviour)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(opsRecorder).ShouldNot(BeNil())
+
+			By("verify ops1 and ops2 have been swapped")
+			opsRequestSlice, _ = opsutil.GetOpsRequestSliceFromCluster(opsRes.Cluster)
+			Expect(len(opsRequestSlice)).Should(Equal(3))
+			Expect(opsRequestSlice[0].Name).Should(Equal(ops1.Name))
+			Expect(opsRequestSlice[1].Name).Should(Equal(ops2.Name))
+			Expect(opsRequestSlice[2].Name).Should(Equal(ops3.Name))
+		})
 	})
 })

pkg/operations/queue_util.go

@@ -23,13 +23,15 @@ import (
 	"context"
 	"fmt"
 	"slices"
+	"strings"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	opsv1alpha1 "github.com/apecloud/kubeblocks/apis/operations/v1alpha1"
+	"github.com/apecloud/kubeblocks/pkg/constant"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	opsutil "github.com/apecloud/kubeblocks/pkg/operations/util"
 )
@@ -132,20 +134,45 @@ func enqueueOpsRequestToClusterAnnotation(ctx context.Context, cli client.Client
 		}
 		opsRequestSlice = append(opsRequestSlice, opsRecorder)
 	default:
-		if !opsRecorder.InQueue {
-			// the opsRequest is already running.
-			return &opsRecorder, nil
-		}
-		if !opsRes.OpsRequest.Spec.Force && existOtherRunningOps(opsRequestSlice, opsRecorder.Type, opsBehaviour) {
-			// if exists other running opsRequest, return.
-			return &opsRecorder, nil
+		var doSwap bool
+		opsRequestSlice, doSwap = swapOpsWithDependentBefore(opsRequestSlice, index, opsRes)
+		// check if dependent ops are after current ops in the slice, if so swap positions and update the queue in the cluster annotation.
+		if !doSwap {
+			if !opsRecorder.InQueue {
+				// the opsRequest is already running.
+				return &opsRecorder, nil
+			}
+			if !opsRes.OpsRequest.Spec.Force && existOtherRunningOps(opsRequestSlice, opsRecorder.Type, opsBehaviour) {
+				// if exists other running opsRequest, return.
+				return &opsRecorder, nil
+			}
+			// mark to handle the next opsRequest
+			opsRequestSlice[index].InQueue = false
 		}
-		// mark to handle the next opsRequest
-		opsRequestSlice[index].InQueue = false
 	}
 	return &opsRecorder, opsutil.UpdateClusterOpsAnnotations(ctx, cli, opsRes.Cluster, opsRequestSlice)
 }
 
+func swapOpsWithDependentBefore(opsRequestSlice []opsv1alpha1.OpsRecorder, currentIndex int, opsRes *OpsResource) ([]opsv1alpha1.OpsRecorder, bool) {
+	dependentOpsStr := opsRes.OpsRequest.Annotations[constant.OpsDependentOnSuccessfulOpsAnnoKey]
+	if dependentOpsStr == "" {
+		return opsRequestSlice, false
+	}
+	dependentSet := make(map[string]struct{})
+	for _, name := range strings.Split(dependentOpsStr, ",") {
+		dependentSet[name] = struct{}{}
+	}
+	var doSwap bool
+	for i := currentIndex + 1; i < len(opsRequestSlice); i++ {
+		if _, ok := dependentSet[opsRequestSlice[i].Name]; ok {
+			opsRequestSlice[currentIndex], opsRequestSlice[i] = opsRequestSlice[i], opsRequestSlice[currentIndex]
+			currentIndex = i
+			doSwap = true
+		}
+	}
+	return opsRequestSlice, doSwap
+}
+
 // existOtherRunningOps checks if exists other running opsRequest.
 func existOtherRunningOps(opsRecorderSlice []opsv1alpha1.OpsRecorder, opsType opsv1alpha1.OpsType, opsBehaviour OpsBehaviour) bool {
 	for i := range opsRecorderSlice {