diff --git a/pkg/common/utils/mysql/mysql.go b/pkg/common/utils/mysql/mysql.go index dbfda27e..aabe3ce5 100644 --- a/pkg/common/utils/mysql/mysql.go +++ b/pkg/common/utils/mysql/mysql.go @@ -186,14 +186,14 @@ func (db *DB) DropBE(nodes []*Backend) error { klog.Infoln("mysql DropBE BE node is empty") return nil } - nodesString := fmt.Sprintf(`"%s:%d"`, nodes[0].Host, nodes[0].HeartbeatPort) - for _, node := range nodes[1:] { - nodesString = nodesString + fmt.Sprintf(`,"%s:%d"`, node.Host, node.HeartbeatPort) - } - alter := fmt.Sprintf("ALTER SYSTEM DROPP BACKEND %s;", nodesString) - _, err := db.Exec(alter) - return err + for _, node := range nodes { + alter := fmt.Sprintf(`ALTER SYSTEM DROPP BACKEND "%s:%d";`, node.Host, node.HeartbeatPort) + if _, err := db.Exec(alter); err != nil { + return err + } + } + return nil } func (db *DB) DropObserver(nodes []*Frontend) error { @@ -201,12 +201,13 @@ func (db *DB) DropObserver(nodes []*Frontend) error { klog.Infoln("DropObserver observer node is empty") return nil } - var alter string for _, node := range nodes { - alter = alter + fmt.Sprintf(`ALTER SYSTEM DROP OBSERVER "%s:%d";`, node.Host, node.EditLogPort) + alter := fmt.Sprintf(`ALTER SYSTEM DROP OBSERVER "%s:%d";`, node.Host, node.EditLogPort) + if _, err := db.Exec(alter); err != nil { + return fmt.Errorf("drop observer %s:%d failed: %w", node.Host, node.EditLogPort, err) + } } - _, err := db.Exec(alter) - return err + return nil } func (db *DB) GetObservers() ([]*Frontend, error) { diff --git a/pkg/common/utils/mysql/mysql_test.go b/pkg/common/utils/mysql/mysql_test.go index 9dd86c25..cb7d8c20 100644 --- a/pkg/common/utils/mysql/mysql_test.go +++ b/pkg/common/utils/mysql/mysql_test.go @@ -20,6 +20,9 @@ package mysql import ( _ "crypto/tls" "database/sql/driver" + "errors" + "fmt" + "regexp" "strconv" "testing" @@ -34,9 +37,9 @@ func Test_ShowFrontends(t *testing.T) { } columns := []string{"Name", "Host", "EditLogPort", "HttpPort", "QueryPort", "RpcPort", "ArrowFlightSqlPort", "Role", "IsMaster", - "ClusterId", "Join", "Alive", "ReplayedJournalId", "LastStartTime", "LastHeartbeat", "IsHelper", "ErrMsg", "Version", "CurrentConnected"} + "ClusterId", "Join", "Alive", "ReplayedJournalId", "LastStartTime", "LastHeartbeat", "IsHelper", "ErrMsg", "Version", "CurrentConnected", "LiveSince", "FutureUnknownColumn"} values := []driver.Value{"fe_36d7bccc_d358_4dfd_ad4c_6e988f94f12d", "doriscluster-sample-fe-0.doriscluster-sample-fe-internal.default.svc.cluster.local", 9010, 8030, 9030, 9020, -1, "FOLLOWER", true, "1807668748", true, true, "15443", "2024-08-21 10:04:29", - "2024-08-22 07:29:55", true, "", "doris-2.1.5-rc02-d5a02e095d", "Yes"} + "2024-08-22 07:29:55", true, "", "doris-2.1.5-rc02-d5a02e095d", "Yes", "2024-08-21 10:04:29", "ignored"} mock.ExpectQuery("show frontends").WillReturnRows(sqlmock.NewRows(columns).AddRows(values)) dorisdb := sqlx.NewDb(mysql_db, "mysql") db := &DB{ @@ -48,18 +51,21 @@ func Test_ShowFrontends(t *testing.T) { t.Errorf("show frontends failed, %s", err.Error()) } if len(fts) != 1 { - t.Errorf("show frontends failed, not retun one frontend.") + t.Fatalf("show frontends failed, expected one frontend, got %d", len(fts)) + } + if fts[0].Host != "doriscluster-sample-fe-0.doriscluster-sample-fe-internal.default.svc.cluster.local" || fts[0].Role != "FOLLOWER" { + t.Errorf("show frontends failed, known fields were not mapped: %+v", fts[0]) } } func Test_ShowBackends(t *testing.T) { columns := []string{"BackendId", "Host", "HeartbeatPort", "BePort", "HttpPort", "BrpcPort", "ArrowFlightSqlPort", "LastStartTime", "LastHeartbeat", "Alive", "SystemDecommissioned", "TabletNum", "DataUsedCapacity", "TrashUsedCapacity", "AvailCapacity", "TotalCapacity", "UsedPct", "MaxDiskUsedPct", - "RemoteUsedCapacity", "Tag", "ErrMsg", "Version", "Status", "HeartbeatFailureCounter", "NodeRole"} + "RemoteUsedCapacity", "Tag", "ErrMsg", "Version", "Status", "HeartbeatFailureCounter", "NodeRole", "LiveSince", "FutureUnknownColumn"} values := []driver.Value{"10009", "doriscluster-sample-be-0.doriscluster-sample-be-internal.default.svc.cluster.local", 9050, 9060, 8040, 8060, -1, "2024-08-21 10:05:37", "2024-08-22 08:29:46", true, false, 24, "0.000", "0.000", "74.619 GB", "439.037 GB", "83.00 %", "83.00 %", "0.000", "{\"location\" : \"default\"}", "", "doris-2.1.5-rc02-d5a02e095d", "{\"lastSuccessReportTabletsTime\":\"2024-08-22 08:29:09\",\"lastStreamLoadTime\":-1,\"isQueryDisabled\":false,\"isLoadDisabled\":false}", - 0, "mix"} + 0, "mix", "2024-08-21 10:05:37", "ignored"} mysql_db, mock, err := sqlmock.New() if err != nil { t.Errorf("sqlmock new failed %s", err.Error()) @@ -76,7 +82,10 @@ func Test_ShowBackends(t *testing.T) { t.Errorf("show backends failed, %s", err.Error()) } if len(bds) != 1 { - t.Errorf("show backends failed, not return one backend.") + t.Fatalf("show backends failed, expected one backend, got %d", len(bds)) + } + if bds[0].BackendID != "10009" || bds[0].NodeRole != "mix" { + t.Errorf("show backends failed, known fields were not mapped: %+v", bds[0]) } } @@ -143,11 +152,10 @@ func Test_DecommissionBE(t *testing.T) { } func Test_DropObserver(t *testing.T) { - version := "doris-2.1.5-rc02-d5a02e095d" - startTime := "2024-08-21 10:04:29" - heartbeat := "2024-08-22 07:29:55" - values := []*Frontend{{"fe_36d7bccc_d358_4dfd_ad4c_6e988f94f12d", "doriscluster-sample-fe-0.doriscluster-sample-fe-internal.default.svc.cluster.local", 9010, 8030, 9030, 9020, -1, "FOLLOWER", true, "1807668748", true, true, "15443", &startTime, - &heartbeat, true, "", &version, "Yes"}} + values := []*Frontend{ + {Host: "doriscluster-sample-fe-4.doriscluster-sample-fe-internal.default.svc.cluster.local", EditLogPort: 9010}, + {Host: "doriscluster-sample-fe-3.doriscluster-sample-fe-internal.default.svc.cluster.local", EditLogPort: 9010}, + } tests := [][]*Frontend{ {}, @@ -158,7 +166,10 @@ func Test_DropObserver(t *testing.T) { if err != nil { t.Errorf("sqlmock new failed %s", err.Error()) } - mock.ExpectExec("ALTER SYSTEM DROP OBSERVER").WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectExec(regexp.QuoteMeta("ALTER SYSTEM DROP OBSERVER \"doriscluster-sample-fe-4.doriscluster-sample-fe-internal.default.svc.cluster.local:9010\";")). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectExec(regexp.QuoteMeta("ALTER SYSTEM DROP OBSERVER \"doriscluster-sample-fe-3.doriscluster-sample-fe-internal.default.svc.cluster.local:9010\";")). + WillReturnResult(sqlmock.NewResult(1, 1)) dorisdb := sqlx.NewDb(mysql_db, "mysql") db := &DB{ DB: dorisdb, @@ -173,6 +184,9 @@ func Test_DropObserver(t *testing.T) { } }) } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("drop observer expectations were not met: %s", err) + } } func Test_GetObservers(t *testing.T) { @@ -228,36 +242,49 @@ func Test_GetFollowers(t *testing.T) { } func Test_DropBE(t *testing.T) { - tests := [][]*Backend{ - { - { - Host: "test", - HeartbeatPort: 9050, - }, { - Host: "test1", - HeartbeatPort: 9050, - }, - }, - {}, - } - - mysql_db, mock, err := sqlmock.New() + mysqlDB, mock, err := sqlmock.New() if err != nil { t.Errorf("sqlmock new failed %s", err.Error()) } - mock.ExpectExec("ALTER SYSTEM DROPP BACKEND").WillReturnResult(sqlmock.NewResult(1, 1)) - dorisdb := sqlx.NewDb(mysql_db, "mysql") db := &DB{ - DB: dorisdb, + DB: sqlx.NewDb(mysqlDB, "mysql"), } defer db.Close() - for i, test := range tests { - t.Run("test"+strconv.Itoa(i), func(t *testing.T) { - err = db.DropBE(test) - if err != nil { - t.Errorf("test decommission failed, err=%s", err.Error()) - } - }) + nodes := []*Backend{ + {Host: "test", HeartbeatPort: 9050}, + {Host: "test1", HeartbeatPort: 9050}, + } + for _, node := range nodes { + query := regexp.QuoteMeta(fmt.Sprintf(`ALTER SYSTEM DROPP BACKEND "%s:%d";`, node.Host, node.HeartbeatPort)) + mock.ExpectExec(query).WillReturnResult(sqlmock.NewResult(1, 1)) + } + if err := db.DropBE(nodes); err != nil { + t.Fatalf("drop backends failed: %v", err) + } + if err := db.DropBE(nil); err != nil { + t.Fatalf("drop empty backends failed: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } +} + +func TestDropBEReturnsUnexpectedError(t *testing.T) { + mysqlDB, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock new failed: %v", err) + } + db := &DB{DB: sqlx.NewDb(mysqlDB, "mysql")} + defer db.Close() + + query := regexp.QuoteMeta(`ALTER SYSTEM DROPP BACKEND "test:9050";`) + mock.ExpectExec(query).WillReturnError(errors.New("access denied")) + + if err := db.DropBE([]*Backend{{Host: "test", HeartbeatPort: 9050}}); err == nil { + t.Fatal("expected unexpected DropBE error to be returned") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) } } diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go index a91a0ec3..bb3bdd32 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go @@ -251,43 +251,30 @@ func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx conte return &sc.Event{Type: sc.EventWarning, Reason: sc.CGStorageTemplateImmutable, Message: msg}, errors.New(msg) } - err := dcgs.preApplyStatefulSet(ctx, st, &est, cluster, cg) - if err != nil { - klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset preApplyStatefulSet namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error()) - return &sc.Event{Type: sc.EventWarning, Reason: sc.CGSqlExecFailed, Message: err.Error()}, err - } - - // be decimmission processing, skip apply statefulset. - if skipApplyStatefulset(cluster, cg) { - return nil, nil + // Direct-drop scale-down is owned by the graceful state machine when the image + // supports it. Decommissioning keeps the legacy pre-apply ordering. + gracefulInProgress := false + if !cluster.Spec.EnableDecommission { + gracefulInProgress = dcgs.reconcileGracefulStatefulSet(ctx, st, &est, cluster, cg) } - // Graceful two-phase restart/shutdown: check if we need to perform a graceful action. - // This must happen after preApply but before the actual StatefulSet apply. - if dcgs.RestConfig != nil { - var cgStatus *dv1.ComputeGroupStatus - for i := range cluster.Status.ComputeGroupStatuses { - if cluster.Status.ComputeGroupStatuses[i].UniqueId == cg.UniqueId { - cgStatus = &cluster.Status.ComputeGroupStatuses[i] - break - } + if !gracefulInProgress { + err := dcgs.preApplyStatefulSet(ctx, st, &est, cluster, cg) + if err != nil { + klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset preApplyStatefulSet namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error()) + return &sc.Event{Type: sc.EventWarning, Reason: sc.CGSqlExecFailed, Message: err.Error()}, err } - if cgStatus != nil { - // If a graceful action is already in progress or needs to start, - // ensure OnDelete strategy to prevent K8s from auto-deleting pods. - skipApply, gracefulErr := dcgs.gracefulRolloutReconcile(ctx, dcgs.RestConfig, st, &est, cluster, cg, cgStatus) - if gracefulErr != nil { - klog.Errorf("reconcileStatefulset gracefulRolloutReconcile failed: %v", gracefulErr) - // Continue with normal reconcile on error, don't block. - } - if skipApply { - // Graceful action is in progress. Apply StatefulSet with OnDelete strategy - // so K8s won't auto-delete pods, but still update the template. - ensureOnDeleteStrategy(st) - } + + // be decommission processing, skip apply statefulset. + if skipApplyStatefulset(cluster, cg) { + return nil, nil } } + if cluster.Spec.EnableDecommission { + dcgs.reconcileGracefulStatefulSet(ctx, st, &est, cluster, cg) + } + if st.Spec.UpdateStrategy.Type == appv1.OnDeleteStatefulSetStrategyType { dcgs.clearStatefulSetRollingUpdate(ctx, st.Namespace, st.Name) } @@ -318,6 +305,38 @@ func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx conte return nil, nil } +func (dcgs *DisaggregatedComputeGroupsController) reconcileGracefulStatefulSet( + ctx context.Context, + st, est *appv1.StatefulSet, + cluster *dv1.DorisDisaggregatedCluster, + cg *dv1.ComputeGroup, +) bool { + if dcgs.RestConfig == nil { + return false + } + var cgStatus *dv1.ComputeGroupStatus + for i := range cluster.Status.ComputeGroupStatuses { + if cluster.Status.ComputeGroupStatuses[i].UniqueId == cg.UniqueId { + cgStatus = &cluster.Status.ComputeGroupStatuses[i] + break + } + } + if cgStatus == nil { + return false + } + + skipApply, err := dcgs.gracefulRolloutReconcile(ctx, dcgs.RestConfig, st, est, cluster, cg, cgStatus) + if err != nil { + klog.Errorf("reconcileStatefulset gracefulRolloutReconcile failed: %v", err) + } + if skipApply { + // Do not fall through to the legacy DropBE workflow while the graceful + // state machine owns this StatefulSet. + ensureOnDeleteStrategy(st) + } + return skipApply +} + func volumeClaimTemplatesEqual(new, old []corev1.PersistentVolumeClaim) bool { if len(new) != len(old) { return false diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/graceful_rollout.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/graceful_rollout.go index 9e4f8a0a..48417a49 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/graceful_rollout.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/graceful_rollout.go @@ -20,6 +20,7 @@ package computegroups import ( "context" "encoding/json" + "errors" "fmt" "sort" "strings" @@ -78,6 +79,8 @@ const ( var execInPod = k8s.ExecInPod +var errBackendNotFound = errors.New("backend not found") + // gracefulRolloutReconcile is the entry point for graceful two-phase restart/shutdown. // It returns true if the caller should skip normal StatefulSet apply (because graceful action is in progress). func (dcgs *DisaggregatedComputeGroupsController) gracefulRolloutReconcile( @@ -102,6 +105,24 @@ func (dcgs *DisaggregatedComputeGroupsController) gracefulRolloutReconcile( return false, nil } + if storedAction != nil && storedAction.Type == dv1.GracefulActionScaleDown { + requestedReplicas := *st.Spec.Replicas + currentReplicas := *est.Spec.Replicas + if storedAction.DesiredReplicas == nil || *storedAction.DesiredReplicas != requestedReplicas { + klog.Infof("gracefulRolloutReconcile: updating scale-down target for cg=%s to %d", cg.UniqueId, requestedReplicas) + storedAction.DesiredReplicas = &requestedReplicas + } + if requestedReplicas >= currentReplicas && !gracefulScaleDownPodStarted(storedAction) { + klog.Infof("gracefulRolloutReconcile: cancelling scale-down for cg=%s because requested replicas=%d current replicas=%d", + cg.UniqueId, requestedReplicas, currentReplicas) + cgStatus.Phase = dv1.Reconciling + if err := dcgs.finalizeGracefulAction(ctx, st); err != nil { + return true, err + } + return false, nil + } + } + // If we have a new action and no existing action, store the action first. // For rolling updates this lets the new StatefulSet template be applied with // OnDelete before any pod is deleted. @@ -162,6 +183,8 @@ func (dcgs *DisaggregatedComputeGroupsController) gracefulRolloutReconcile( ga.LastMessage = err.Error() klog.Errorf("gracefulRolloutReconcile: state machine error for cg=%s pod=%s phase=%s: %v", cg.UniqueId, ga.CurrentPod, ga.Phase, err) + prepareGracefulStatefulSet(st, est, ga) + setGracefulAction(st, ga) return true, err } @@ -333,6 +356,11 @@ func (dcgs *DisaggregatedComputeGroupsController) handleTriggerDrain( if epoch, ok := backendProcessEpoch(backend); ok { ga.InitialBackendEpoch = epoch } + } else if ga.Type == dv1.GracefulActionScaleDown && errors.Is(backendErr, errBackendNotFound) { + klog.Infof("handleTriggerDrain: backend for scale-down pod %s is already absent, skipping drain", ga.CurrentPod) + ga.LastMessage = fmt.Sprintf("Backend for pod %s is already absent", ga.CurrentPod) + ga.Phase = dv1.GracefulPhaseDeletePod + return nil } else { klog.Warningf("handleTriggerDrain: failed to capture initial backend generation for pod %s uid=%s containerID=%s: %v", ga.CurrentPod, ga.InitialPodUID, ga.InitialContainerID, backendErr) @@ -462,12 +490,17 @@ func (dcgs *DisaggregatedComputeGroupsController) handleDeletePod( est *appv1.StatefulSet, ga *dv1.GracefulAction, ) error { + if ga.Type == dv1.GracefulActionScaleDown { + if err := dcgs.dropBackendByPodName(ctx, cluster, cgStatus, ga.CurrentPod); err != nil { + return fmt.Errorf("failed to drop backend for scale-down pod %s: %w", ga.CurrentPod, err) + } + } + pod, err := dcgs.getPod(ctx, cluster.Namespace, ga.CurrentPod) if err != nil { if apierrors.IsNotFound(err) { klog.Infof("handleDeletePod: pod %s already deleted", ga.CurrentPod) - dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) - return nil + return dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) } return err } @@ -478,8 +511,7 @@ func (dcgs *DisaggregatedComputeGroupsController) handleDeletePod( if err := dcgs.K8sclient.Delete(ctx, pod); err != nil { if apierrors.IsNotFound(err) { klog.Infof("handleDeletePod: pod %s uid=%s already deleted before delete call completed", ga.CurrentPod, string(pod.UID)) - dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) - return nil + return dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) } return fmt.Errorf("failed to delete pod %s: %w", ga.CurrentPod, err) } @@ -487,8 +519,7 @@ func (dcgs *DisaggregatedComputeGroupsController) handleDeletePod( dcgs.K8srecorder.Eventf(cluster, string(sc.EventNormal), string(sc.GracefulPodDeleted), "Deleted pod %s during graceful %s", ga.CurrentPod, ga.Type) - dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) - return nil + return dcgs.afterPodDeleted(ga, cluster, cg, cgStatus, est) } // afterPodDeleted determines the next phase after a pod is deleted. @@ -498,7 +529,7 @@ func (dcgs *DisaggregatedComputeGroupsController) afterPodDeleted( cg *dv1.ComputeGroup, cgStatus *dv1.ComputeGroupStatus, est *appv1.StatefulSet, -) { +) error { switch ga.Type { case dv1.GracefulActionRollingUpdate: // Wait for replacement pod to become ready. @@ -508,32 +539,37 @@ func (dcgs *DisaggregatedComputeGroupsController) afterPodDeleted( // For scale down, update StatefulSet replicas after pod is deleted. // This prevents StatefulSet from recreating the deleted pod. newReplicas := ga.CurrentOrdinal // replicas = current ordinal (0-indexed) - dcgs.updateStatefulSetReplicas(context.Background(), est, newReplicas) + if err := dcgs.updateStatefulSetReplicas(context.Background(), est, newReplicas); err != nil { + return err + } dcgs.advanceToNextPod(ga) case dv1.GracefulActionDelete: newReplicas := ga.CurrentOrdinal - dcgs.updateStatefulSetReplicas(context.Background(), est, newReplicas) + if err := dcgs.updateStatefulSetReplicas(context.Background(), est, newReplicas); err != nil { + return err + } dcgs.advanceToNextPod(ga) } + return nil } // updateStatefulSetReplicas patches the StatefulSet replicas to the given value. -func (dcgs *DisaggregatedComputeGroupsController) updateStatefulSetReplicas(ctx context.Context, est *appv1.StatefulSet, replicas int32) { +func (dcgs *DisaggregatedComputeGroupsController) updateStatefulSetReplicas(ctx context.Context, est *appv1.StatefulSet, replicas int32) error { var current appv1.StatefulSet if err := dcgs.K8sclient.Get(ctx, types.NamespacedName{Namespace: est.Namespace, Name: est.Name}, ¤t); err != nil { - klog.Errorf("updateStatefulSetReplicas: failed to get StatefulSet %s/%s: %v", est.Namespace, est.Name, err) - return + return fmt.Errorf("failed to get StatefulSet %s/%s: %w", est.Namespace, est.Name, err) } if *current.Spec.Replicas == replicas { - return + est.Spec.Replicas = &replicas + return nil } current.Spec.Replicas = &replicas if err := dcgs.K8sclient.Update(ctx, ¤t); err != nil { - klog.Errorf("updateStatefulSetReplicas: failed to update StatefulSet %s/%s replicas to %d: %v", - est.Namespace, est.Name, replicas, err) - } else { - klog.Infof("updateStatefulSetReplicas: updated StatefulSet %s/%s replicas to %d", est.Namespace, est.Name, replicas) + return fmt.Errorf("failed to update StatefulSet %s/%s replicas to %d: %w", est.Namespace, est.Name, replicas, err) } + est.Spec.Replicas = &replicas + klog.Infof("updateStatefulSetReplicas: updated StatefulSet %s/%s replicas to %d", est.Namespace, est.Name, replicas) + return nil } // handleWaitPodReady waits for the replacement pod (same ordinal) to become Ready. @@ -784,7 +820,39 @@ func (dcgs *DisaggregatedComputeGroupsController) getBackendByPodName( return backend, nil } } - return nil, fmt.Errorf("backend for pod %s not found", podName) + return nil, fmt.Errorf("%w for pod %s", errBackendNotFound, podName) +} + +func (dcgs *DisaggregatedComputeGroupsController) dropBackendByPodName( + ctx context.Context, + cluster *dv1.DorisDisaggregatedCluster, + cgStatus *dv1.ComputeGroupStatus, + podName string, +) error { + sqlClient, err := dcgs.getMasterSqlClient(ctx, cluster) + if err != nil { + return err + } + defer sqlClient.Close() + + backends, err := sqlClient.GetBackendsByComputeGroupId(cgStatus.ComputeGroupId) + if err != nil { + return err + } + for _, backend := range backends { + if backendMatchesPod(backend, podName) { + return dropBackendEnsuringAbsent(sqlClient, backend) + } + } + klog.Infof("dropBackendByPodName: backend for pod %s is already absent", podName) + return nil +} + +func gracefulScaleDownPodStarted(ga *dv1.GracefulAction) bool { + if ga == nil || ga.Type != dv1.GracefulActionScaleDown { + return false + } + return ga.CurrentPod != "" || ga.DrainTriggered || ga.Phase != dv1.GracefulPhaseTriggerDrain } func backendIsShutdown(backend *mysql.Backend) (bool, error) { diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go index 907d39e2..10be02f3 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go @@ -19,6 +19,8 @@ package computegroups import ( "context" + "errors" + "fmt" "strconv" "strings" @@ -108,10 +110,10 @@ func (dcgs *DisaggregatedComputeGroupsController) scaledOutBENodesByDecommission return nil } -func getOperationType(st, est *appv1.StatefulSet, phase dv1.Phase) string { +func getOperationType(st, est *appv1.StatefulSet, _ dv1.Phase) string { //Should not check 'phase == dv1.Ready', because the default value of the state initialization is Reconciling in the new Reconcile // *st.Spec.Replicas < *est.Spec.Replicas represents need initial scaleDown, it belongs to the start phase. - if *(st.Spec.Replicas) < *(est.Spec.Replicas) || phase == dv1.Decommissioning || phase == dv1.ScaleDownFailed { + if *(st.Spec.Replicas) < *(est.Spec.Replicas) { return "scaleDown" } return "" @@ -131,11 +133,38 @@ func (dcgs *DisaggregatedComputeGroupsController) scaledOutBENodesByDrop( if len(dropNodes) == 0 { return nil } - err = masterDBClient.DropBE(dropNodes) - if err != nil { - klog.Errorf("scaledOutBENodesByDrop cgid %s DropBENodes failed, err:%s ", cgid, err.Error()) - return err + for _, node := range dropNodes { + if err := dropBackendEnsuringAbsent(masterDBClient, node); err != nil { + klog.Errorf("scaledOutBENodesByDrop cgid %s DropBENode failed, err:%s ", cgid, err.Error()) + return err + } + } + return nil +} + +func dropBackendEnsuringAbsent(sqlClient *mysql.DB, target *mysql.Backend) error { + if target == nil { + return errors.New("drop backend target is nil") } + + dropErr := sqlClient.DropBE([]*mysql.Backend{target}) + if dropErr == nil { + return nil + } + + backends, verifyErr := sqlClient.ShowBackends() + if verifyErr != nil { + return fmt.Errorf("drop backend %s:%d failed: %w; verify backend state failed: %v", + target.Host, target.HeartbeatPort, dropErr, verifyErr) + } + for _, backend := range backends { + if backend.Host == target.Host && backend.HeartbeatPort == target.HeartbeatPort { + return fmt.Errorf("drop backend %s:%d failed and backend is still present: %w", + target.Host, target.HeartbeatPort, dropErr) + } + } + + klog.Infof("drop backend %s:%d returned an error but the backend is already absent", target.Host, target.HeartbeatPort) return nil } diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/statefulset_test.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/statefulset_test.go index f81b02f0..b92634ae 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/statefulset_test.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/statefulset_test.go @@ -19,14 +19,17 @@ package computegroups import ( "context" + "errors" "fmt" "strings" "testing" "time" + "github.com/DATA-DOG/go-sqlmock" dv1 "github.com/apache/doris-operator/api/disaggregated/v1" "github.com/apache/doris-operator/pkg/common/utils/mysql" "github.com/apache/doris-operator/pkg/common/utils/resource" + "github.com/jmoiron/sqlx" appv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -38,6 +41,62 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/fake" ) +func TestDropBackendEnsuringAbsent(t *testing.T) { + tests := []struct { + name string + showRows *sqlmock.Rows + showErr error + wantErrText string + }{ + { + name: "drop error but backend is absent", + showRows: sqlmock.NewRows([]string{"Host", "HeartbeatPort"}), + }, + { + name: "drop error and backend is still present", + showRows: sqlmock.NewRows([]string{"Host", "HeartbeatPort"}). + AddRow("test-be-2.test-be-internal.default.svc.cluster.local", 9050), + wantErrText: "backend is still present", + }, + { + name: "drop error and verification fails", + showErr: errors.New("show backends failed"), + wantErrText: "verify backend state failed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mysqlDB, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock new failed: %v", err) + } + db := &mysql.DB{DB: sqlx.NewDb(mysqlDB, "mysql")} + defer db.Close() + + target := &mysql.Backend{Host: "test-be-2.test-be-internal.default.svc.cluster.local", HeartbeatPort: 9050} + mock.ExpectExec("ALTER SYSTEM DROPP BACKEND").WillReturnError(errors.New("drop failed")) + showExpectation := mock.ExpectQuery("show backends") + if tt.showErr != nil { + showExpectation.WillReturnError(tt.showErr) + } else { + showExpectation.WillReturnRows(tt.showRows) + } + + err = dropBackendEnsuringAbsent(db, target) + if tt.wantErrText == "" && err != nil { + t.Fatalf("expected success, got: %v", err) + } + if tt.wantErrText != "" && (err == nil || !strings.Contains(err.Error(), tt.wantErrText)) { + t.Fatalf("expected error containing %q, got: %v", tt.wantErrText, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sql expectations: %v", err) + } + }) + } +} + func Test_NewPodTemplateSpec_TerminationGracePeriodSeconds(t *testing.T) { ddc := &dv1.DorisDisaggregatedCluster{ ObjectMeta: metav1.ObjectMeta{ @@ -336,6 +395,72 @@ func TestGracefulRolloutReconcile_EnablesGracefulActionWhenSentinelSupported(t * } } +func TestGracefulRolloutReconcileCancelsUntouchedScaleDownAfterScaleUp(t *testing.T) { + dcgs, cluster, cg, cgStatus, desired, existing := newGracefulScaleDownTestObjects(t) + requestedReplicas := int32(2) + desired.Spec.Replicas = &requestedReplicas + storedDesiredReplicas := int32(1) + setGracefulAction(existing, &dv1.GracefulAction{ + Type: dv1.GracefulActionScaleDown, + Phase: dv1.GracefulPhaseTriggerDrain, + DesiredReplicas: &storedDesiredReplicas, + }) + if err := dcgs.K8sclient.Update(context.Background(), existing.DeepCopy()); err != nil { + t.Fatalf("update existing StatefulSet: %v", err) + } + + skipApply, err := dcgs.gracefulRolloutReconcile(context.Background(), &rest.Config{}, desired, existing, cluster, cg, cgStatus) + if err != nil { + t.Fatalf("gracefulRolloutReconcile failed: %v", err) + } + if skipApply { + t.Fatal("expected untouched scale-down action to be cancelled") + } + if cgStatus.Phase != dv1.Reconciling { + t.Fatalf("expected phase %s, got %s", dv1.Reconciling, cgStatus.Phase) + } + + live := &appv1.StatefulSet{} + if err := dcgs.K8sclient.Get(context.Background(), client.ObjectKeyFromObject(existing), live); err != nil { + t.Fatalf("get live StatefulSet: %v", err) + } + if hasGracefulAction(live) { + t.Fatalf("expected graceful action annotation to be cleared, got %q", gracefulAnnotationValue(live)) + } +} + +func TestUpdateStatefulSetReplicasUpdatesLiveAndCachedState(t *testing.T) { + dcgs, _, _, _, _, existing := newGracefulScaleDownTestObjects(t) + + if err := dcgs.updateStatefulSetReplicas(context.Background(), existing, 1); err != nil { + t.Fatalf("updateStatefulSetReplicas failed: %v", err) + } + if got := *existing.Spec.Replicas; got != 1 { + t.Fatalf("expected cached StatefulSet replicas 1, got %d", got) + } + + live := &appv1.StatefulSet{} + if err := dcgs.K8sclient.Get(context.Background(), client.ObjectKeyFromObject(existing), live); err != nil { + t.Fatalf("get live StatefulSet: %v", err) + } + if got := *live.Spec.Replicas; got != 1 { + t.Fatalf("expected live StatefulSet replicas 1, got %d", got) + } +} + +func TestGetOperationTypeDoesNotRetryScaleDownAfterScaleUp(t *testing.T) { + desired := newGracefulTestStatefulSet("default", "doris-cg1", 3) + existing := newGracefulTestStatefulSet("default", "doris-cg1", 3) + if got := getOperationType(desired, existing, dv1.ScaleDownFailed); got != "" { + t.Fatalf("expected no scale-down when desired equals existing, got %q", got) + } + + *desired.Spec.Replicas = 2 + if got := getOperationType(desired, existing, dv1.ScaleDownFailed); got != "scaleDown" { + t.Fatalf("expected replica difference to trigger scale-down, got %q", got) + } +} + func TestFinalizeGracefulAction_KeepsOnDeleteStrategy(t *testing.T) { scheme := runtime.NewScheme() if err := appv1.AddToScheme(scheme); err != nil {