[fix] Use MS state for disaggregated BE decommission

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.

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 配置的权限。

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
+}