diff --git a/doc/operation/scale_up_down.md b/doc/operation/scale_up_down.md index c5105309..70b2052b 100644 --- a/doc/operation/scale_up_down.md +++ b/doc/operation/scale_up_down.md @@ -47,3 +47,32 @@ doriscluster-sample-be-0 1/1 Running 0 3d2h doriscluster-sample-be-2 1/1 Running 0 12m doriscluster-sample-be-1 1/1 Running 0 12m ``` + +## Disaggregated compute group decommission + +With `spec.enableDecommission: true`, lowering a compute group's `replicas` +starts a persistent scale-down task. The operator uses MS `get_cluster` node +states to decide when decommission has finished. `SystemDecommissioned` and +`TabletNum` alone are not completion signals across SelectDB Cloud versions. + +The operator adopts manual decommission SQL for nodes in the requested scale-down +range. On legacy kernels, FE may already report `SystemDecommissioned=true` +while MS still reports `RUNNING`; the operator waits instead of submitting again. +Manual decommission without a CR replica change does not remove Kubernetes pods. + +After MS reports `DECOMMISSIONED`, the operator waits two FE synchronization +periods (at least 20 seconds), using both configured and effective FE settings. +Every registered FE, including observers, must be reachable. The operator checks +FE decommission state before DROP and waits until both MS and all FE views have +removed each target across a further synchronization wait before lowering StatefulSet replicas. MS API errors, unknown +states, changed node identities and DROP failures block scale-down. + +Progress is stored in the StatefulSet annotation +`doris.disaggregated.cluster/decommission-action`. No CRD update is needed. +Inspect it and cluster warning events if a task stalls. Keep the replica target +and `enableDecommission` unchanged until the task finishes. A SQL submission +whose outcome is uncertain is not automatically retried; verify the target's +MS/FE state and complete the decommission manually if necessary. Do not clear +an active task annotation to force a retry without checking its recorded targets. +The operator requires network access to the MS administrative HTTP port and each +FE query port, plus permission to inspect FE configuration. diff --git a/doc/operation/scale_up_down_cn.md b/doc/operation/scale_up_down_cn.md index 863b316b..4ea9376e 100644 --- a/doc/operation/scale_up_down_cn.md +++ b/doc/operation/scale_up_down_cn.md @@ -49,3 +49,24 @@ doriscluster-sample-be-0 1/1 Running 0 3d2h doriscluster-sample-be-2 1/1 Running 0 12m doriscluster-sample-be-1 1/1 Running 0 12m ``` + +## 存算分离计算组退役缩容 + +设置 `spec.enableDecommission: true` 后，下调计算组的 `replicas` 会启动持久化缩容任务。 +Operator 通过 MS `get_cluster` 返回的节点状态判断退役完成，不再仅凭 +`SystemDecommissioned` 或 `TabletNum` 推断，避免不同 SelectDB Cloud 版本的字段语义差异。 + +对于本次缩容范围内的节点，Operator 可以接管用户手动执行的退役 SQL。 +旧版内核可能已在 FE 标记 `SystemDecommissioned=true`，而 MS 仍为 `RUNNING`； +此时继续等待，不重复提交。只执行 SQL、不修改 CR 副本数，不会触发 Kubernetes Pod 缩容。 + +MS 返回 `DECOMMISSIONED` 后，等待两个 FE 同步周期（至少 20 秒），兼顾配置文件和 FE 当前配置。 +所有已注册 FE（包含 Observer）需要可访问。DROP 前检查 FE 的退役状态， +DROP 后确认 MS 和所有 FE 均已移除目标，并在同步等待期后再次确认，再下调 StatefulSet 副本数。 +MS 查询失败、未知状态、节点身份变化或 DROP 失败都会阻断缩容。 + +任务记录保存在 StatefulSet 注解 `doris.disaggregated.cluster/decommission-action`，无需更新 CRD。 +任务停滞时检查该注解和集群 Warning 事件；完成前保持副本目标和 `enableDecommission` 不变。 +SQL 提交结果不确定时不会自动重发，应核对目标的 MS/FE 状态，必要时手动完成退役； +未经核对记录中的目标节点，不要清除活动任务注解来强制重试。 +Operator 需要访问 MS 管理 HTTP 端口和各 FE 查询端口，并具有查看 FE 配置的权限。 diff --git a/pkg/common/utils/metaservice/client.go b/pkg/common/utils/metaservice/client.go new file mode 100644 index 00000000..b2dbf2e7 --- /dev/null +++ b/pkg/common/utils/metaservice/client.go @@ -0,0 +1,181 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 metaservice + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +type NodeStatus string + +const ( + Running NodeStatus = "NODE_STATUS_RUNNING" + Decommissioning NodeStatus = "NODE_STATUS_DECOMMISSIONING" + Decommissioned NodeStatus = "NODE_STATUS_DECOMMISSIONED" +) + +// Node is the subset of NodeInfoPB needed for compute group scale-down. +type Node struct { + CloudUniqueID string `json:"cloud_unique_id"` + Host string `json:"host"` + IP string `json:"ip"` + HeartbeatPort int `json:"heartbeat_port"` + Status NodeStatus `json:"status"` +} + +// UnmarshalJSON accepts protobuf enum names and their numeric representation. +func (s *NodeStatus) UnmarshalJSON(data []byte) error { + var name string + if err := json.Unmarshal(data, &name); err == nil { + *s = NodeStatus(name) + return nil + } + var number int + if err := json.Unmarshal(data, &number); err != nil { + return errors.New("invalid MS node status") + } + switch number { + case 1: + *s = Running + case 3: + *s = Decommissioning + case 4: + *s = Decommissioned + default: + *s = NodeStatus(fmt.Sprintf("UNKNOWN_%d", number)) + } + return nil +} + +type Cluster struct { + ID string `json:"cluster_id"` + Nodes []Node `json:"nodes"` +} + +type Client struct { + endpoint string + token string + httpClient *http.Client +} + +func NewClient(endpoint, token string) *Client { + if !strings.Contains(endpoint, "://") { + endpoint = "http://" + endpoint + } + return &Client{endpoint: strings.TrimRight(endpoint, "/"), token: token, + httpClient: &http.Client{Timeout: 10 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }}} +} + +// GetCluster uses the administrative HTTP API shared by SelectDB Cloud 3.x and +// 4.x. With cluster_id specified, result is a single ClusterPB, not a list. +func (c *Client) GetCluster(ctx context.Context, instanceID, cloudUniqueID, clusterID string) (*Cluster, error) { + if cloudUniqueID == "" || clusterID == "" { + return nil, errors.New("MS cluster query requires cloud_unique_id and cluster_id") + } + body, err := json.Marshal(map[string]string{"instance_id": instanceID, "cloud_unique_id": cloudUniqueID, "cluster_id": clusterID}) + if err != nil { + return nil, err + } + var cluster Cluster + if err := c.query(ctx, "get_cluster", nil, body, &cluster); err != nil { + return nil, err + } + if cluster.ID != clusterID { + return nil, errors.New("MS cluster response is missing the requested cluster") + } + return &cluster, nil +} + +// GetInstanceID resolves legacy cloud_unique_id values instead of assuming that +// older deployments use the operator's cluster hash as their MS instance ID. +func (c *Client) GetInstanceID(ctx context.Context, cloudUniqueID string) (string, error) { + if cloudUniqueID == "" { + return "", errors.New("MS instance query requires cloud_unique_id") + } + var instance struct { + ID string `json:"instance_id"` + } + if err := c.query(ctx, "get_instance", url.Values{"cloud_unique_id": {cloudUniqueID}}, nil, &instance); err != nil { + return "", err + } + if instance.ID == "" { + return "", errors.New("MS instance response is missing instance_id") + } + return instance.ID, nil +} + +func (c *Client) query(ctx context.Context, path string, query url.Values, body []byte, result interface{}) error { + endpoint, err := url.Parse(c.endpoint + "/MetaService/http/" + path) + if err != nil || endpoint.Host == "" { + return errors.New("invalid MS endpoint") + } + if query == nil { + query = url.Values{} + } + query.Set("token", c.token) + // MS reads query values through brpc's URI::GetQuery without URL decoding. + // Colons are legal in a URI query and must remain literal in cloud_unique_id + // (for example, 1:instance:backend), rather than being sent as %3A. + endpoint.RawQuery = strings.ReplaceAll(query.Encode(), "%3A", ":") + method := http.MethodPost + if body == nil { + method = http.MethodGet + } + req, err := http.NewRequestWithContext(ctx, method, endpoint.String(), bytes.NewReader(body)) + if err != nil { + return errors.New("cannot construct MS query") + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + // net/http errors include the request URL, which contains the MS token. + return errors.New("MS query failed (network error or timeout)") + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("MS query returned HTTP %d", resp.StatusCode) + } + var envelope struct { + Code string `json:"code"` + Result json.RawMessage `json:"result"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&envelope); err != nil { + return errors.New("invalid MS response") + } + if envelope.Code != "OK" { + return errors.New("MS query did not return OK") + } + if len(envelope.Result) == 0 || string(envelope.Result) == "null" { + return errors.New("MS response is missing result") + } + if err := json.Unmarshal(envelope.Result, result); err != nil { + return errors.New("invalid MS result") + } + return nil +} diff --git a/pkg/common/utils/metaservice/client_test.go b/pkg/common/utils/metaservice/client_test.go new file mode 100644 index 00000000..d97e6cec --- /dev/null +++ b/pkg/common/utils/metaservice/client_test.go @@ -0,0 +1,165 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 metaservice + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestGetCluster(t *testing.T) { + for _, state := range []string{`"NODE_STATUS_DECOMMISSIONING"`, `3`} { + t.Run(state, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/MetaService/http/get_cluster" || r.URL.Query().Get("token") != "a+b&c" { + t.Errorf("unexpected MS request: method=%s path=%s", r.Method, r.URL.Path) + } + var body map[string]string + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + } + if body["instance_id"] != "instance" || body["cloud_unique_id"] != "1:instance:operator" || body["cluster_id"] != "cg" { + t.Errorf("body=%v", body) + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"code":"OK","result":{"cluster_id":"cg","nodes":[{"cloud_unique_id":"be-id","host":"doris-cg-1.example","heartbeat_port":9050,"status":` + state + `}]}}`)) + })) + defer server.Close() + cluster, err := NewClient(server.URL, "a+b&c").GetCluster(context.Background(), "instance", "1:instance:operator", "cg") + if err != nil { + t.Fatal(err) + } + if cluster.ID != "cg" || len(cluster.Nodes) != 1 || cluster.Nodes[0].Status != Decommissioning { + t.Fatalf("unexpected cluster: %+v", cluster) + } + }) + } +} + +func TestGetClusterRejectsUncertainResponses(t *testing.T) { + tests := []struct { + name string + status int + body string + }{ + {"http failure", 503, `unavailable`}, + {"not found is not absence", 404, `{"code":"NOT_FOUND"}`}, + {"business failure", 200, `{"code":"INTERNAL_ERROR","result":{"cluster_id":"cg"}}`}, + {"missing code", 200, `{"result":{"cluster_id":"cg"}}`}, + {"missing result", 200, `{"code":"OK"}`}, + {"wrong cluster", 200, `{"code":"OK","result":{"cluster_id":"other"}}`}, + {"malformed", 200, `{"code":`}, + {"wrong result shape", 200, `{"code":"OK","result":{"cluster":[{"cluster_id":"cg"}]}}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(tt.status); w.Write([]byte(tt.body)) })) + defer server.Close() + _, err := NewClient(server.URL, "sensitive-token").GetCluster(context.Background(), "instance", "id", "cg") + if err == nil { + t.Fatal("uncertain response must fail closed") + } + if strings.Contains(err.Error(), "sensitive-token") { + t.Fatal("error leaked token") + } + }) + } +} + +func TestGetClusterEmptyNodes(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"code":"OK","result":{"cluster_id":"cg"}}`)) + })) + defer server.Close() + cluster, err := NewClient(server.URL, "token").GetCluster(context.Background(), "instance", "id", "cg") + if err != nil || len(cluster.Nodes) != 0 { + t.Fatalf("empty protobuf repeated field: cluster=%v err=%v", cluster, err) + } +} + +func TestGetClusterCancellationAndRedirectDoNotLeakToken(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, "http://example.invalid", http.StatusFound) + })) + defer server.Close() + client := NewClient(server.URL, "sensitive-token") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for _, ctx := range []context.Context{ctx, context.Background()} { + _, err := client.GetCluster(ctx, "instance", "id", "cg") + if err == nil || strings.Contains(err.Error(), "sensitive-token") { + t.Fatalf("unsafe error: %v", err) + } + } + client.httpClient.Timeout = time.Nanosecond + _, err := client.GetCluster(context.Background(), "instance", "id", "cg") + if err == nil || strings.Contains(err.Error(), "sensitive-token") { + t.Fatalf("unsafe timeout error: %v", err) + } +} + +func TestGetInstanceIDForLegacyCloudUniqueID(t *testing.T) { + for _, tt := range []struct { + name, body string + wantErr bool + }{ + {name: "resolved", body: `{"code":"OK","result":{"instance_id":"legacy-instance","obj_info":[{"sk":"not-returned"}]}}`}, + {name: "missing ID", body: `{"code":"OK","result":{}}`, wantErr: true}, + {name: "missing instance", body: `{"code":"NOT_FOUND"}`, wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" || r.URL.Path != "/MetaService/http/get_instance" || r.URL.Query().Get("cloud_unique_id") != "opaque-legacy-id" { + t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path) + } + w.Write([]byte(tt.body)) + })) + defer server.Close() + instanceID, err := NewClient(server.URL, "token").GetInstanceID(context.Background(), "opaque-legacy-id") + if (err != nil) != tt.wantErr { + t.Fatalf("instance=%s err=%v", instanceID, err) + } + if !tt.wantErr && instanceID != "legacy-instance" { + t.Fatalf("instance=%s", instanceID) + } + }) + } +} + +func TestGetInstanceIDPreservesColonsForBRPC(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Inspect the wire query, as brpc does, rather than net/url's decoded + // Query() result, which would hide an incompatible escaped colon. + if r.URL.RawQuery != "cloud_unique_id=1:instance:backend&token=token" { + t.Errorf("unexpected raw query: %s", r.URL.RawQuery) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Write([]byte(`{"code":"OK","result":{"instance_id":"instance"}}`)) + })) + defer server.Close() + id, err := NewClient(server.URL, "token").GetInstanceID(context.Background(), "1:instance:backend") + if err != nil || id != "instance" { + t.Fatalf("instance=%s err=%v", id, err) + } +} diff --git a/pkg/common/utils/mysql/mysql.go b/pkg/common/utils/mysql/mysql.go index aabe3ce5..4dd89b9a 100644 --- a/pkg/common/utils/mysql/mysql.go +++ b/pkg/common/utils/mysql/mysql.go @@ -24,6 +24,8 @@ import ( "encoding/json" "errors" "fmt" + "net/url" + "time" "github.com/go-sql-driver/mysql" _ "github.com/go-sql-driver/mysql" @@ -42,6 +44,8 @@ type DBConfig struct { Host string Port string Database string + // Timeout bounds connection establishment and each network read/write when set. + Timeout time.Duration } type TLSConfig struct { @@ -65,6 +69,12 @@ type DB struct { func NewDorisSqlDB(cfg DBConfig, tlsConfig *TLSConfig, secret *corev1.Secret) (*DB, error) { dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Database) + params := url.Values{} + if cfg.Timeout > 0 { + for _, key := range []string{"timeout", "readTimeout", "writeTimeout"} { + params.Set(key, cfg.Timeout.String()) + } + } rootCertPool := x509.NewCertPool() if tlsConfig != nil && secret != nil { @@ -89,9 +99,12 @@ func NewDorisSqlDB(cfg DBConfig, tlsConfig *TLSConfig, secret *corev1.Secret) (* }); err != nil { return nil, errors.New("NewDorisSqlDB register tls config failed," + err.Error()) } - dsn = dsn + "?tls=" + registerKey + params.Set("tls", registerKey) } + if len(params) > 0 { + dsn += "?" + params.Encode() + } db, err := sqlx.Open("mysql", dsn) if err != nil { klog.Errorf("NewDorisSqlDB sqlx.Open failed open doris sql client connection, err: %s \n", err.Error()) @@ -99,6 +112,7 @@ func NewDorisSqlDB(cfg DBConfig, tlsConfig *TLSConfig, secret *corev1.Secret) (* } if err = db.Ping(); err != nil { + db.Close() klog.Errorf("NewDorisSqlDB sqlx.Open.Ping failed ping doris sql client connection, err: %s \n", err.Error()) return nil, err } @@ -129,6 +143,7 @@ func NewDorisMasterSqlDB(dbConf DBConfig, tlsConfig *TLSConfig, secret *corev1.S Host: master.Host, Port: dbConf.Port, Database: "mysql", + Timeout: dbConf.Timeout, }, tlsConfig, secret) if err != nil { klog.Errorf("NewDorisMasterSqlDB failed, get fe master connection err:%s", err.Error()) diff --git a/pkg/common/utils/mysql/mysql_test.go b/pkg/common/utils/mysql/mysql_test.go index cb7d8c20..ad68ab46 100644 --- a/pkg/common/utils/mysql/mysql_test.go +++ b/pkg/common/utils/mysql/mysql_test.go @@ -22,9 +22,11 @@ import ( "database/sql/driver" "errors" "fmt" + "net" "regexp" "strconv" "testing" + "time" "github.com/DATA-DOG/go-sqlmock" "github.com/jmoiron/sqlx" @@ -288,3 +290,36 @@ func TestDropBEReturnsUnexpectedError(t *testing.T) { t.Fatalf("unmet sql expectations: %v", err) } } + +func TestNewDorisSqlDBBoundsStalledHandshake(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + release := make(chan struct{}) + defer close(release) + go func() { + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + <-release + }() + host, port, err := net.SplitHostPort(listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + started := time.Now() + db, err := NewDorisSqlDB(DBConfig{Host: host, Port: port, Database: "mysql", Timeout: 50 * time.Millisecond}, nil, nil) + if db != nil { + db.Close() + } + if err == nil { + t.Fatal("stalled handshake succeeded") + } + if time.Since(started) > 2*time.Second { + t.Fatal("SQL timeout did not bound the stalled handshake") + } +} diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go index bb3bdd32..fd680ab5 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go @@ -251,8 +251,21 @@ func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx conte return &sc.Event{Type: sc.EventWarning, Reason: sc.CGStorageTemplateImmutable, Message: msg}, errors.New(msg) } + // Finish an existing graceful action before starting SQL decommission. Keep + // replicas unchanged while that action owns the pod lifecycle. + if cluster.Spec.EnableDecommission && hasGracefulAction(&est) { + replicas := *est.Spec.Replicas + st.Spec.Replicas = &replicas + } + if pending, err := dcgs.reconcileDecommissionScaleDown(ctx, st, &est, cluster, cg); pending || err != nil { + if err != nil { + return &sc.Event{Type: sc.EventWarning, Reason: sc.CGSqlExecFailed, Message: err.Error()}, err + } + 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. + // supports it. SQL decommission is handled above before applying resources. gracefulInProgress := false if !cluster.Spec.EnableDecommission { gracefulInProgress = dcgs.reconcileGracefulStatefulSet(ctx, st, &est, cluster, cg) diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission.go new file mode 100644 index 00000000..a57d91a9 --- /dev/null +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission.go @@ -0,0 +1,577 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 computegroups + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + dv1 "github.com/apache/doris-operator/api/disaggregated/v1" + "github.com/apache/doris-operator/pkg/common/utils/k8s" + ms "github.com/apache/doris-operator/pkg/common/utils/metaservice" + "github.com/apache/doris-operator/pkg/common/utils/mysql" + "github.com/apache/doris-operator/pkg/common/utils/resource" + sc "github.com/apache/doris-operator/pkg/controller/sub_controller" + appv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const decommissionActionAnnotation = "doris.disaggregated.cluster/decommission-action" + +const ( + decommissionPrepared = "Prepared" + decommissionSubmitted = "Submitted" + decommissionWaitFESync = "WaitFESync" + decommissionDropping = "Dropping" +) + +// The task is persisted before any SQL is sent. A Submitted task is deliberately +// not resubmitted on restart or transport failure: SQL may already have committed. +type decommissionAction struct { + ComputeGroupID string `json:"computeGroupID"` + FromReplicas int32 `json:"fromReplicas"` + Replicas int32 `json:"replicas"` + CloudUniqueID string `json:"cloudUniqueID"` + InstanceID string `json:"instanceID"` + Targets []decommissionTarget `json:"targets"` +} + +type decommissionTarget struct { + Ordinal int32 `json:"ordinal"` + Host string `json:"host,omitempty"` + HeartbeatPort int `json:"heartbeatPort,omitempty"` + BackendID string `json:"backendID,omitempty"` + CloudUniqueID string `json:"cloudUniqueID,omitempty"` + Phase string `json:"phase"` + CompletedAt *metav1.Time `json:"completedAt,omitempty"` + AbsentSince *metav1.Time `json:"absentSince,omitempty"` +} + +type decommissionSnapshot struct { + Nodes map[int32]ms.Node + Backends map[int32]*mysql.Backend + // One observation per registered FE, including observers. No FE may retain a + // target when Kubernetes removes its pod. + FEViews []map[int32]*mysql.Backend + SyncWait time.Duration +} + +type decommissionSQL interface { + DecommissionBE([]*mysql.Backend) error + DropBE([]*mysql.Backend) error +} + +// reconcileDecommissionScaleDown owns replica changes while decommission is +// enabled. Metadata-only patches leave the running pod template untouched. +func (dcgs *DisaggregatedComputeGroupsController) reconcileDecommissionScaleDown(ctx context.Context, st, est *appv1.StatefulSet, cluster *dv1.DorisDisaggregatedCluster, cg *dv1.ComputeGroup) (bool, error) { + raw := est.Annotations[decommissionActionAnnotation] + if raw == "" && (!cluster.Spec.EnableDecommission || *st.Spec.Replicas >= *est.Spec.Replicas) { + return false, 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 cgStatus == nil { + return true, errors.New("compute group status is missing during decommission") + } + cgStatus.Phase = dv1.Decommissioning + if !cluster.Spec.EnableDecommission { + return true, errors.New("restore enableDecommission to finish the active decommission task") + } + if *st.Spec.Replicas < 0 || cgStatus.ComputeGroupId == "" { + return true, errors.New("decommission requires a valid compute group ID and nonnegative replicas") + } + var action decommissionAction + if raw != "" { + if err := json.Unmarshal([]byte(raw), &action); err != nil { + return true, errors.New("invalid decommission action annotation") + } + if err := action.validate(*st.Spec.Replicas, *est.Spec.Replicas, cgStatus.ComputeGroupId); err != nil { + return true, err + } + } + sqlClient, err := dcgs.getMasterSqlClient(ctx, cluster) + if err != nil { + return true, err + } + defer sqlClient.Close() + backends, err := sqlClient.GetBackendsByComputeGroupId(cgStatus.ComputeGroupId) + if err != nil { + return true, err + } + if raw == "" { + action = decommissionAction{ComputeGroupID: cgStatus.ComputeGroupId, FromReplicas: *est.Spec.Replicas, Replicas: *st.Spec.Replicas} + for ordinal := action.FromReplicas - 1; ordinal >= action.Replicas; ordinal-- { + action.Targets = append(action.Targets, decommissionTarget{Ordinal: ordinal, Phase: decommissionPrepared}) + } + feConf := dcgs.GetConfigValuesFromConfigMaps(cluster.Namespace, resource.FE_RESOLVEKEY, cluster.Spec.FeSpec.ConfigMaps) + if configured, ok := feConf["cloud_unique_id"]; ok { + action.CloudUniqueID = fmt.Sprint(configured) + } + for _, backend := range backends { + var tag struct { + CloudUniqueID string `json:"cloud_unique_id"` + } + if err := json.Unmarshal([]byte(backend.Tag), &tag); err != nil { + return true, err + } + if tag.CloudUniqueID != "" { + action.CloudUniqueID = tag.CloudUniqueID + break + } + } + if action.CloudUniqueID == "" { + // A group can already be empty before a scale-down is requested. DDC + // images use this degraded ID format for the operator's CLUSTER_ID. + action.CloudUniqueID = fmt.Sprintf("1:%d:operator", cluster.GetInstanceHashId()) + } + } + snapshot, err := dcgs.decommissionSnapshot(ctx, cluster, cg, &action, sqlClient, backends) + if err != nil { + return true, err + } + // Manual SQL on retained nodes does not expand the operator's deletion range. + // Surface the discrepancy without cancelling or dropping those nodes. + retiringRetained := map[int32]bool{} + for ordinal, node := range snapshot.Nodes { + if ordinal < action.Replicas && (node.Status == ms.Decommissioning || node.Status == ms.Decommissioned) { + retiringRetained[ordinal] = true + } + } + for _, view := range snapshot.FEViews { + for ordinal, backend := range view { + if ordinal < action.Replicas && backend.SystemDecommissioned { + retiringRetained[ordinal] = true + } + } + } + for ordinal := range retiringRetained { + dcgs.K8srecorder.Eventf(cluster, string(sc.EventWarning), "RetainedBackendDecommissioning", "Compute group %s backend ordinal %d is under decommission outside the requested scale-down range", cg.UniqueId, ordinal) + } + save := func() error { return dcgs.saveDecommissionAction(ctx, est, &action) } + if raw == "" { + // Observe and persist the identities of this CR's targets before SQL. + for i := range action.Targets { + if err := bindDecommissionTarget(&action.Targets[i], snapshot); err != nil { + return true, err + } + } + return true, save() + } + done, err := advanceDecommission(&action, snapshot, time.Now(), sqlClient, save) + if err != nil { + dcgs.K8srecorder.Eventf(cluster, string(sc.EventWarning), string(sc.CGSqlExecFailed), "Decommission of compute group %s is blocked: %s", cg.UniqueId, err) + return true, err + } + if !done { + return true, nil + } + if err := dcgs.finishDecommissionAction(ctx, est, action.Replicas); err != nil { + return true, err + } + cgStatus.Phase = dv1.Scaling + return true, nil +} + +// Remove the task and lower replicas atomically, after every MS and FE view +// confirms absence. A restart cannot lose progress before replicas are changed. +func (dcgs *DisaggregatedComputeGroupsController) finishDecommissionAction(ctx context.Context, est *appv1.StatefulSet, replicas int32) error { + before := est.DeepCopy() + delete(est.Annotations, decommissionActionAnnotation) + est.Spec.Replicas = &replicas + return dcgs.K8sclient.Patch(ctx, est, client.MergeFromWithOptions(before, client.MergeFromWithOptimisticLock{})) +} + +func (a *decommissionAction) validate(replicas, current int32, cgid string) error { + if a.ComputeGroupID == "" || a.ComputeGroupID != cgid || a.CloudUniqueID == "" || a.InstanceID == "" || a.Replicas < 0 || a.FromReplicas <= a.Replicas || len(a.Targets) != int(a.FromReplicas-a.Replicas) { + return errors.New("invalid decommission task identity or replica range") + } + if replicas != a.Replicas || current != a.FromReplicas { + return errors.New("replica target changed during decommission; restore the original target to finish the active task") + } + for i, target := range a.Targets { + if target.Ordinal != a.FromReplicas-1-int32(i) { + return errors.New("invalid decommission target ordinal") + } + switch target.Phase { + case decommissionPrepared, decommissionSubmitted, decommissionWaitFESync, decommissionDropping: + default: + return errors.New("invalid decommission target phase") + } + } + return nil +} + +func (dcgs *DisaggregatedComputeGroupsController) saveDecommissionAction(ctx context.Context, est *appv1.StatefulSet, action *decommissionAction) error { + value, err := json.Marshal(action) + if err != nil { + return err + } + before := est.DeepCopy() + if est.Annotations == nil { + est.Annotations = map[string]string{} + } + est.Annotations[decommissionActionAnnotation] = string(value) + return dcgs.K8sclient.Patch(ctx, est, client.MergeFromWithOptions(before, client.MergeFromWithOptimisticLock{})) +} + +func (dcgs *DisaggregatedComputeGroupsController) decommissionSnapshot(ctx context.Context, cluster *dv1.DorisDisaggregatedCluster, cg *dv1.ComputeGroup, action *decommissionAction, sqlClient *mysql.DB, backends []*mysql.Backend) (*decommissionSnapshot, error) { + conf := dcgs.GetConfigValuesFromConfigMaps(cluster.Namespace, resource.MS_RESOLVEKEY, cluster.Spec.MetaService.ConfigMaps) + endpoint := fmt.Sprintf("%s.%s:%d", cluster.GetMSServiceName(), cluster.Namespace, resource.GetPort(conf, resource.BRPC_LISTEN_PORT)) + token := resource.DefaultMsToken + if value, ok := conf[resource.DefaultMsTokenKey]; ok { + token = fmt.Sprint(value) + } + msClient := ms.NewClient(endpoint, token) + if action.InstanceID == "" { + instanceID, err := msClient.GetInstanceID(ctx, action.CloudUniqueID) + if err != nil { + return nil, err + } + action.InstanceID = instanceID + action.CloudUniqueID = "1:" + instanceID + ":operator" + } + remote, err := msClient.GetCluster(ctx, action.InstanceID, action.CloudUniqueID, action.ComputeGroupID) + if err != nil { + return nil, err + } + podOrdinals := map[string]int32{} + var pods corev1.PodList + if err := dcgs.K8sclient.List(ctx, &pods, client.InNamespace(cluster.Namespace), client.MatchingLabels(dcgs.newCGPodsSelector(cluster.Name, cg.UniqueId))); err != nil { + return nil, err + } + stsName := cluster.GetCGStatefulsetName(cg) + for _, pod := range pods.Items { + ordinal, err := decommissionOrdinal(stsName, pod.Name, nil) + if err != nil { + return nil, err + } + if pod.Status.PodIP != "" { + podOrdinals[pod.Status.PodIP] = ordinal + } + } + snapshot := &decommissionSnapshot{Nodes: map[int32]ms.Node{}, Backends: map[int32]*mysql.Backend{}} + for _, node := range remote.Nodes { + host := node.Host + if host == "" { + host = node.IP + } + ordinal, err := decommissionOrdinal(stsName, host, podOrdinals) + if err != nil { + return nil, err + } + if node.CloudUniqueID == "" || node.HeartbeatPort <= 0 { + return nil, errors.New("MS node identity or heartbeat port is missing") + } + if _, exists := snapshot.Nodes[ordinal]; exists { + return nil, errors.New("multiple MS nodes map to the same pod ordinal") + } + snapshot.Nodes[ordinal] = node + } + snapshot.Backends, err = indexDecommissionBackends(stsName, podOrdinals, backends) + if err != nil { + return nil, err + } + // Read every registered FE directly: a query through the Service or master + // alone cannot prove that followers and observers have synchronized removal. + frontends, err := sqlClient.ShowFrontends() + if err != nil { + return nil, err + } + if len(frontends) == 0 { + return nil, errors.New("no FE observations are available") + } + feConf := dcgs.GetConfigValuesFromConfigMaps(cluster.Namespace, resource.FE_RESOLVEKEY, cluster.Spec.FeSpec.ConfigMaps) + interval := 10 * time.Second + if value, ok := feConf["cloud_cluster_check_interval_second"]; ok { + seconds, err := strconv.Atoi(fmt.Sprint(value)) + if err != nil || seconds <= 0 || seconds > 86400 { + return nil, errors.New("invalid cloud_cluster_check_interval_second") + } + if time.Duration(seconds)*time.Second > interval { + interval = time.Duration(seconds) * time.Second + } + } + // Allow two configured synchronization periods, and still verify FE state. + snapshot.SyncWait = 2 * interval + user, password := dcgs.GetManagementAdminUserAndPWD(ctx, cluster) + tlsConfig, secretName := dcgs.DisaggregatedSubDefaultController.FindSecretTLSConfig(feConf, cluster) + secret, err := k8s.GetSecret(ctx, dcgs.K8sclient, cluster.Namespace, secretName) + if tlsConfig != nil && err != nil { + return nil, err + } + for _, fe := range frontends { + if !fe.Alive { + return nil, fmt.Errorf("waiting for FE %s to become alive and synchronize", fe.Host) + } + view := snapshot.Backends + feDB := sqlClient + if !fe.IsMaster { + db, err := mysql.NewDorisSqlDB(mysql.DBConfig{User: user, Password: password, Host: fe.Host, Port: strconv.Itoa(fe.QueryPort), Database: "mysql", Timeout: 10 * time.Second}, tlsConfig, secret) + if err != nil { + return nil, fmt.Errorf("cannot observe FE %s: %w", fe.Host, err) + } + feDB = db + nodes, queryErr := db.GetBackendsByComputeGroupId(action.ComputeGroupID) + if queryErr != nil { + db.Close() + return nil, queryErr + } + view, err = indexDecommissionBackends(stsName, podOrdinals, nodes) + if err != nil { + db.Close() + return nil, err + } + } + actualInterval, configErr := frontendSyncInterval(feDB) + if !fe.IsMaster { + feDB.Close() + } + if configErr != nil { + return nil, configErr + } + if 2*actualInterval > snapshot.SyncWait { + snapshot.SyncWait = 2 * actualInterval + } + snapshot.FEViews = append(snapshot.FEViews, view) + } + return snapshot, nil +} + +// Early 3.x keeps the interval in FeConstants and returns no matching row. +// Later versions expose it through FE config, including changes made via SQL. +func frontendSyncInterval(db *mysql.DB) (time.Duration, error) { + var rows []struct { + Value string `db:"Value"` + } + if err := db.USelect(&rows, "ADMIN SHOW FRONTEND CONFIG LIKE 'cloud_cluster_check_interval_second'"); err != nil { + return 0, err + } + if len(rows) == 0 { + return 10 * time.Second, nil + } + seconds, err := strconv.Atoi(rows[0].Value) + if err != nil || seconds <= 0 || seconds > 86400 { + return 0, errors.New("invalid FE synchronization interval") + } + return time.Duration(seconds) * time.Second, nil +} + +func decommissionOrdinal(stsName, host string, podIPs map[string]int32) (int32, error) { + if ordinal, ok := podIPs[host]; ok { + return ordinal, nil + } + name := strings.Split(host, ".")[0] + prefix := stsName + "-" + if !strings.HasPrefix(name, prefix) { + return 0, fmt.Errorf("cannot map backend host %q to StatefulSet %s", host, stsName) + } + ordinal, err := strconv.ParseInt(strings.TrimPrefix(name, prefix), 10, 32) + if err != nil || ordinal < 0 { + return 0, fmt.Errorf("invalid backend ordinal in host %q", host) + } + return int32(ordinal), nil +} + +func indexDecommissionBackends(stsName string, podIPs map[string]int32, backends []*mysql.Backend) (map[int32]*mysql.Backend, error) { + indexed := map[int32]*mysql.Backend{} + for _, backend := range backends { + ordinal, err := decommissionOrdinal(stsName, backend.Host, podIPs) + if err != nil { + return nil, err + } + if _, exists := indexed[ordinal]; exists { + return nil, errors.New("multiple FE backends map to the same pod ordinal") + } + indexed[ordinal] = backend + } + return indexed, nil +} + +func bindDecommissionTarget(target *decommissionTarget, snapshot *decommissionSnapshot) error { + if node, ok := snapshot.Nodes[target.Ordinal]; ok { + if target.CloudUniqueID != "" && target.CloudUniqueID != node.CloudUniqueID { + return errors.New("MS target identity changed during decommission") + } + target.CloudUniqueID = node.CloudUniqueID + } + if backend := snapshot.Backends[target.Ordinal]; backend != nil { + if backend.BackendID == "" || backend.HeartbeatPort <= 0 { + return errors.New("FE target identity or heartbeat port is missing") + } + var tag struct { + CloudUniqueID string `json:"cloud_unique_id"` + } + if backend.Tag != "" { + if err := json.Unmarshal([]byte(backend.Tag), &tag); err != nil { + return errors.New("invalid target backend tag") + } + if tag.CloudUniqueID != "" && target.CloudUniqueID != "" && tag.CloudUniqueID != target.CloudUniqueID { + return errors.New("MS and FE target identities disagree") + } + } + if target.BackendID != "" && target.BackendID != backend.BackendID { + return errors.New("FE target identity changed during decommission") + } + if target.Host != "" && (target.Host != backend.Host || target.HeartbeatPort != backend.HeartbeatPort) { + return errors.New("target endpoint changed during decommission") + } + target.BackendID, target.Host, target.HeartbeatPort = backend.BackendID, backend.Host, backend.HeartbeatPort + } + for _, view := range snapshot.FEViews { + if backend := view[target.Ordinal]; backend != nil && target.BackendID != "" && backend.BackendID != target.BackendID { + return errors.New("FE views disagree about the target backend identity") + } + } + return nil +} + +func advanceDecommission(action *decommissionAction, snapshot *decommissionSnapshot, now time.Time, sqlClient decommissionSQL, save func() error) (bool, error) { + if len(snapshot.FEViews) == 0 || snapshot.SyncWait < 10*time.Second { + return false, errors.New("FE observations and a synchronization interval are required before decommission") + } + allAbsent := true + for i := range action.Targets { + target := &action.Targets[i] + if err := bindDecommissionTarget(target, snapshot); err != nil { + return false, err + } + node, inMS := snapshot.Nodes[target.Ordinal] + backend := snapshot.Backends[target.Ordinal] + inFE, feSynchronized := false, true + for _, view := range snapshot.FEViews { + if be := view[target.Ordinal]; be != nil { + inFE = true + if !be.SystemDecommissioned { + feSynchronized = false + } + } + } + if !inMS && !inFE && backend == nil { + // Some MS versions implement state changes using a remove/add pair. + // A single missing observation must not be mistaken for a durable DROP. + if target.AbsentSince == nil { + observedAt := metav1.NewTime(now.Truncate(time.Second).Add(time.Second)) + target.AbsentSince = &observedAt + if err := save(); err != nil { + return false, err + } + allAbsent = false + } else if now.Before(target.AbsentSince.Add(snapshot.SyncWait)) { + allAbsent = false + } + continue + } + allAbsent = false + if target.AbsentSince != nil { + target.AbsentSince = nil + if err := save(); err != nil { + return false, err + } + } + if !inMS { + // MS may already have removed this node while FE still exposes it. + // Wait for every FE; do not issue SQL against a stale FE-only entry. + continue + } + switch node.Status { + case ms.Running: + if target.CompletedAt != nil || target.Phase == decommissionDropping { + return false, errors.New("MS target returned to RUNNING after completion; refusing to drop it") + } + // In 3.x and early 4.0, manual SQL marks FE first and may leave MS RUNNING. + manuallyStarted := backend != nil && backend.SystemDecommissioned + for _, view := range snapshot.FEViews { + if be := view[target.Ordinal]; be != nil && be.SystemDecommissioned { + manuallyStarted = true + } + } + if manuallyStarted || target.Phase == decommissionSubmitted { + if target.Phase != decommissionSubmitted { + target.Phase = decommissionSubmitted + if err := save(); err != nil { + return false, err + } + } + continue + } + if backend == nil { + continue + } // Wait for registration to reach the master. + target.Phase = decommissionSubmitted + if err := save(); err != nil { + return false, err + } + if err := sqlClient.DecommissionBE([]*mysql.Backend{backend}); err != nil { + return false, fmt.Errorf("decommission submission result is uncertain; inspect backend %s before retrying: %w", backend.Host, err) + } + case ms.Decommissioning: + if target.Phase == decommissionDropping { + return false, errors.New("MS target regressed after DROP was requested") + } + if target.Phase != decommissionSubmitted || target.CompletedAt != nil { + target.Phase = decommissionSubmitted + target.CompletedAt = nil + if err := save(); err != nil { + return false, err + } + } + case ms.Decommissioned: + if target.CompletedAt == nil { + completedAt := metav1.NewTime(now.Truncate(time.Second).Add(time.Second)) + target.CompletedAt = &completedAt + target.Phase = decommissionWaitFESync + if err := save(); err != nil { + return false, err + } + continue + } + if now.Before(target.CompletedAt.Add(snapshot.SyncWait)) || !feSynchronized { + continue + } + if backend == nil { + continue + } // MS still contains the node; no safe SQL endpoint yet. + if target.Phase != decommissionDropping { + target.Phase = decommissionDropping + if err := save(); err != nil { + return false, err + } + } + // Retrying DROP is safe: the snapshot above still proves this exact node is + // completed. Failures block replica changes and are verified on the next poll. + if err := sqlClient.DropBE([]*mysql.Backend{backend}); err != nil { + return false, fmt.Errorf("drop decommissioned backend %s failed: %w", backend.Host, err) + } + default: + return false, fmt.Errorf("unknown MS state %q for ordinal %d", node.Status, target.Ordinal) + } + } + return allAbsent, nil +} diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission_test.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission_test.go new file mode 100644 index 00000000..7a7122e4 --- /dev/null +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/decommission_test.go @@ -0,0 +1,468 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 computegroups + +import ( + "context" + "encoding/json" + "errors" + "regexp" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + dv1 "github.com/apache/doris-operator/api/disaggregated/v1" + ms "github.com/apache/doris-operator/pkg/common/utils/metaservice" + "github.com/apache/doris-operator/pkg/common/utils/mysql" + "github.com/jmoiron/sqlx" + appv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +type decommissionSQLRecorder struct { + decommissionCalls, dropCalls int + decommissionErr, dropErr error + beforeDecommission, beforeDrop func() +} + +func (r *decommissionSQLRecorder) DecommissionBE([]*mysql.Backend) error { + if r.beforeDecommission != nil { + r.beforeDecommission() + } + r.decommissionCalls++ + return r.decommissionErr +} +func (r *decommissionSQLRecorder) DropBE([]*mysql.Backend) error { + if r.beforeDrop != nil { + r.beforeDrop() + } + r.dropCalls++ + return r.dropErr +} + +func decommissionFixture() (*decommissionAction, *decommissionSnapshot) { + backend := &mysql.Backend{BackendID: "101", Host: "doris-cg-1.example", HeartbeatPort: 9050, Tag: `{"cloud_unique_id":"be-1"}`} + action := &decommissionAction{ComputeGroupID: "cg", FromReplicas: 2, Replicas: 1, CloudUniqueID: "1:instance:operator", InstanceID: "instance", Targets: []decommissionTarget{{Ordinal: 1, Phase: decommissionPrepared}}} + view := map[int32]*mysql.Backend{1: backend} + snapshot := &decommissionSnapshot{Nodes: map[int32]ms.Node{1: {CloudUniqueID: "be-1", Host: backend.Host, HeartbeatPort: 9050, Status: ms.Running}}, Backends: view, FEViews: []map[int32]*mysql.Backend{view}, SyncWait: 20 * time.Second} + return action, snapshot +} + +func setMSState(snapshot *decommissionSnapshot, state ms.NodeStatus) { + node := snapshot.Nodes[1] + node.Status = state + snapshot.Nodes[1] = node +} + +func TestDecommissionPersistsBeforeSQLAndDoesNotResubmitAfterRestart(t *testing.T) { + action, snapshot := decommissionFixture() + var persisted []byte + save := func() error { var err error; persisted, err = json.Marshal(action); return err } + sql := &decommissionSQLRecorder{beforeDecommission: func() { + var saved decommissionAction + if err := json.Unmarshal(persisted, &saved); err != nil { + t.Fatal(err) + } + if saved.Targets[0].Phase != decommissionSubmitted || saved.Targets[0].BackendID != "101" { + t.Fatalf("SQL preceded durable intent: %+v", saved) + } + }} + if done, err := advanceDecommission(action, snapshot, time.Now(), sql, save); done || err != nil { + t.Fatalf("done=%v err=%v", done, err) + } + restarted := new(decommissionAction) + if err := json.Unmarshal(persisted, restarted); err != nil { + t.Fatal(err) + } + if done, err := advanceDecommission(restarted, snapshot, time.Now(), sql, func() error { return nil }); done || err != nil { + t.Fatalf("done=%v err=%v", done, err) + } + if sql.decommissionCalls != 1 || sql.dropCalls != 0 { + t.Fatalf("SQL calls=%+v", sql) + } +} + +func TestDecommissionAdoptsManualSQLAcrossKernelVersions(t *testing.T) { + for _, tt := range []struct { + name string + state ms.NodeStatus + feStarted bool + }{ + {"3.x manual SQL before MS update", ms.Running, true}, + {"early 4.0 manual SQL before MS update", ms.Running, true}, + {"new 4.x manual SQL", ms.Decommissioning, false}, + } { + t.Run(tt.name, func(t *testing.T) { + action, snapshot := decommissionFixture() + setMSState(snapshot, tt.state) + snapshot.Backends[1].SystemDecommissioned = tt.feStarted + sql := &decommissionSQLRecorder{} + done, err := advanceDecommission(action, snapshot, time.Now(), sql, func() error { return nil }) + if err != nil || done || sql.decommissionCalls != 0 || sql.dropCalls != 0 || action.Targets[0].Phase != decommissionSubmitted { + t.Fatalf("done=%v err=%v action=%+v SQL=%+v", done, err, action, sql) + } + }) + } +} + +func TestDecommissionWaitsForMSCompletionAndFESync(t *testing.T) { + action, snapshot := decommissionFixture() + sql := &decommissionSQLRecorder{} + now := time.Now().Truncate(time.Second) + snapshot.Backends[1].SystemDecommissioned = true + snapshot.Backends[1].TabletNum = 0 + if _, err := advanceDecommission(action, snapshot, now, sql, func() error { return nil }); err != nil { + t.Fatal(err) + } + if sql.dropCalls != 0 { + t.Fatal("legacy FE true with zero tablets is not MS completion") + } + setMSState(snapshot, ms.Decommissioned) + // A completed MS state is authoritative even if SHOW BACKENDS still reports tablets. + snapshot.Backends[1].TabletNum = 100 + if _, err := advanceDecommission(action, snapshot, now, sql, func() error { return nil }); err != nil { + t.Fatal(err) + } + for _, elapsed := range []time.Duration{0, 10 * time.Second, 19 * time.Second} { + if _, err := advanceDecommission(action, snapshot, now.Add(elapsed), sql, func() error { return nil }); err != nil { + t.Fatal(err) + } + if sql.dropCalls != 0 { + t.Fatal("DROP sent before FE sync delay") + } + } + stale := *snapshot.Backends[1] + stale.SystemDecommissioned = false + snapshot.FEViews = append(snapshot.FEViews, map[int32]*mysql.Backend{1: &stale}) + if _, err := advanceDecommission(action, snapshot, now.Add(time.Minute), sql, func() error { return nil }); err != nil { + t.Fatal(err) + } + if sql.dropCalls != 0 { + t.Fatal("DROP sent while an FE still has an unsynchronized state") + } + stale.SystemDecommissioned = true + sql.beforeDrop = func() { + if action.Targets[0].Phase != decommissionDropping { + t.Fatal("DROP intent not persisted") + } + } + if _, err := advanceDecommission(action, snapshot, now.Add(time.Minute), sql, func() error { return nil }); err != nil { + t.Fatal(err) + } + if sql.dropCalls != 1 { + t.Fatal("synchronized completed target was not dropped") + } + delete(snapshot.Nodes, 1) + if done, err := advanceDecommission(action, snapshot, now.Add(2*time.Minute), sql, func() error { return nil }); err != nil || done { + t.Fatalf("FE still has target: done=%v err=%v", done, err) + } + delete(snapshot.Backends, 1) + if done, err := advanceDecommission(action, snapshot, now.Add(2*time.Minute), sql, func() error { return nil }); err != nil || done { + t.Fatalf("observer still has target: done=%v err=%v", done, err) + } + delete(snapshot.FEViews[1], 1) + if done, err := advanceDecommission(action, snapshot, now.Add(2*time.Minute), sql, func() error { return nil }); err != nil || done { + t.Fatalf("first missing observation must wait: done=%v err=%v", done, err) + } + if done, err := advanceDecommission(action, snapshot, now.Add(3*time.Minute), sql, func() error { return nil }); err != nil || !done { + t.Fatalf("stable absence: done=%v err=%v", done, err) + } +} + +func TestDecommissionBlocksUnsafeTransitions(t *testing.T) { + tests := []struct { + name string + prepare func(*decommissionAction, *decommissionSnapshot) + sqlErr, saveErr bool + }{ + {name: "unknown MS state", prepare: func(a *decommissionAction, s *decommissionSnapshot) { setMSState(s, "NODE_STATUS_UNKNOWN") }}, + {name: "MS identity replaced", prepare: func(a *decommissionAction, s *decommissionSnapshot) { a.Targets[0].CloudUniqueID = "old" }}, + {name: "FE identity replaced", prepare: func(a *decommissionAction, s *decommissionSnapshot) { a.Targets[0].BackendID = "old" }}, + {name: "FE and MS disagree", prepare: func(a *decommissionAction, s *decommissionSnapshot) { + s.Backends[1].Tag = `{"cloud_unique_id":"other"}` + }}, + {name: "MS completion regressed", prepare: func(a *decommissionAction, s *decommissionSnapshot) { + now := metav1.Now() + a.Targets[0].CompletedAt = &now + }}, + {name: "no FE observations", prepare: func(a *decommissionAction, s *decommissionSnapshot) { s.FEViews = nil }}, + {name: "persistence failed", saveErr: true}, + {name: "SQL submission uncertain", sqlErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + action, snapshot := decommissionFixture() + if tt.prepare != nil { + tt.prepare(action, snapshot) + } + sql := &decommissionSQLRecorder{} + if tt.sqlErr { + sql.decommissionErr = errors.New("connection lost") + } + save := func() error { + if tt.saveErr { + return errors.New("conflict") + } + return nil + } + done, err := advanceDecommission(action, snapshot, time.Now(), sql, save) + if err == nil || done || sql.dropCalls != 0 { + t.Fatalf("unsafe transition: done=%v err=%v SQL=%+v", done, err, sql) + } + if tt.saveErr && sql.decommissionCalls != 0 { + t.Fatal("SQL sent after persistence failure") + } + if tt.sqlErr { + sql.decommissionErr = nil + if _, err := advanceDecommission(action, snapshot, time.Now(), sql, save); err != nil { + t.Fatal(err) + } + if sql.decommissionCalls != 1 { + t.Fatal("uncertain SQL was retried") + } + } + }) + } +} + +func TestDecommissionDropFailureBlocksScaleDown(t *testing.T) { + action, snapshot := decommissionFixture() + setMSState(snapshot, ms.Decommissioned) + now := time.Now() + completedAt := metav1.NewTime(now.Add(-time.Minute)) + action.Targets[0].CompletedAt = &completedAt + action.Targets[0].Phase = decommissionWaitFESync + snapshot.Backends[1].SystemDecommissioned = true + sql := &decommissionSQLRecorder{dropErr: errors.New("DROP failed")} + if done, err := advanceDecommission(action, snapshot, now, sql, func() error { return nil }); err == nil || done { + t.Fatalf("done=%v err=%v", done, err) + } + // A lost DROP response can be recovered by observing absence, without SQL. + delete(snapshot.Nodes, 1) + delete(snapshot.Backends, 1) + if done, err := advanceDecommission(action, snapshot, now, sql, func() error { return nil }); err != nil || done { + t.Fatalf("first absence: done=%v err=%v", done, err) + } + if done, err := advanceDecommission(action, snapshot, now.Add(time.Minute), sql, func() error { return nil }); err != nil || !done || sql.dropCalls != 1 { + t.Fatalf("confirmed absence: done=%v err=%v calls=%d", done, err, sql.dropCalls) + } +} + +func TestDecommissionPartialCompletionDoesNotTouchRetainedNodes(t *testing.T) { + action, snapshot := decommissionFixture() + action.FromReplicas = 3 + action.Targets = append([]decommissionTarget{{Ordinal: 2, Phase: decommissionPrepared}}, action.Targets...) + snapshot.Nodes[0] = ms.Node{CloudUniqueID: "retained", Status: ms.Decommissioned} + snapshot.Nodes[2] = ms.Node{CloudUniqueID: "be-2", Status: ms.Decommissioning} + snapshot.Backends[0] = &mysql.Backend{BackendID: "100", Host: "doris-cg-0", HeartbeatPort: 9050, SystemDecommissioned: true} + sql := &decommissionSQLRecorder{} + if done, err := advanceDecommission(action, snapshot, time.Now(), sql, func() error { return nil }); err != nil || done { + t.Fatalf("done=%v err=%v", done, err) + } + if sql.decommissionCalls != 1 || sql.dropCalls != 0 { + t.Fatalf("SQL=%+v", sql) + } + delete(snapshot.Nodes, 2) + if done, err := advanceDecommission(action, snapshot, time.Now(), sql, func() error { return nil }); err != nil || done { + t.Fatalf("another target is pending: done=%v err=%v", done, err) + } +} + +func TestDecommissionAnnotationSurvivesRestartAndPreservesOtherMetadata(t *testing.T) { + scheme := runtime.NewScheme() + if err := appv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + replicas := int32(2) + sts := &appv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "doris-cg", Namespace: "default", Annotations: map[string]string{"other": "keep"}}, Spec: appv1.StatefulSetSpec{Replicas: &replicas}} + controller := &DisaggregatedComputeGroupsController{} + controller.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(sts).Build() + key := types.NamespacedName{Name: sts.Name, Namespace: sts.Namespace} + if err := controller.K8sclient.Get(context.Background(), key, sts); err != nil { + t.Fatal(err) + } + action, _ := decommissionFixture() + if err := controller.saveDecommissionAction(context.Background(), sts, action); err != nil { + t.Fatal(err) + } + var restored appv1.StatefulSet + if err := controller.K8sclient.Get(context.Background(), key, &restored); err != nil { + t.Fatal(err) + } + var saved decommissionAction + if err := json.Unmarshal([]byte(restored.Annotations[decommissionActionAnnotation]), &saved); err != nil { + t.Fatal(err) + } + if err := saved.validate(1, 2, "cg"); err != nil { + t.Fatal(err) + } + if restored.Annotations["other"] != "keep" || *restored.Spec.Replicas != 2 { + t.Fatalf("unexpected resource mutation: %+v", restored) + } + if err := saved.validate(2, 2, "cg"); err == nil { + t.Fatal("changed target was accepted") + } + if err := saved.validate(1, 2, "other"); err == nil { + t.Fatal("changed compute group was accepted") + } + saved.Targets = nil + if err := saved.validate(1, 2, "cg"); err == nil { + t.Fatal("truncated task was accepted") + } + if err := controller.finishDecommissionAction(context.Background(), &restored, 1); err != nil { + t.Fatal(err) + } + var finished appv1.StatefulSet + if err := controller.K8sclient.Get(context.Background(), key, &finished); err != nil { + t.Fatal(err) + } + if *finished.Spec.Replicas != 1 || finished.Annotations[decommissionActionAnnotation] != "" || finished.Annotations["other"] != "keep" { + t.Fatalf("unexpected completed resource: %+v", finished) + } +} + +func TestDecommissionOrdinalMapping(t *testing.T) { + for _, host := range []string{"doris-cg-12", "doris-cg-12.doris-cg.default.svc.cluster.local", "10.0.0.12"} { + ordinal, err := decommissionOrdinal("doris-cg", host, map[string]int32{"10.0.0.12": 12}) + if err != nil || ordinal != 12 { + t.Fatalf("host=%s ordinal=%d err=%v", host, ordinal, err) + } + } + for _, host := range []string{"other-cg-12.example", "doris-cg-x", "doris-cg--1", "10.0.0.13"} { + if _, err := decommissionOrdinal("doris-cg", host, nil); err == nil { + t.Fatalf("unrecognized host %s accepted", host) + } + } +} + +func TestFrontendSyncIntervalAcrossKernelVersions(t *testing.T) { + for _, tt := range []struct { + name, value string + empty, wantErr bool + want time.Duration + }{ + {name: "3.x constant", empty: true, want: 10 * time.Second}, + {name: "4.x default", value: "10", want: 10 * time.Second}, + {name: "4.x SQL override", value: "45", want: 45 * time.Second}, + {name: "invalid", value: "bad", wantErr: true}, + {name: "nonpositive", value: "0", wantErr: true}, + } { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + rows := sqlmock.NewRows([]string{"Key", "Value", "Type", "IsMutable"}) + if !tt.empty { + rows.AddRow("cloud_cluster_check_interval_second", tt.value, "int", true) + } + mock.ExpectQuery(regexp.QuoteMeta("ADMIN SHOW FRONTEND CONFIG LIKE 'cloud_cluster_check_interval_second'")).WillReturnRows(rows) + interval, err := frontendSyncInterval(&mysql.DB{DB: sqlx.NewDb(db, "mysql")}) + if (err != nil) != tt.wantErr || interval != tt.want { + t.Fatalf("interval=%v err=%v", interval, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } + }) + } +} + +func TestDecommissionConflictDoesNotChangeReplicas(t *testing.T) { + scheme := runtime.NewScheme() + if err := appv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + replicas := int32(2) + sts := &appv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "doris-cg", Namespace: "default"}, Spec: appv1.StatefulSetSpec{Replicas: &replicas}} + controller := &DisaggregatedComputeGroupsController{} + controller.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(sts).Build() + key := types.NamespacedName{Name: sts.Name, Namespace: sts.Namespace} + if err := controller.K8sclient.Get(context.Background(), key, sts); err != nil { + t.Fatal(err) + } + stale := sts.DeepCopy() + action, _ := decommissionFixture() + if err := controller.saveDecommissionAction(context.Background(), sts, action); err != nil { + t.Fatal(err) + } + if err := controller.finishDecommissionAction(context.Background(), stale, 1); err == nil { + t.Fatal("stale replica patch succeeded") + } + var actual appv1.StatefulSet + if err := controller.K8sclient.Get(context.Background(), key, &actual); err != nil { + t.Fatal(err) + } + if *actual.Spec.Replicas != 2 || actual.Annotations[decommissionActionAnnotation] == "" { + t.Fatal("conflicting patch changed replica count or lost task") + } +} + +func TestReconcileStatefulsetBlocksDisablingActiveDecommission(t *testing.T) { + scheme := runtime.NewScheme() + if err := appv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + ddc := newTestDDC() + cg := newTestCG("cg1") + ddc.Status.ComputeGroupStatuses = []dv1.ComputeGroupStatus{{UniqueId: cg.UniqueId, ComputeGroupId: "cg"}} + existing := newTestStatefulSet(ddc.Namespace, ddc.GetCGStatefulsetName(cg), "100Gi") + desired := existing.DeepCopy() + current, target := int32(2), int32(1) + existing.Spec.Replicas = ¤t + desired.Spec.Replicas = &target + existing.Annotations = map[string]string{decommissionActionAnnotation: `{"active":true}`} + controller := &DisaggregatedComputeGroupsController{} + controller.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(existing).Build() + event, err := controller.reconcileStatefulset(context.Background(), desired, ddc, cg) + if err == nil || event == nil { + t.Fatal("active task bypassed by disabling decommission") + } + var actual appv1.StatefulSet + if err := controller.K8sclient.Get(context.Background(), types.NamespacedName{Namespace: existing.Namespace, Name: existing.Name}, &actual); err != nil { + t.Fatal(err) + } + if *actual.Spec.Replicas != 2 || actual.Annotations[decommissionActionAnnotation] == "" { + t.Fatal("blocked task changed StatefulSet") + } +} + +func TestDecommissionTransientMSRemovalDoesNotShrinkOrResubmit(t *testing.T) { + action, snapshot := decommissionFixture() + action.Targets[0].Phase = decommissionSubmitted + now := time.Now() + node, backend := snapshot.Nodes[1], snapshot.Backends[1] + delete(snapshot.Nodes, 1) + delete(snapshot.Backends, 1) + sql := &decommissionSQLRecorder{} + if done, err := advanceDecommission(action, snapshot, now, sql, func() error { return nil }); err != nil || done { + t.Fatalf("transient absence: done=%v err=%v", done, err) + } + snapshot.Nodes[1] = node + snapshot.Backends[1] = backend + if done, err := advanceDecommission(action, snapshot, now.Add(10*time.Second), sql, func() error { return nil }); err != nil || done { + t.Fatalf("reappeared: done=%v err=%v", done, err) + } + if action.Targets[0].AbsentSince != nil || sql.decommissionCalls != 0 || sql.dropCalls != 0 { + t.Fatal("transient MS removal advanced the task or repeated SQL") + } +} 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 10be02f3..a20d9d29 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/prepare_modify.go @@ -23,6 +23,7 @@ import ( "fmt" "strconv" "strings" + "time" dv1 "github.com/apache/doris-operator/api/disaggregated/v1" "github.com/apache/doris-operator/pkg/common/utils/k8s" @@ -45,6 +46,9 @@ func (dcgs *DisaggregatedComputeGroupsController) preApplyStatefulSet(ctx contex switch optType { case "scaleDown": + if cluster.Spec.EnableDecommission { + return nil // The persistent MS-driven state machine owns this operation. + } err := dcgs.scaleOut(ctx, cgStatus, cluster, cg) if err != nil { return err @@ -68,45 +72,12 @@ func (dcgs *DisaggregatedComputeGroupsController) scaleOut(ctx context.Context, cgKeepAmount := *cg.Replicas cgid := cgStatus.ComputeGroupId - if cluster.Spec.EnableDecommission { - if err := dcgs.scaledOutBENodesByDecommission(cluster, cgStatus, sqlClient, cgid, cgKeepAmount); err != nil { - return err - } - } else { // not decommission , drop node - if err := dcgs.scaledOutBENodesByDrop(sqlClient, cgid, cgKeepAmount); err != nil { - cgStatus.Phase = dv1.ScaleDownFailed - klog.Errorf("ScaleOut scaledOutBENodesByDrop ddcName:%s, namespace:%s, computeGroupName:%s, drop nodes failed:%s ", cluster.Name, cluster.Namespace, cgid, err.Error()) - return err - } - cgStatus.Phase = dv1.Scaling - } - // return nil will apply sts - return nil -} - -func (dcgs *DisaggregatedComputeGroupsController) scaledOutBENodesByDecommission(cluster *dv1.DorisDisaggregatedCluster, cgStatus *dv1.ComputeGroupStatus, sqlClient *mysql.DB, cgid string, cgKeepAmount int32) error { - decommissionPhase, err := dcgs.decommissionProgressCheck(sqlClient, cgid, cgKeepAmount) - if err != nil { + if err := dcgs.scaledOutBENodesByDrop(sqlClient, cgid, cgKeepAmount); err != nil { + cgStatus.Phase = dv1.ScaleDownFailed return err } - switch decommissionPhase { - case resource.DecommissionAcceptable: - err = dcgs.decommissionBENodes(sqlClient, cgid, cgKeepAmount) - if err != nil { - cgStatus.Phase = dv1.ScaleDownFailed - klog.Errorf("scaledOutBENodesByDecommission ddcName:%s, namespace:%s, computeGroupId:%s , Decommission failed, err:%s ", cluster.Name, cluster.Namespace, cgid, err.Error()) - return err - } - cgStatus.Phase = dv1.Decommissioning - return nil - case resource.Decommissioning, resource.DecommissionPhaseUnknown: - cgStatus.Phase = dv1.Decommissioning - klog.Infof("scaledOutBENodesByDecommission ddcName:%s, namespace:%s, computeGroupId:%s, Decommission in progress", cluster.Name, cluster.Namespace, cgid) - return nil - case resource.Decommissioned: - dcgs.scaledOutBENodesByDrop(sqlClient, cgid, cgKeepAmount) - } cgStatus.Phase = dv1.Scaling + // return nil will apply sts return nil } @@ -168,28 +139,6 @@ func dropBackendEnsuringAbsent(sqlClient *mysql.DB, target *mysql.Backend) error return nil } -func (dcgs *DisaggregatedComputeGroupsController) decommissionBENodes( - masterDBClient *mysql.DB, - cgName string, - cgKeepAmount int32) error { - - dropNodes, err := getScaledOutBENode(masterDBClient, cgName, cgKeepAmount) - if err != nil { - klog.Errorf("decommissionBENodes getScaledOutBENode cgName %s failed, err:%s ", cgName, err.Error()) - return err - } - - if len(dropNodes) == 0 { - return nil - } - err = masterDBClient.DecommissionBE(dropNodes) - if err != nil { - klog.Errorf("decommissionBENodes cgName %s DropBENodes failed, err:%s ", cgName, err.Error()) - return err - } - return nil -} - func (dcgs *DisaggregatedComputeGroupsController) getMasterSqlClient(ctx context.Context, cluster *dv1.DorisDisaggregatedCluster) (*mysql.DB, error) { // get user and password adminUserName, password := dcgs.GetManagementAdminUserAndPWD(ctx, cluster) @@ -209,6 +158,9 @@ func (dcgs *DisaggregatedComputeGroupsController) getMasterSqlClient(ctx context Port: strconv.FormatInt(int64(queryPort), 10), Database: "mysql", } + if cluster.Spec.EnableDecommission { + dbConf.Timeout = 10 * time.Second + } tlsConfig, secretName := dcgs.DisaggregatedSubDefaultController.FindSecretTLSConfig(confMap, cluster) secret, _ := k8s.GetSecret(context.Background(), dcgs.K8sclient, cluster.Namespace, secretName) @@ -221,17 +173,6 @@ func (dcgs *DisaggregatedComputeGroupsController) getMasterSqlClient(ctx context return masterDBClient, nil } -// isDecommissionProgressFinished check decommission status -func (dcgs *DisaggregatedComputeGroupsController) decommissionProgressCheck(masterDBClient *mysql.DB, cgid string, cgKeepAmount int32) (resource.DecommissionPhase, error) { - allBackends, err := masterDBClient.GetBackendsByComputeGroupId(cgid) - if err != nil { - klog.Errorf("decommissionProgressCheck failed, cgid %s ShowBackends err:%s", cgid, err.Error()) - return resource.DecommissionPhaseUnknown, err - } - dts := resource.ConstructDecommissionTaskStatus(allBackends, cgKeepAmount) - return dts.GetDecommissionPhase(), nil -} - func getScaledOutBENode( masterDBClient *mysql.DB, cgid string,