Merge pull request #522 from qube-rt/fix/cn-clearresources-nil-cnspec fix(cn-controller): guard dereference while trying to clear resources
diff --git a/README-CN.md b/README-CN.md index 6ad1dbe..7e3a990 100644 --- a/README-CN.md +++ b/README-CN.md
@@ -52,7 +52,11 @@ - [config fe specification](https://doris.apache.org/zh-CN/docs/3.0/install/deploy-on-kubernetes/separating-storage-compute/config-fe) - [config compute specification](http://doris.apache.org/zh-CN/docs/3.0/install/deploy-on-kubernetes/separating-storage-compute/config-cg) +## 文档 + +- [动态配置持久化与旧集群迁移](doc/operation/runtime_config_persistence.md) + ## 开源支持 - [CN Forum](https://ask.selectdb.com/) - [github issues](https://github.com/apache/doris-operator/issues) -- [Slack](https://apachedoriscommunity.slack.com/archives/C02T886T5AR) \ No newline at end of file +- [Slack](https://apachedoriscommunity.slack.com/archives/C02T886T5AR)
diff --git a/README.md b/README.md index f54213a..7b949f5 100644 --- a/README.md +++ b/README.md
@@ -53,7 +53,11 @@ - [config fe specification](https://doris.apache.org/docs/3.0/install/deploy-on-kubernetes/separating-storage-compute/config-fe) - [config compute specification](http://doris.apache.org/docs/3.0/install/deploy-on-kubernetes/separating-storage-compute/config-cg) +## Documentation + +- [Runtime configuration persistence and migration](doc/operation/runtime_config_persistence.md) + ## Open source support - [CN Forum](https://ask.selectdb.com/) - [github issues](https://github.com/apache/doris-operator/issues) -- [Slack](https://apachedoriscommunity.slack.com/archives/C02T886T5AR) \ No newline at end of file +- [Slack](https://apachedoriscommunity.slack.com/archives/C02T886T5AR)
diff --git a/api/disaggregated/v1/types.go b/api/disaggregated/v1/types.go index 002968f..99c5f5a 100644 --- a/api/disaggregated/v1/types.go +++ b/api/disaggregated/v1/types.go
@@ -23,6 +23,11 @@ ) type DorisDisaggregatedClusterSpec struct { + // RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + // When omitted, new clusters enable it where safe storage exists; existing + // clusters keep their previous behavior until explicitly migrated. + RuntimeConfigPersistence *RuntimeConfigPersistenceSpec `json:"runtimeConfigPersistence,omitempty"` + //VaultConfigmap specify the configmap that have configuration of file object information. example S3. //configmap have to config, please reference the doc. //InstanceConfigMap string `json:"instanceConfigMap,omitempty"` @@ -594,3 +599,11 @@ func init() { SchemeBuilder.Register(&DorisDisaggregatedCluster{}, &DorisDisaggregatedClusterList{}) } + +// RuntimeConfigPersistenceSpec controls persistence of FE, BE and MS mutable configuration. +type RuntimeConfigPersistenceSpec struct { + // Enabled overrides the operator's recorded default. Enabling an existing + // cluster requires the runtime configuration migration tool. Disabling an + // already enabled cluster is not supported because it would lose overrides. + Enabled bool `json:"enabled"` +}
diff --git a/api/disaggregated/v1/zz_generated.deepcopy.go b/api/disaggregated/v1/zz_generated.deepcopy.go index 02162b7..bd829f4 100644 --- a/api/disaggregated/v1/zz_generated.deepcopy.go +++ b/api/disaggregated/v1/zz_generated.deepcopy.go
@@ -289,6 +289,11 @@ // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *DorisDisaggregatedClusterSpec) DeepCopyInto(out *DorisDisaggregatedClusterSpec) { *out = *in + if in.RuntimeConfigPersistence != nil { + in, out := &in.RuntimeConfigPersistence, &out.RuntimeConfigPersistence + *out = new(RuntimeConfigPersistenceSpec) + **out = **in + } in.MetaService.DeepCopyInto(&out.MetaService) in.FeSpec.DeepCopyInto(&out.FeSpec) if in.ComputeGroups != nil { @@ -567,6 +572,21 @@ } // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RuntimeConfigPersistenceSpec) DeepCopyInto(out *RuntimeConfigPersistenceSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RuntimeConfigPersistenceSpec. +func (in *RuntimeConfigPersistenceSpec) DeepCopy() *RuntimeConfigPersistenceSpec { + if in == nil { + return nil + } + out := new(RuntimeConfigPersistenceSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Secret) DeepCopyInto(out *Secret) { *out = *in }
diff --git a/api/doris/v1/types.go b/api/doris/v1/types.go index f533e3f..cb69119 100644 --- a/api/doris/v1/types.go +++ b/api/doris/v1/types.go
@@ -30,6 +30,11 @@ // DorisClusterSpec defines the desired state of DorisCluster type DorisClusterSpec struct { + // RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + // When omitted, new clusters enable it where safe storage exists; existing + // clusters keep their previous behavior until explicitly migrated. + RuntimeConfigPersistence *RuntimeConfigPersistenceSpec `json:"runtimeConfigPersistence,omitempty"` + //defines the fe cluster state that will be created by operator. FeSpec *FeSpec `json:"feSpec,omitempty"` @@ -546,3 +551,11 @@ func init() { SchemeBuilder.Register(&DorisCluster{}, &DorisClusterList{}) } + +// RuntimeConfigPersistenceSpec controls persistence of FE, BE and MS mutable configuration. +type RuntimeConfigPersistenceSpec struct { + // Enabled overrides the operator's recorded default. Enabling an existing + // cluster requires the runtime configuration migration tool. Disabling an + // already enabled cluster is not supported because it would lose overrides. + Enabled bool `json:"enabled"` +}
diff --git a/api/doris/v1/zz_generated.deepcopy.go b/api/doris/v1/zz_generated.deepcopy.go index fdcc5be..0436d77 100644 --- a/api/doris/v1/zz_generated.deepcopy.go +++ b/api/doris/v1/zz_generated.deepcopy.go
@@ -417,6 +417,11 @@ // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *DorisClusterSpec) DeepCopyInto(out *DorisClusterSpec) { *out = *in + if in.RuntimeConfigPersistence != nil { + in, out := &in.RuntimeConfigPersistence, &out.RuntimeConfigPersistence + *out = new(RuntimeConfigPersistenceSpec) + **out = **in + } if in.FeSpec != nil { in, out := &in.FeSpec, &out.FeSpec *out = new(FeSpec) @@ -939,6 +944,21 @@ } // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RuntimeConfigPersistenceSpec) DeepCopyInto(out *RuntimeConfigPersistenceSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RuntimeConfigPersistenceSpec. +func (in *RuntimeConfigPersistenceSpec) DeepCopy() *RuntimeConfigPersistenceSpec { + if in == nil { + return nil + } + out := new(RuntimeConfigPersistenceSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Secret) DeepCopyInto(out *Secret) { *out = *in }
diff --git a/config/crd/bases/crds.yaml b/config/crd/bases/crds.yaml index 86cc968..bdbd96f 100644 --- a/config/crd/bases/crds.yaml +++ b/config/crd/bases/crds.yaml
@@ -9239,6 +9239,21 @@ description: Krb5ConfigMap is the name of configmap within 'krb5.conf' type: string type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object sharedPersistentVolumeClaims: description: SharedPersistentVolumeClaims used to configure the shared pvc that needs to be mounted on the pod @@ -16544,6 +16559,21 @@ type: object type: array type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object type: object status: properties:
diff --git a/config/crd/bases/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml b/config/crd/bases/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml index 0547f69..ad81c18 100644 --- a/config/crd/bases/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml +++ b/config/crd/bases/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml
@@ -7047,6 +7047,21 @@ type: object type: array type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object type: object status: properties:
diff --git a/config/crd/bases/doris.apache.com_dorisclusters.yaml b/config/crd/bases/doris.apache.com_dorisclusters.yaml index eb9b96e..c02e110 100644 --- a/config/crd/bases/doris.apache.com_dorisclusters.yaml +++ b/config/crd/bases/doris.apache.com_dorisclusters.yaml
@@ -9239,6 +9239,21 @@ description: Krb5ConfigMap is the name of configmap within 'krb5.conf' type: string type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object sharedPersistentVolumeClaims: description: SharedPersistentVolumeClaims used to configure the shared pvc that needs to be mounted on the pod
diff --git a/config/crd/bases/doris.selectdb.com_dorisclusters.yaml b/config/crd/bases/doris.selectdb.com_dorisclusters.yaml index eb9b96e..c02e110 100644 --- a/config/crd/bases/doris.selectdb.com_dorisclusters.yaml +++ b/config/crd/bases/doris.selectdb.com_dorisclusters.yaml
@@ -9239,6 +9239,21 @@ description: Krb5ConfigMap is the name of configmap within 'krb5.conf' type: string type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object sharedPersistentVolumeClaims: description: SharedPersistentVolumeClaims used to configure the shared pvc that needs to be mounted on the pod
diff --git a/doc/operation/runtime_config_persistence.md b/doc/operation/runtime_config_persistence.md new file mode 100644 index 0000000..51def1e --- /dev/null +++ b/doc/operation/runtime_config_persistence.md
@@ -0,0 +1,139 @@ +# Runtime configuration persistence + +FE and BE store persisted mutable settings in `fe_custom.conf` and +`be_custom.conf`. Cloud BE also saves configuration received in FE heartbeats. +MS normally persists mutable settings into its startup configuration file, which +its container entrypoint can replace from a ConfigMap on restart. + +The operator can preserve these settings separately from user ConfigMaps for +DorisCluster FE/BE and DorisDisaggregatedCluster FE/BE/MS. It does not change user +ConfigMaps, mount an empty volume over the image's `conf` directory, create new +PVCs, or change StatefulSet PVC templates. + +## Defaults and compatibility + +On its first reconciliation, the operator records a cluster-level default in +`doris.apache.org/runtime-config-policy`: + +- New clusters use automatic persistence where a safe existing PVC is available. +- Existing clusters retain their previous Pod templates. Upgrading the operator + does not migrate or restart them. +- The recorded choice survives operator restarts and individual StatefulSet + deletion/recreation. Do not manually remove the policy annotation. + +A new-cluster component without a suitable PVC keeps its existing behavior and +emits `RuntimeConfigStorageUnavailable`. Its StatefulSet is marked unavailable +for automatic persistence; adding storage later does not silently migrate it. +Stateless clusters are **not** made durable by this feature. + +An explicit request is strict: + +```yaml +spec: + runtimeConfigPersistence: + enabled: true +``` + +Missing safe storage or an incomplete existing-cluster migration blocks this +request. Set `enabled: false` before initial deployment to opt out. Turning off +an already activated policy is rejected: doing so could discard configuration +that overrides the ConfigMap. No automatic reverse migration is provided. + +## Storage and configuration layout + +The operator selects a writable, existing per-Pod PVC mount in this order: + +1. A log PVC. +2. An FE metadata PVC (`meta_dir` / `doris-meta`). +3. A BE data PVC (`storage_root_path` / `storage`). + +A `file_cache_path` root or overlapping directory is excluded because cache +maintenance can delete the entire root. EmptyDir, read-only mounts and mounts +using an existing `subPathExpr` are excluded. A cache-only cloud BE needs a safe +persistent log/configuration volume before explicit enablement. Directory +selection is recorded on the StatefulSet and is not silently switched later. + +A directory `.doris-runtime-config/<pod-name>` inside the selected mount is +exposed at `/var/lib/doris/runtime-config`. The Pod name separates overrides even +when an existing PVC is shared. Existing `subPath` mounts retain their prefix. + +For DDC compute-group scale-down, the operator retains the selected runtime +configuration PVC so scaling back to the same ordinal can recover its overrides +and migration marker. Other PVCs, including file cache PVCs, follow the existing +cleanup policy. Retained configuration claims continue to consume storage while +the ordinal is absent. This does not change deletion of the cluster or manual +PVC deletion, and does not initialize cloud identity for entirely new nodes. + +An init container uses the **same image and security context** as the main +container to create the directory and prepare a writable copy of the base +ConfigMap configuration. When no base ConfigMap exists, it uses the image's +base configuration. It injects: + +| Component | Injected setting | Persisted file | +| --- | --- | --- | +| FE | `custom_config_dir=/var/lib/doris/runtime-config` | `fe_custom.conf` | +| BE | `custom_config_dir=/var/lib/doris/runtime-config` | `be_custom.conf` | +| MS | `custom_conf_path=/var/lib/doris/runtime-config/ms_custom.conf` | `ms_custom.conf` | + +The main container retains its normal entrypoint. Runtime files are created only +when absent; image/configuration updates never truncate them. Directory mounts +allow the kernel to create temporary files and replace files with `rename`. + +Explicit custom paths already backed by writable PVCs remain user-managed. An +explicit nondefault path without durable backing is rejected. ConfigMap-backed +initial `*_custom.conf` files seed the persistent file once and are excluded +from the prepared ConfigMap directory to prevent read-only symlinks. + +Images must contain Bash, cp, awk and the existing entrypoint, and support the +corresponding kernel configuration setting. Nonroot deployments must supply +working PVC/emptyDir permissions (for example the appropriate Pod `fsGroup`). +This feature does not change container users or elevate privileges. + +## Migrating an existing cluster + +1. Upgrade CRDs and the operator first. Leave the persistence spec unset. +2. Stop dynamic configuration updates, scaling, image updates, and concurrent + automation for the migration window. All existing Pods must be Ready. +3. Back up the live overrides **before** replacing any Pod: + + ```sh + python3 tools/runtime-config-migrate.py \ + --kind ddc --namespace example --name doris \ + --backup-dir ./runtime-backup + # Use --kind dcr for DorisCluster; --context selects a kubectl context. + ``` + +4. Review the private backup directory. Files are mode 0600 and the directory is + mode 0700; configuration values are not printed or stored in annotations. + In particular, MS candidates contain effective settings that differ from the + currently mounted ConfigMap. The tool cannot reconstruct which historical + values were deliberately changed: review MS candidates and remove obsolete + overrides. FDB endpoints and configuration-path settings are excluded. +5. Apply the reviewed backup: + + ```sh + python3 tools/runtime-config-migrate.py \ + --kind ddc --namespace example --name doris \ + --backup-dir ./runtime-backup --apply + ``` + +The tool checks cluster, StatefulSet and Pod identities, replicas, generations, +and unchanged live source configuration. It copies each backup through the old +Pod's existing PVC mount, verifies the copy, and only then patches the migration +annotation and `enabled: true`. It never silently replaces a different existing +backup. The operator validates the approval before changing workloads. Migrated +ordinals require their on-disk migration marker; newly added ordinals initialize +normally. Follow the operator's normal rolling update and verify loaded settings +and a subsequent Pod replacement before reopening configuration updates. + +If there is no suitable PVC, an explicit custom path already exists, an MS base +ConfigMap is unavailable, or membership changes, the tool stops. Such clusters +need an individually planned storage/configuration migration. The tool does not +recreate StatefulSets or recover overrides after an old Pod is already lost. +Keep local backups until validation is complete. + +## Scope + +Persistence closes the loss of mutable configuration during Pod replacement. +A new cloud BE still needs its first FE heartbeat. Kernel query admission and FE +scheduling must independently prevent queries before required initialization.
diff --git a/doc/operation/scale_up_down.md b/doc/operation/scale_up_down.md index c510530..70b2052 100644 --- a/doc/operation/scale_up_down.md +++ b/doc/operation/scale_up_down.md
@@ -47,3 +47,32 @@ 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 863b316..4ea9376 100644 --- a/doc/operation/scale_up_down_cn.md +++ b/doc/operation/scale_up_down_cn.md
@@ -49,3 +49,24 @@ 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/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml b/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml index 0547f69..ad81c18 100644 --- a/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml +++ b/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml
@@ -7047,6 +7047,21 @@ type: object type: array type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object type: object status: properties:
diff --git a/helm-charts/doris-operator/crds/doris.apache.com_dorisclusters.yaml b/helm-charts/doris-operator/crds/doris.apache.com_dorisclusters.yaml index eb9b96e..c02e110 100644 --- a/helm-charts/doris-operator/crds/doris.apache.com_dorisclusters.yaml +++ b/helm-charts/doris-operator/crds/doris.apache.com_dorisclusters.yaml
@@ -9239,6 +9239,21 @@ description: Krb5ConfigMap is the name of configmap within 'krb5.conf' type: string type: object + runtimeConfigPersistence: + description: |- + RuntimeConfigPersistence preserves mutable configuration on an existing PVC. + When omitted, new clusters enable it where safe storage exists; existing + clusters keep their previous behavior until explicitly migrated. + properties: + enabled: + description: |- + Enabled overrides the operator's recorded default. Enabling an existing + cluster requires the runtime configuration migration tool. Disabling an + already enabled cluster is not supported because it would lose overrides. + type: boolean + required: + - enabled + type: object sharedPersistentVolumeClaims: description: SharedPersistentVolumeClaims used to configure the shared pvc that needs to be mounted on the pod
diff --git a/pkg/common/utils/metaservice/client.go b/pkg/common/utils/metaservice/client.go new file mode 100644 index 0000000..b2dbf2e --- /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 0000000..d97e6ce --- /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 aabe3ce..4dd89b9 100644 --- a/pkg/common/utils/mysql/mysql.go +++ b/pkg/common/utils/mysql/mysql.go
@@ -24,6 +24,8 @@ "encoding/json" "errors" "fmt" + "net/url" + "time" "github.com/go-sql-driver/mysql" _ "github.com/go-sql-driver/mysql" @@ -42,6 +44,8 @@ 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 @@ 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 @@ }); 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 @@ } 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 @@ 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 cb7d8c2..ad68ab4 100644 --- a/pkg/common/utils/mysql/mysql_test.go +++ b/pkg/common/utils/mysql/mysql_test.go
@@ -22,9 +22,11 @@ "database/sql/driver" "errors" "fmt" + "net" "regexp" "strconv" "testing" + "time" "github.com/DATA-DOG/go-sqlmock" "github.com/jmoiron/sqlx" @@ -288,3 +290,36 @@ 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/common/utils/runtimeconfig/pod.go b/pkg/common/utils/runtimeconfig/pod.go new file mode 100644 index 0000000..c24e5cf --- /dev/null +++ b/pkg/common/utils/runtimeconfig/pod.go
@@ -0,0 +1,376 @@ +// 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 runtimeconfig + +import ( + "context" + _ "embed" + "encoding/json" + "fmt" + "path" + "sort" + "strings" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + RuntimeDir = "/var/lib/doris/runtime-config" + StoreDir = ".doris-runtime-config" + configVolume = "runtime-config-prepared" + initName = "prepare-runtime-config" +) + +//go:embed prepare.sh +var prepareScript string + +type storage struct { + Volume string `json:"volume"` + Claim string `json:"claim"` + MountPath string `json:"mountPath"` + SubPath string `json:"subPath,omitempty"` +} + +func component(name string) string { + switch strings.ToLower(name) { + case "fe": + return "fe" + case "be", "compute": + return "be" + case "metaservice": + return "ms" + } + return "" +} + +func supportedContainer(containers []corev1.Container) bool { + for _, c := range containers { + if component(c.Name) != "" { + return true + } + } + return false +} + +// Prepare only changes the Pod template, never the user's ConfigMap or PVC +// templates. Auto mode leaves workloads without safe persistent storage alone. +func Prepare(ctx context.Context, c client.Client, recorder record.EventRecorder, owner client.Object, st *appsv1.StatefulSet, config map[string]interface{}) error { + policy := owner.GetAnnotations()[PolicyAnnotation] + if policy != Auto && policy != Enabled { + return nil + } + var existing appsv1.StatefulSet + err := c.Get(ctx, types.NamespacedName{Namespace: st.Namespace, Name: st.Name}, &existing) + if err != nil && !apierrors.IsNotFound(err) { + return err + } + recorded := existing.Annotations[StorageAnnotation] + // An automatically skipped workload must not be silently migrated later. + if policy == Auto && err == nil && (recorded == "" || recorded == "unavailable") { + if recorded != "" { + setStorage(st, recorded) + } + return nil + } + for i := range st.Spec.Template.Spec.Containers { + main := &st.Spec.Template.Spec.Containers[i] + kind := component(main.Name) + if kind == "" { + continue + } + key := "custom_config_dir" + if kind == "ms" { + key = "custom_conf_path" + } + configured, _ := config[key].(string) + if strings.TrimSpace(configured) != "" { + // Explicit paths belong to the user. Verify durable backing rather than + // replacing paths, creating directories, or changing their loading rules. + target := strings.TrimSpace(configured) + for _, m := range main.VolumeMounts { + if persistent(st, m.Name) && !m.ReadOnly && within(target, m.MountPath) { + setStorage(st, "external") + return nil + } + } + home := "/opt/apache-doris/" + kind + defaultPath := home + "/conf" + if kind == "ms" { + defaultPath += "/doris_cloud.conf" + } + expanded := strings.ReplaceAll(target, "${DORIS_HOME}", home) + if path.Clean(expanded) != defaultPath { + return fmt.Errorf("%s: explicit %s=%s is not backed by a writable PVC", st.Name, key, target) + } + } + if recorded == "external" { + return fmt.Errorf("%s: removing a user-managed runtime configuration path requires migration", st.Name) + } + chosen, ok := selectStorage(st, main, kind, config, recorded) + if !ok { + if recorded != "" && recorded != "unavailable" { + return fmt.Errorf("%s: previously selected runtime configuration PVC is missing or changed; migrate before changing storage", st.Name) + } + msg := fmt.Sprintf("%s: runtime configuration requires a writable log PVC, FE metadata PVC, or DCR BE storage PVC; cache roots and emptyDir are not safe", st.Name) + if policy == Enabled { + return fmt.Errorf("%s", msg) + } + setStorage(st, "unavailable") + if recorder != nil { + recorder.Event(owner, corev1.EventTypeWarning, "RuntimeConfigStorageUnavailable", msg) + } + return nil + } + if recorded != "" && recorded != "unavailable" && recorded != "external" { + var old storage + if err := json.Unmarshal([]byte(recorded), &old); err != nil || old != chosen { + return fmt.Errorf("%s: runtime configuration storage changed; migrate it before changing volume mounts", st.Name) + } + } + migratedReplicas := int32(0) + var approval migration + if policy == Enabled && json.Unmarshal([]byte(owner.GetAnnotations()[MigrationAnnotation]), &approval) == nil { + migratedReplicas = approval.Workloads[st.Name].Replicas + } + if err := preparePod(st, main, kind, chosen, migratedReplicas); err != nil { + return err + } + raw, _ := json.Marshal(chosen) + setStorage(st, string(raw)) + return nil + } + return nil +} + +func setStorage(st *appsv1.StatefulSet, value string) { + if st.Annotations == nil { + st.Annotations = map[string]string{} + } + st.Annotations[StorageAnnotation] = value +} + +// RetainPVCOnScaleDown preserves the selected configuration store when an +// ordinal is removed. Other claims, including file cache claims, remain subject +// to normal cleanup. Explicit user-managed paths keep their existing policy. +func RetainPVCOnScaleDown(st *appsv1.StatefulSet, pvcName string) (bool, error) { + recorded := st.Annotations[StorageAnnotation] + if recorded == "" || recorded == "unavailable" || recorded == "external" { + return false, nil + } + var selected storage + if err := json.Unmarshal([]byte(recorded), &selected); err != nil || selected.Volume == "" || selected.Claim == "" { + return false, fmt.Errorf("%s: invalid runtime configuration storage; refusing PVC cleanup", st.Name) + } + if claimIdentity(st, selected.Volume) != selected.Claim { + return false, fmt.Errorf("%s: runtime configuration storage changed; refusing PVC cleanup", st.Name) + } + if selected.Claim == "template:"+selected.Volume { + prefix := selected.Volume + "-" + st.Name + "-" + return strings.HasPrefix(pvcName, prefix), nil + } + return selected.Claim == "claim:"+pvcName, nil +} + +func within(target, root string) bool { + return path.IsAbs(target) && (path.Clean(target) == path.Clean(root) || strings.HasPrefix(path.Clean(target), path.Clean(root)+"/")) +} + +func persistent(st *appsv1.StatefulSet, name string) bool { + for _, v := range st.Spec.Template.Spec.Volumes { + if v.Name == name && v.PersistentVolumeClaim != nil && !v.PersistentVolumeClaim.ReadOnly { + return true + } + } + for _, v := range st.Spec.VolumeClaimTemplates { + if v.Name == name { + return true + } + } + return false +} + +func selectStorage(st *appsv1.StatefulSet, main *corev1.Container, kind string, config map[string]interface{}, recorded string) (storage, bool) { + type candidate struct { + storage + rank int + } + var candidates []candidate + var old storage + _ = json.Unmarshal([]byte(recorded), &old) + for _, m := range main.VolumeMounts { + if !persistent(st, m.Name) || m.ReadOnly || m.SubPathExpr != "" { + continue + } + s := storage{Volume: m.Name, Claim: claimIdentity(st, m.Name), MountPath: m.MountPath, SubPath: m.SubPath} + if old.Volume != "" { + if old == s && !cacheRoot(config, m.MountPath) { + return s, true + } + continue + } + rank := 0 + if path.Base(m.MountPath) == "log" || equalsConfigPath(config, "log_dir", m.MountPath) || equalsConfigPath(config, "LOG_DIR", m.MountPath) { + rank = 1 + } + if kind == "fe" && (path.Base(m.MountPath) == "doris-meta" || equalsConfigPath(config, "meta_dir", m.MountPath)) { + rank = 2 + } + if strings.EqualFold(main.Name, "be") && (path.Base(m.MountPath) == "storage" || storageRoot(config, m.MountPath)) { + rank = 3 + } + // file_cache_path roots may be recursively removed, including custom directories. + if cacheRoot(config, m.MountPath) { + rank = 0 + } + if rank != 0 { + candidates = append(candidates, candidate{s, rank}) + } + } + sort.Slice(candidates, func(i, j int) bool { + if candidates[i].rank != candidates[j].rank { + return candidates[i].rank < candidates[j].rank + } + return candidates[i].MountPath < candidates[j].MountPath + }) + if len(candidates) == 0 { + return storage{}, false + } + return candidates[0].storage, true +} + +func equalsConfigPath(config map[string]interface{}, key, p string) bool { + v, _ := config[key].(string) + return v != "" && path.Clean(v) == path.Clean(p) +} +func storageRoot(config map[string]interface{}, p string) bool { + v, _ := config["storage_root_path"].(string) + for _, root := range strings.Split(v, ";") { + if path.Clean(strings.Split(strings.TrimSpace(root), ",")[0]) == path.Clean(p) { + return true + } + } + return false +} +func cacheRoot(config map[string]interface{}, p string) bool { + v, _ := config["file_cache_path"].(string) + var roots []struct { + Path string `json:"path"` + } + if json.Unmarshal([]byte(v), &roots) == nil { + for _, r := range roots { + r.Path = strings.ReplaceAll(r.Path, "${DORIS_HOME}", "/opt/apache-doris/be") + r.Path = strings.ReplaceAll(r.Path, "$DORIS_HOME", "/opt/apache-doris/be") + if within(p, r.Path) || within(r.Path, p) { + return true + } + } + } + return false +} + +func preparePod(st *appsv1.StatefulSet, main *corev1.Container, kind string, s storage, migratedReplicas int32) error { + pod := &st.Spec.Template.Spec + for _, c := range pod.InitContainers { + if c.Name == initName { + return fmt.Errorf("%s: init container name %s is reserved", st.Name, initName) + } + } + for _, m := range main.VolumeMounts { + if path.Clean(m.MountPath) == RuntimeDir { + return fmt.Errorf("%s: mount path %s is reserved; configure an explicit custom path to manage it yourself", st.Name, RuntimeDir) + } + } + home := "/opt/apache-doris/" + kind + base := kind + ".conf" + if kind == "ms" { + base = "doris_cloud.conf" + } + source := "/etc/doris" + envKey := "CONFIGMAP_MOUNT_PATH" + if kind == "ms" { + envKey = "CONFIGMAP_PATH" + } + for _, e := range main.Env { + if e.Name == envKey && e.Value != "" { + source = e.Value + } + } + for _, v := range pod.Volumes { + if v.Name == configVolume { + return fmt.Errorf("%s: volume name %s is reserved", st.Name, configVolume) + } + } + init := corev1.Container{ + Name: initName, Image: main.Image, ImagePullPolicy: main.ImagePullPolicy, + Command: []string{"/bin/bash", "-ec", prepareScript}, + Env: []corev1.EnvVar{ + {Name: "POD_NAME", ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"}}}, + {Name: "RUNTIME_COMPONENT", Value: kind}, {Name: "RUNTIME_BASE", Value: base}, {Name: "RUNTIME_HOME", Value: home}, + {Name: "RUNTIME_MIGRATED_REPLICAS", Value: fmt.Sprint(migratedReplicas)}, + }, + SecurityContext: main.SecurityContext, Resources: main.Resources, + VolumeMounts: []corev1.VolumeMount{ + {Name: s.Volume, MountPath: "/runtime-store", SubPath: s.SubPath}, + {Name: configVolume, MountPath: "/runtime-prepared"}, + }, + } + foundSource := false + for i := range main.VolumeMounts { + m := &main.VolumeMounts[i] + if path.Clean(m.MountPath) == path.Clean(source) { + src := *m + src.MountPath = "/runtime-source" + src.ReadOnly = true + init.VolumeMounts = append(init.VolumeMounts, src) + m.Name = configVolume + m.ReadOnly = false + m.SubPath = "" + m.SubPathExpr = "" + foundSource = true + } else if path.Clean(m.MountPath) == home+"/conf" { + return fmt.Errorf("%s: direct conf mounts require an explicit PVC-backed custom configuration path", st.Name) + } + } + if !foundSource { + main.VolumeMounts = append(main.VolumeMounts, corev1.VolumeMount{Name: configVolume, MountPath: source}) + } + pod.Volumes = append(pod.Volumes, corev1.Volume{Name: configVolume, VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}}) + main.VolumeMounts = append(main.VolumeMounts, corev1.VolumeMount{Name: s.Volume, MountPath: RuntimeDir, SubPathExpr: path.Join(s.SubPath, StoreDir, "$(POD_NAME)")}) + // The initializer must not run the image's normal entrypoint. + pod.InitContainers = append(pod.InitContainers, init) + return nil +} + +func claimIdentity(st *appsv1.StatefulSet, name string) string { + for _, v := range st.Spec.VolumeClaimTemplates { + if v.Name == name { + return "template:" + name + } + } + for _, v := range st.Spec.Template.Spec.Volumes { + if v.Name == name && v.PersistentVolumeClaim != nil { + return "claim:" + v.PersistentVolumeClaim.ClaimName + } + } + return "" +}
diff --git a/pkg/common/utils/runtimeconfig/policy.go b/pkg/common/utils/runtimeconfig/policy.go new file mode 100644 index 0000000..97d7914 --- /dev/null +++ b/pkg/common/utils/runtimeconfig/policy.go
@@ -0,0 +1,199 @@ +// 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 runtimeconfig preserves mutable Doris configuration without changing +// user ConfigMaps, PVC templates, or the layout of image-provided conf directories. +package runtimeconfig + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + + dv1 "github.com/apache/doris-operator/api/disaggregated/v1" + v1 "github.com/apache/doris-operator/api/doris/v1" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + PolicyAnnotation = "doris.apache.org/runtime-config-policy" + MigrationAnnotation = "doris.apache.org/runtime-config-migration" + StorageAnnotation = "doris.apache.org/runtime-config-storage" + Auto = "auto" + Legacy = "legacy" + Enabled = "enabled" + Disabled = "disabled" +) + +// EnsurePolicy records the default before any child workload is created. The +// cluster-level decision survives deletion/recreation of an individual workload. +func EnsurePolicy(ctx context.Context, c client.Client, obj client.Object) error { + if !obj.GetDeletionTimestamp().IsZero() { + return nil + } + var explicit *bool + var hasStatus bool + switch o := obj.(type) { + case *v1.DorisCluster: + if o.Spec.RuntimeConfigPersistence != nil { + explicit = &o.Spec.RuntimeConfigPersistence.Enabled + } + hasStatus = !reflect.ValueOf(o.Status).IsZero() + case *dv1.DorisDisaggregatedCluster: + if o.Spec.RuntimeConfigPersistence != nil { + explicit = &o.Spec.RuntimeConfigPersistence.Enabled + } + hasStatus = !reflect.ValueOf(o.Status).IsZero() + default: + return fmt.Errorf("unsupported runtime configuration owner %T", obj) + } + current := obj.GetAnnotations()[PolicyAnnotation] + if current != "" && current != Auto && current != Legacy && current != Enabled && current != Disabled { + return fmt.Errorf("invalid %s value %q", PolicyAnnotation, current) + } + if (explicit == nil || (*explicit && current == Enabled)) && current != "" { + return nil + } + if explicit != nil && !*explicit && (current == Enabled || current == Auto) { + return fmt.Errorf("runtime configuration persistence cannot be disabled after activation; migrate overrides first") + } + var workloads appsv1.StatefulSetList + if err := c.List(ctx, &workloads, client.InNamespace(obj.GetNamespace())); err != nil { + return err + } + owned := make([]appsv1.StatefulSet, 0) + for _, st := range workloads.Items { + if ownedBy(&st, obj) { + owned = append(owned, st) + } + } + desired := current + if desired == "" { + desired = Auto + if hasStatus || len(owned) != 0 { + desired = Legacy + } + } + if explicit != nil { + if *explicit { + if desired != Enabled && (len(owned) != 0 || desired == Legacy) { + if err := validateMigration(obj, owned); err != nil { + return err + } + if err := validateMigrationPods(ctx, c, obj, owned); err != nil { + return err + } + } + desired = Enabled + } else { + desired = Disabled + } + } + if current == desired { + return nil + } + before := obj.DeepCopyObject().(client.Object) + annotations := obj.GetAnnotations() + if annotations == nil { + annotations = make(map[string]string) + } + annotations[PolicyAnnotation] = desired + obj.SetAnnotations(annotations) + return c.Patch(ctx, obj, client.MergeFrom(before)) +} + +type migrationWorkload struct { + UID string `json:"uid"` + Generation int64 `json:"generation"` + Replicas int32 `json:"replicas"` +} + +type migration struct { + ClusterUID string `json:"clusterUID"` + Workloads map[string]migrationWorkload `json:"workloads"` + PodUIDs map[string]string `json:"podUIDs"` +} + +func validateMigration(obj client.Object, workloads []appsv1.StatefulSet) error { + var approval migration + if err := json.Unmarshal([]byte(obj.GetAnnotations()[MigrationAnnotation]), &approval); err != nil || approval.ClusterUID != string(obj.GetUID()) { + return fmt.Errorf("runtime configuration migration required: run tools/runtime-config-migrate.py before enabling persistence") + } + if len(workloads) == 0 { + return fmt.Errorf("cannot validate runtime configuration migration without existing workloads") + } + for _, st := range workloads { + if !supportedContainer(st.Spec.Template.Spec.Containers) { + continue + } + old, ok := approval.Workloads[st.Name] + replicas := int32(1) + if st.Spec.Replicas != nil { + replicas = *st.Spec.Replicas + } + if !ok || old.UID != string(st.UID) || old.Generation != st.Generation || old.Replicas != replicas { + return fmt.Errorf("runtime configuration migration for %s is missing or stale; rerun the migration tool", st.Name) + } + } + return nil +} + +func validateMigrationPods(ctx context.Context, c client.Client, obj client.Object, workloads []appsv1.StatefulSet) error { + var approval migration + if err := json.Unmarshal([]byte(obj.GetAnnotations()[MigrationAnnotation]), &approval); err != nil { + return err + } + var pods corev1.PodList + if err := c.List(ctx, &pods, client.InNamespace(obj.GetNamespace())); err != nil { + return err + } + for _, st := range workloads { + if !supportedContainer(st.Spec.Template.Spec.Containers) { + continue + } + count := int32(0) + for _, pod := range pods.Items { + if !ownedBy(&pod, &st) { + continue + } + if !pod.DeletionTimestamp.IsZero() || approval.PodUIDs[pod.Name] != string(pod.UID) { + return fmt.Errorf("Pod %s changed since migration; rerun the migration tool", pod.Name) + } + count++ + } + if count != approval.Workloads[st.Name].Replicas { + return fmt.Errorf("Pod membership for %s changed since migration", st.Name) + } + } + return nil +} + +// Older operator resources use owner references without Controller=true. +func ownedBy(child, owner client.Object) bool { + if owner.GetUID() == "" { + return false + } + for _, ref := range child.GetOwnerReferences() { + if ref.UID == owner.GetUID() { + return true + } + } + return false +}
diff --git a/pkg/common/utils/runtimeconfig/prepare.sh b/pkg/common/utils/runtimeconfig/prepare.sh new file mode 100644 index 0000000..4102bd2 --- /dev/null +++ b/pkg/common/utils/runtimeconfig/prepare.sh
@@ -0,0 +1,52 @@ +# 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. + +set -euo pipefail +store="/runtime-store/.doris-runtime-config/$POD_NAME" +mkdir -p "$store" +ordinal="${POD_NAME##*-}" +if (( ordinal < RUNTIME_MIGRATED_REPLICAS )) && [[ ! -f "$store/.migration-complete" ]]; then + echo 'Runtime configuration migration is missing; refusing to start with empty overrides.' >&2 + exit 1 +fi +if [[ -d /runtime-source ]]; then + cp -RL /runtime-source/. /runtime-prepared/ +fi +if [[ ! -f "/runtime-prepared/$RUNTIME_BASE" ]]; then + cp "$RUNTIME_HOME/conf/$RUNTIME_BASE" "/runtime-prepared/$RUNTIME_BASE" +fi +custom="${RUNTIME_COMPONENT}_custom.conf" +key=custom_config_dir +value=/var/lib/doris/runtime-config +if [[ "$RUNTIME_COMPONENT" == ms ]]; then + key=custom_conf_path + value="$value/$custom" +fi +# ConfigMaps may contain initial overrides. Seed once, then remove the source +# entry so the image's entrypoint cannot symlink the mutable file to a ConfigMap. +if [[ ! -e "$store/$custom" ]]; then + if [[ -f "/runtime-prepared/$custom" ]]; then + cp "/runtime-prepared/$custom" "$store/$custom" + else + : > "$store/$custom" + fi +fi +rm -f "/runtime-prepared/$custom" +# Remove prior assignments to avoid parser-specific duplicate-key behavior. +awk -v key="$key" '$0 !~ "^[[:space:]]*" key "[[:space:]]*=" {print}' "/runtime-prepared/$RUNTIME_BASE" > /runtime-prepared/base.tmp +printf '\n%s = %s\n' "$key" "$value" >> /runtime-prepared/base.tmp +mv /runtime-prepared/base.tmp "/runtime-prepared/$RUNTIME_BASE"
diff --git a/pkg/common/utils/runtimeconfig/runtimeconfig_test.go b/pkg/common/utils/runtimeconfig/runtimeconfig_test.go new file mode 100644 index 0000000..553027c --- /dev/null +++ b/pkg/common/utils/runtimeconfig/runtimeconfig_test.go
@@ -0,0 +1,396 @@ +// 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 runtimeconfig + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + + dv1 "github.com/apache/doris-operator/api/disaggregated/v1" + v1 "github.com/apache/doris-operator/api/doris/v1" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/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" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func fakeClient(objects ...client.Object) client.Client { + scheme := runtime.NewScheme() + _ = appsv1.AddToScheme(scheme) + _ = corev1.AddToScheme(scheme) + _ = v1.AddToScheme(scheme) + _ = dv1.AddToScheme(scheme) + return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build() +} +func owner() *v1.DorisCluster { + return &v1.DorisCluster{ObjectMeta: metav1.ObjectMeta{Name: "cluster", Namespace: "test", UID: "cluster-uid"}} +} +func workload(kind string) *appsv1.StatefulSet { + n := int32(2) + return &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "cluster-" + kind, Namespace: "test", UID: types.UID(kind + "-uid"), Generation: 3, OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(owner(), v1.GroupVersion.WithKind("DorisCluster"))}}, Spec: appsv1.StatefulSetSpec{Replicas: &n, Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: kind, Image: "doris:test", Env: []corev1.EnvVar{{Name: "POD_NAME", ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"}}}}, VolumeMounts: []corev1.VolumeMount{{Name: "log", MountPath: "/opt/apache-doris/" + kind + "/log"}, {Name: "config", MountPath: "/etc/doris", ReadOnly: true}}}}, Volumes: []corev1.Volume{{Name: "log", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "existing"}}}, {Name: "config", VolumeSource: corev1.VolumeSource{ConfigMap: &corev1.ConfigMapVolumeSource{LocalObjectReference: corev1.LocalObjectReference{Name: "user-config"}}}}}}}}} +} + +func TestPolicyDefaultsAndMigration(t *testing.T) { + ctx := context.Background() + for _, legacy := range []bool{false, true} { + t.Run(map[bool]string{false: "new", true: "existing"}[legacy], func(t *testing.T) { + obj := owner() + objects := []client.Object{obj} + if legacy { + st := workload("fe") + st.OwnerReferences[0].Controller = nil + objects = append(objects, st) + } + c := fakeClient(objects...) + if err := EnsurePolicy(ctx, c, obj); err != nil { + t.Fatal(err) + } + want := Auto + if legacy { + want = Legacy + } + if obj.Annotations[PolicyAnnotation] != want { + t.Fatalf("policy=%v", obj.Annotations) + } + // A later resource deletion/recreation must not change the recorded decision. + if legacy { + if err := c.Delete(ctx, workload("fe")); err != nil { + t.Fatal(err) + } + } + if err := EnsurePolicy(ctx, c, obj); err != nil { + t.Fatal(err) + } + if obj.Annotations[PolicyAnnotation] != want { + t.Fatal("policy changed") + } + }) + } + obj := owner() + obj.Spec.RuntimeConfigPersistence = &v1.RuntimeConfigPersistenceSpec{Enabled: true} + st := workload("fe") + *st.Spec.Replicas = 0 + c := fakeClient(obj, st) + if err := EnsurePolicy(ctx, c, obj); err == nil { + t.Fatal("accepted migration without backup") + } + approval := migration{ClusterUID: string(obj.UID), Workloads: map[string]migrationWorkload{st.Name: {UID: string(st.UID), Generation: st.Generation, Replicas: *st.Spec.Replicas}}} + raw, _ := json.Marshal(approval) + obj.Annotations = map[string]string{MigrationAnnotation: string(raw)} + if err := EnsurePolicy(ctx, c, obj); err != nil { + t.Fatal(err) + } + if obj.Annotations[PolicyAnnotation] != Enabled { + t.Fatal("not enabled") + } + obj.Spec.RuntimeConfigPersistence.Enabled = false + if err := EnsurePolicy(ctx, c, obj); err == nil { + t.Fatal("accepted lossy disable") + } + approval.Workloads[st.Name] = migrationWorkload{UID: "stale", Generation: st.Generation, Replicas: 2} + raw, _ = json.Marshal(approval) + obj.Annotations[MigrationAnnotation] = string(raw) + if err := validateMigration(obj, []appsv1.StatefulSet{*st}); err == nil { + t.Fatal("accepted stale migration") + } +} + +func TestDDCPolicyAndExistingStatus(t *testing.T) { + obj := &dv1.DorisDisaggregatedCluster{ObjectMeta: metav1.ObjectMeta{Name: "ddc", Namespace: "test"}} + c := fakeClient(obj) + if err := EnsurePolicy(context.Background(), c, obj); err != nil { + t.Fatal(err) + } + if obj.Annotations[PolicyAnnotation] != Auto { + t.Fatal(obj.Annotations) + } + old := owner() + old.Status.FEStatus = &v1.ComponentStatus{} + c = fakeClient(old) + if err := EnsurePolicy(context.Background(), c, old); err != nil { + t.Fatal(err) + } + if old.Annotations[PolicyAnnotation] != Legacy { + t.Fatal("existing status was ignored") + } +} + +func TestPrepareComponentsAndKeepTemplates(t *testing.T) { + for _, kind := range []string{"fe", "be", "compute", "metaservice"} { + t.Run(kind, func(t *testing.T) { + obj := owner() + obj.Annotations = map[string]string{PolicyAnnotation: Auto} + st := workload(kind) + before := st.DeepCopy() + if err := Prepare(context.Background(), fakeClient(), nil, obj, st, nil); err != nil { + t.Fatal(err) + } + pod := st.Spec.Template.Spec + if len(pod.InitContainers) != 1 || pod.InitContainers[0].Image != "doris:test" { + t.Fatalf("init=%+v", pod.InitContainers) + } + if !reflect.DeepEqual(before.Spec.VolumeClaimTemplates, st.Spec.VolumeClaimTemplates) { + t.Fatal("PVC templates changed") + } + main := pod.Containers[0] + if main.VolumeMounts[len(main.VolumeMounts)-1].SubPathExpr != StoreDir+"/$(POD_NAME)" { + t.Fatal(main.VolumeMounts) + } + if main.VolumeMounts[1].Name != configVolume || main.VolumeMounts[1].ReadOnly { + t.Fatal("base config not staged") + } + for _, v := range pod.Volumes { + if v.Name == "config" && v.ConfigMap.Name != "user-config" { + t.Fatal("user ConfigMap changed") + } + } + // Reconcile should produce the same template and selected storage. + again := before.DeepCopy() + if err := Prepare(context.Background(), fakeClient(st), nil, obj, again, nil); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(st.Spec, again.Spec) || !reflect.DeepEqual(st.Annotations, again.Annotations) { + t.Fatal("non-idempotent prepare") + } + }) + } +} + +func TestStorageSafetyAndCompatibility(t *testing.T) { + ctx := context.Background() + obj := owner() + obj.Annotations = map[string]string{PolicyAnnotation: Auto} + st := workload("compute") + st.Spec.Template.Spec.Containers[0].VolumeMounts[0].MountPath = "/cache" + cfg := map[string]interface{}{"file_cache_path": `[{"path":"/cache"}]`} + original := st.Spec.Template.DeepCopy() + if err := Prepare(ctx, fakeClient(), nil, obj, st, cfg); err != nil { + t.Fatal(err) + } + if st.Annotations[StorageAnnotation] != "unavailable" || !reflect.DeepEqual(original, &st.Spec.Template) { + t.Fatal("cache root used") + } + obj.Annotations[PolicyAnnotation] = Enabled + if err := Prepare(ctx, fakeClient(), nil, obj, st, cfg); err == nil { + t.Fatal("explicit enable accepted cache-only PVC") + } + obj.Annotations[PolicyAnnotation] = Legacy + st = workload("be") + original = st.Spec.Template.DeepCopy() + if err := Prepare(ctx, fakeClient(), nil, obj, st, nil); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(original, &st.Spec.Template) { + t.Fatal("legacy changed") + } + obj.Annotations[PolicyAnnotation] = Auto + if err := Prepare(ctx, fakeClient(st), nil, obj, st, nil); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(original, &st.Spec.Template) { + t.Fatal("existing unmanaged workload changed") + } + // A user-owned durable path is respected. + st = workload("fe") + if err := Prepare(ctx, fakeClient(), nil, obj, st, map[string]interface{}{"custom_config_dir": "/opt/apache-doris/fe/log/custom"}); err != nil { + t.Fatal(err) + } + if len(st.Spec.Template.Spec.InitContainers) != 0 || st.Annotations[StorageAnnotation] != "external" { + t.Fatal("explicit path replaced") + } + if err := Prepare(ctx, fakeClient(), nil, obj, workload("fe"), map[string]interface{}{"custom_config_dir": "/tmp/custom"}); err == nil { + t.Fatal("ephemeral custom path accepted") + } + // VolumeClaimTemplates also represent durable storage before the Pod exists. + st = workload("fe") + st.Spec.Template.Spec.Volumes = st.Spec.Template.Spec.Volumes[1:] + st.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{{ObjectMeta: metav1.ObjectMeta{Name: "log"}}} + if err := Prepare(ctx, fakeClient(), nil, obj, st, nil); err != nil { + t.Fatal(err) + } + if len(st.Spec.Template.Spec.InitContainers) == 0 { + t.Fatal("claim template not recognized") + } +} + +func TestPrepareShellPreservesOverrides(t *testing.T) { + for _, kind := range []string{"fe", "be", "ms"} { + t.Run(kind, func(t *testing.T) { + root := t.TempDir() + store := filepath.Join(root, "store") + src := filepath.Join(root, "source") + out := filepath.Join(root, "out") + home := filepath.Join(root, "home") + for _, dir := range []string{store, src, out, filepath.Join(home, "conf")} { + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + } + base := kind + ".conf" + if kind == "ms" { + base = "doris_cloud.conf" + } + if err := os.WriteFile(filepath.Join(src, base), []byte("port=1234\n"), 0644); err != nil { + t.Fatal(err) + } + script := strings.NewReplacer("/runtime-store", store, "/runtime-source", src, "/runtime-prepared", out).Replace(prepareScript) + run := func(pod, replicas string) error { + cmd := exec.Command("bash", "-ec", script) + cmd.Env = append(os.Environ(), "POD_NAME="+pod, "RUNTIME_MIGRATED_REPLICAS="+replicas, "RUNTIME_COMPONENT="+kind, "RUNTIME_BASE="+base, "RUNTIME_HOME="+home) + output, err := cmd.CombinedOutput() + if err != nil { + return &shellError{err, string(output)} + } + return nil + } + if err := run("cluster-0", "0"); err != nil { + t.Fatal(err) + } + custom := filepath.Join(store, StoreDir, "cluster-0", kind+"_custom.conf") + if err := os.WriteFile(custom, []byte("dynamic=value\n"), 0644); err != nil { + t.Fatal(err) + } + // A ConfigMap/image restart cannot overwrite runtime modifications. + if err := run("cluster-0", "0"); err != nil { + t.Fatal(err) + } + got, _ := os.ReadFile(custom) + if string(got) != "dynamic=value\n" { + t.Fatal(string(got)) + } + got, _ = os.ReadFile(filepath.Join(out, base)) + if !strings.Contains(string(got), RuntimeDir) { + t.Fatal(string(got)) + } + if err := run("cluster-0", "1"); err == nil { + t.Fatal("missing migration marker accepted") + } + if err := os.WriteFile(filepath.Join(filepath.Dir(custom), ".migration-complete"), nil, 0644); err != nil { + t.Fatal(err) + } + if err := run("cluster-0", "1"); err != nil { + t.Fatal(err) + } + if err := run("cluster-1", "1"); err != nil { + t.Fatal("scale-out blocked:", err) + } + }) + } +} + +type shellError struct { + err error + output string +} + +func (e *shellError) Error() string { return e.err.Error() + ": " + e.output } + +func TestMigrationPodIdentityAndStorageIdentity(t *testing.T) { + obj := owner() + st := workload("fe") + *st.Spec.Replicas = 1 + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "cluster-fe-0", Namespace: "test", UID: "pod-new", OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(st, appsv1.SchemeGroupVersion.WithKind("StatefulSet"))}}} + approval := migration{ClusterUID: string(obj.UID), Workloads: map[string]migrationWorkload{st.Name: {UID: string(st.UID), Generation: st.Generation, Replicas: 1}}, PodUIDs: map[string]string{pod.Name: "pod-old"}} + raw, _ := json.Marshal(approval) + obj.Annotations = map[string]string{MigrationAnnotation: string(raw)} + c := fakeClient(obj, st, pod) + if err := validateMigrationPods(context.Background(), c, obj, []appsv1.StatefulSet{*st}); err == nil { + t.Fatal("replacement Pod accepted") + } + approval.PodUIDs[pod.Name] = string(pod.UID) + raw, _ = json.Marshal(approval) + obj.Annotations[MigrationAnnotation] = string(raw) + if err := validateMigrationPods(context.Background(), c, obj, []appsv1.StatefulSet{*st}); err != nil { + t.Fatal(err) + } + obj.Annotations[PolicyAnnotation] = Auto + st = workload("fe") + if err := Prepare(context.Background(), fakeClient(), nil, obj, st, nil); err != nil { + t.Fatal(err) + } + desired := workload("fe") + desired.Spec.Template.Spec.Volumes[0].PersistentVolumeClaim.ClaimName = "different-disk" + if err := Prepare(context.Background(), fakeClient(st), nil, obj, desired, nil); err == nil { + t.Fatal("PVC replacement silently lost configuration") + } +} + +func TestShellImageDefaultsAndConfigMapSeed(t *testing.T) { + root := t.TempDir() + store := filepath.Join(root, "store") + src := filepath.Join(root, "source") + out := filepath.Join(root, "out") + home := filepath.Join(root, "home") + for _, dir := range []string{store, out, filepath.Join(home, "conf")} { + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(filepath.Join(home, "conf", "be.conf"), []byte("image_default = true\n"), 0644); err != nil { + t.Fatal(err) + } + script := strings.NewReplacer("/runtime-store", store, "/runtime-source", src, "/runtime-prepared", out).Replace(prepareScript) + run := func() { + t.Helper() + cmd := exec.Command("bash", "-ec", script) + cmd.Env = append(os.Environ(), "POD_NAME=cluster-0", "RUNTIME_MIGRATED_REPLICAS=0", "RUNTIME_COMPONENT=be", "RUNTIME_BASE=be.conf", "RUNTIME_HOME="+home) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("%v: %s", err, output) + } + } + run() + base, _ := os.ReadFile(filepath.Join(out, "be.conf")) + if !strings.Contains(string(base), "image_default = true") { + t.Fatal("image defaults lost") + } + if err := os.MkdirAll(src, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(src, "be_custom.conf"), []byte("seed=initial\n"), 0644); err != nil { + t.Fatal(err) + } + custom := filepath.Join(store, StoreDir, "cluster-0", "be_custom.conf") + if err := os.Remove(custom); err != nil { + t.Fatal(err) + } + run() + value, _ := os.ReadFile(custom) + if string(value) != "seed=initial\n" { + t.Fatal("ConfigMap seed lost") + } + if _, err := os.Stat(filepath.Join(out, "be_custom.conf")); !os.IsNotExist(err) { + t.Fatal("mutable file exposed through source ConfigMap") + } + if err := os.WriteFile(custom, []byte("seed=changed\n"), 0644); err != nil { + t.Fatal(err) + } + run() + value, _ = os.ReadFile(custom) + if string(value) != "seed=changed\n" { + t.Fatal("runtime override replaced by seed") + } +}
diff --git a/pkg/controller/disaggregated_cluster_controller.go b/pkg/controller/disaggregated_cluster_controller.go index 69b753d..a29bac9 100644 --- a/pkg/controller/disaggregated_cluster_controller.go +++ b/pkg/controller/disaggregated_cluster_controller.go
@@ -26,6 +26,7 @@ dv1 "github.com/apache/doris-operator/api/disaggregated/v1" "github.com/apache/doris-operator/pkg/common/utils/hash" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" sc "github.com/apache/doris-operator/pkg/controller/sub_controller" dcgs "github.com/apache/doris-operator/pkg/controller/sub_controller/disaggregated_cluster/computegroups" dfe "github.com/apache/doris-operator/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe" @@ -220,6 +221,14 @@ klog.Warningf("disaggreatedClusterReconciler not find resource DorisDisaggregatedCluster namespaceName %s", req.NamespacedName) return ctrl.Result{}, nil } + if err != nil { + return ctrl.Result{}, err + } + if err := runtimeconfig.EnsurePolicy(ctx, dc.Client, &ddc); err != nil { + dc.Recorder.Event(&ddc, corev1.EventTypeWarning, "RuntimeConfigPersistenceBlocked", err.Error()) + return ctrl.Result{}, err + } + hv := hash.HashObject(ddc.Spec) var res ctrl.Result
diff --git a/pkg/controller/doriscluster_controller.go b/pkg/controller/doriscluster_controller.go index a5961e0..958d044 100644 --- a/pkg/controller/doriscluster_controller.go +++ b/pkg/controller/doriscluster_controller.go
@@ -38,6 +38,7 @@ dorisv1 "github.com/apache/doris-operator/api/doris/v1" "github.com/apache/doris-operator/pkg/common/utils/k8s" "github.com/apache/doris-operator/pkg/common/utils/resource" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "github.com/apache/doris-operator/pkg/controller/sub_controller" "github.com/apache/doris-operator/pkg/controller/sub_controller/be" bk "github.com/apache/doris-operator/pkg/controller/sub_controller/broker" @@ -135,6 +136,11 @@ return ctrl.Result{}, nil } + if err := runtimeconfig.EnsurePolicy(ctx, r.Client, dcr); err != nil { + r.Recorder.Event(dcr, corev1.EventTypeWarning, "RuntimeConfigPersistenceBlocked", err.Error()) + return ctrl.Result{}, err + } + if dcr.Spec.EnableRestartWhenConfigChange { coreConfigMaps := resource.GetDorisCoreConfigMapNames(dcr) for componentType := range coreConfigMaps {
diff --git a/pkg/controller/sub_controller/be/controller.go b/pkg/controller/sub_controller/be/controller.go index ae6772b..84fab24 100644 --- a/pkg/controller/sub_controller/be/controller.go +++ b/pkg/controller/sub_controller/be/controller.go
@@ -22,6 +22,7 @@ "github.com/apache/doris-operator/api/doris/v1" "github.com/apache/doris-operator/pkg/common/utils/k8s" "github.com/apache/doris-operator/pkg/common/utils/resource" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "github.com/apache/doris-operator/pkg/controller/sub_controller" appv1 "k8s.io/api/apps/v1" "k8s.io/client-go/tools/record" @@ -100,6 +101,9 @@ } st := be.buildBEStatefulSet(dcr, config) + if err := runtimeconfig.Prepare(ctx, be.K8sclient, be.K8srecorder, dcr, &st, config); err != nil { + return err + } if !be.PrepareReconcileResources(ctx, dcr, v1.Component_BE) { klog.Infof("be controller sync preparing resource for reconciling namespace %s name %s!", dcr.Namespace, dcr.Name) return nil
diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go index bb3bdd3..be93ef4 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go
@@ -22,6 +22,7 @@ "encoding/json" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "regexp" "strconv" "strings" @@ -151,6 +152,9 @@ } cvs := dcgs.GetConfigValuesFromConfigMaps(ddc.Namespace, resource.BE_RESOLVEKEY, cg.CommonSpec.ConfigMaps) st := dcgs.NewStatefulset(ddc, cg, cvs) + if err := runtimeconfig.Prepare(ctx, dcgs.K8sclient, dcgs.K8srecorder, ddc, st, cvs); err != nil { + return nil, err + } internalSvc := dcgs.newInternalService(ddc, cg, cvs) externalSvc := dcgs.newExternalService(ddc, cg, cvs) dcgs.initialCGStatus(ddc, cg) @@ -251,8 +255,21 @@ 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) @@ -689,6 +706,13 @@ continue } if int32(index) >= replicas { + retain, err := runtimeconfig.RetainPVCOnScaleDown(sts, pvcName) + if err != nil { + return err + } + if retain { + continue + } clearPVC = append(clearPVC, pvcName) } }
diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go index 87249a4..c4e15f6 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go
@@ -25,9 +25,12 @@ sc "github.com/apache/doris-operator/pkg/controller/sub_controller" appv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" 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" "sigs.k8s.io/controller-runtime/pkg/client/fake" ) @@ -120,3 +123,59 @@ }, } } + +func TestScaleDownRetainsRuntimeConfigPVC(t *testing.T) { + for _, tc := range []struct { + name string + recorded string + keepLog bool + wantError bool + }{ + {name: "legacy keeps existing cleanup"}, + {name: "selected log survives", recorded: `{"volume":"be-log","claim":"template:be-log","mountPath":"/opt/apache-doris/be/log"}`, keepLog: true}, + {name: "invalid storage blocks cleanup", recorded: `{`, keepLog: true, wantError: true}, + } { + t.Run(tc.name, func(t *testing.T) { + scheme := runtime.NewScheme() + for _, add := range []func(*runtime.Scheme) error{appv1.AddToScheme, corev1.AddToScheme, dv1.AddToScheme} { + if err := add(scheme); err != nil { + t.Fatal(err) + } + } + ddc := newTestDDC() + cg := newTestCG("cg1") + ddc.Spec.ComputeGroups = []dv1.ComputeGroup{*cg} + replicas := int32(1) + st := newTestStatefulSet(ddc.Namespace, ddc.GetCGStatefulsetName(cg), "1Gi") + st.Spec.Replicas = &replicas + st.Spec.VolumeClaimTemplates[0].Name = "be-log" + st.Annotations = map[string]string{"doris.apache.org/runtime-config-storage": tc.recorded} + dcgs := &DisaggregatedComputeGroupsController{} + labels := dcgs.newCGPodsSelector(ddc.Name, cg.UniqueId) + objects := []client.Object{st} + for _, volume := range []string{"be-log", "be-file-cache"} { + for _, ordinal := range []string{"0", "1"} { + objects = append(objects, &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: volume + "-" + st.Name + "-" + ordinal, Namespace: ddc.Namespace, Labels: labels}}) + } + } + dcgs.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build() + err := dcgs.ClearStatefulsetUnusedPVCs(context.Background(), ddc, dv1.ComputeGroupStatus{UniqueId: cg.UniqueId}) + if (err != nil) != tc.wantError { + t.Fatalf("cleanup error = %v", err) + } + for _, volume := range []string{"be-log", "be-file-cache"} { + for _, ordinal := range []string{"0", "1"} { + var pvc corev1.PersistentVolumeClaim + err := dcgs.K8sclient.Get(context.Background(), types.NamespacedName{Namespace: ddc.Namespace, Name: volume + "-" + st.Name + "-" + ordinal}, &pvc) + want := ordinal == "0" || tc.wantError || (volume == "be-log" && tc.keepLog) + if want && err != nil { + t.Fatalf("expected retained PVC %s/%s: %v", volume, ordinal, err) + } + if !want && !apierrors.IsNotFound(err) { + t.Fatalf("expected deleted PVC %s/%s: %v", volume, ordinal, err) + } + } + } + }) + } +}
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 0000000..a57d91a --- /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 0000000..7a7122e --- /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 10be02f..a20d9d2 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 @@ "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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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,
diff --git a/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go index f9c60d4..eecd92b 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go
@@ -21,6 +21,7 @@ "context" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "strconv" "strings" @@ -87,6 +88,9 @@ svc := dfc.newService(ddc, confMap) st := dfc.NewStatefulset(ddc, confMap) + if err := runtimeconfig.Prepare(ctx, dfc.K8sclient, dfc.K8srecorder, ddc, st, confMap); err != nil { + return err + } //initial fe status on start. in resource process step, may be use the status record the process. dfc.initialFEStatus(ddc)
diff --git a/pkg/controller/sub_controller/disaggregated_cluster/metaservice/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/metaservice/controller.go index a7a3936..407dc9d 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/metaservice/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/metaservice/controller.go
@@ -21,6 +21,7 @@ "context" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "strconv" "strings" @@ -169,6 +170,9 @@ svc := dms.newService(ddc, confMap) st := dms.newStatefulset(ddc, confMap, fdbEndpoint) + if err := runtimeconfig.Prepare(ctx, dms.K8sclient, dms.K8srecorder, ddc, st, confMap); err != nil { + return err + } dms.initMSStatus(ddc) dms.CheckSecretMountPath(ddc, ddc.Spec.MetaService.Secrets)
diff --git a/pkg/controller/sub_controller/fe/controller.go b/pkg/controller/sub_controller/fe/controller.go index 6b10d8b..cebc800 100644 --- a/pkg/controller/sub_controller/fe/controller.go +++ b/pkg/controller/sub_controller/fe/controller.go
@@ -22,6 +22,7 @@ v1 "github.com/apache/doris-operator/api/doris/v1" "github.com/apache/doris-operator/pkg/common/utils/k8s" "github.com/apache/doris-operator/pkg/common/utils/resource" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "github.com/apache/doris-operator/pkg/controller/sub_controller" appv1 "k8s.io/api/apps/v1" "k8s.io/client-go/tools/record" @@ -130,6 +131,9 @@ } st := fc.buildFEStatefulSet(cluster, config) + if err := runtimeconfig.Prepare(ctx, fc.K8sclient, fc.K8srecorder, cluster, &st, config); err != nil { + return err + } if err = k8s.ApplyStatefulSet(ctx, fc.K8sclient, &st, func(new *appv1.StatefulSet, old *appv1.StatefulSet) bool { fc.RestrictConditionsEqual(new, old) return resource.StatefulSetDeepEqual(new, old, false)
diff --git a/tools/runtime-config-migrate.py b/tools/runtime-config-migrate.py new file mode 100755 index 0000000..e7748cf --- /dev/null +++ b/tools/runtime-config-migrate.py
@@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +# 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. + +"""Back up, review, and migrate runtime overrides using existing PVC mounts. + +Requires Python 3 and kubectl. Run without --apply first, review the private +backup directory (especially MS overrides), then rerun with --apply. Stop mutable +configuration updates and workload scaling/image changes during both steps. +No configuration values are printed or written to Kubernetes annotations. +""" + +import argparse +import hashlib +import json +import os +from pathlib import Path, PurePosixPath +import subprocess +import sys + +POLICY = "doris.apache.org/runtime-config-policy" +MIGRATION = "doris.apache.org/runtime-config-migration" +STORE = ".doris-runtime-config" +KINDS = {"ddc": "dorisdisaggregatedclusters.disaggregated.cluster.doris.com", + "dcr": "dorisclusters.doris.selectdb.com"} + + +def parse_conf(text): + result = {} + for line in text.splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, value = line.split("=", 1) + result[key.strip()] = value.strip() + return result + + +def within(target, root): + return target == root or target.startswith(root.rstrip("/") + "/") + + +def choose_mount(st, container, kind, config): + """Keep ranking and exclusions in sync with runtimeconfig.selectStorage.""" + persistent = {v["name"] for v in st["spec"]["template"]["spec"].get("volumes", []) + if "persistentVolumeClaim" in v and not v["persistentVolumeClaim"].get("readOnly")} + persistent.update(v["metadata"]["name"] for v in st["spec"].get("volumeClaimTemplates", [])) + roots = [entry["path"].replace("${DORIS_HOME}", "/opt/apache-doris/be").replace("$DORIS_HOME", "/opt/apache-doris/be") for entry in json.loads(config.get("file_cache_path", "[]"))] + candidates = [] + for mount in container.get("volumeMounts", []): + if mount["name"] not in persistent or mount.get("readOnly") or mount.get("subPathExpr"): + continue + path = mount["mountPath"].rstrip("/") + rank = 0 + if PurePosixPath(path).name == "log" or path in (config.get("log_dir"), config.get("LOG_DIR")): + rank = 1 + if kind == "fe" and (PurePosixPath(path).name == "doris-meta" or path == config.get("meta_dir")): + rank = 2 + if container["name"].lower() == "be" and (PurePosixPath(path).name == "storage" or path in + [p.strip().split(",")[0] for p in config.get("storage_root_path", "").split(";")]): + rank = 3 + if any(within(path, p) or within(p, path) for p in roots): + rank = 0 + if rank: + candidates.append((rank, path, mount)) + if not candidates: + raise RuntimeError(f"{st['metadata']['name']}: no safe existing PVC; cache roots/emptyDir cannot be used") + return min(candidates, key=lambda entry: entry[:2])[2] + + +def ms_overrides(effective, baseline): + # FDB endpoint is injected by the entrypoint. Never freeze it in an override. + ignored = {"fdb_cluster", "fdb_cluster_file_path", "custom_conf_path"} + return "".join(f"{key} = {value}\n" for key, value in sorted(effective.items()) + if key not in ignored and baseline.get(key) != value) + + +def private_write(path, data): + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w") as stream: + stream.write(data) + + +class Kubectl: + def __init__(self, namespace, context): + self.prefix = ["kubectl"] + if context: + self.prefix += ["--context", context] + self.prefix += ["--namespace", namespace] + + def run(self, args, data=None): + result = subprocess.run(self.prefix + args, input=data, text=True, capture_output=True) + if result.returncode: + # Do not propagate stdout/stderr: API/exec failures can contain config values. + raise RuntimeError(f"kubectl {args[0]} failed (exit {result.returncode}); inspect the cluster separately") + return result.stdout + + def get(self, kind, name=None): + return json.loads(self.run(["get", kind] + ([name] if name else []) + ["-o", "json"])) + + def read(self, pod, container, path): + return self.run(["exec", pod, "-c", container, "--", "bash", "-ec", + 'if [[ -f "$1" ]]; then cat -- "$1"; fi', "read-config", path]) + + +def controlled(resource, uid): + return any(ref["uid"] == uid + for ref in resource["metadata"].get("ownerReferences", [])) + + +def collect(kube, kind, name): + cluster = kube.get(KINDS[kind], name) + if cluster["metadata"].get("annotations", {}).get(POLICY) in ("enabled", "auto"): + raise RuntimeError("cluster has already enabled automatic persistence; do not migrate active overrides") + workloads = [s for s in kube.get("statefulsets")["items"] if controlled(s, cluster["metadata"]["uid"])] + if not workloads: + raise RuntimeError("no existing StatefulSets found") + pods = kube.get("pods")["items"] + plan = {"clusterUID": cluster["metadata"]["uid"], "workloads": {}, "podUIDs": {}, "pods": []} + for st in workloads: + containers = st["spec"]["template"]["spec"]["containers"] + main = next((c for c in containers if c["name"].lower() in ("fe", "be", "compute", "metaservice")), None) + if not main: + continue + comp = {"compute": "be", "metaservice": "ms"}.get(main["name"].lower(), main["name"].lower()) + meta = st["metadata"] + replicas = st["spec"].get("replicas", 1) + status = st.get("status", {}) + if replicas and (status.get("observedGeneration", 0) < meta["generation"] or + status.get("currentRevision") != status.get("updateRevision")): + raise RuntimeError(f"{meta['name']}: wait for the existing rollout to finish") + members = [p for p in pods if controlled(p, meta["uid"]) and not p["metadata"].get("deletionTimestamp")] + if len(members) != replicas: + raise RuntimeError(f"{meta['name']}: replicas are not stable") + plan["workloads"][meta["name"]] = {"uid": meta["uid"], "generation": meta["generation"], "replicas": replicas} + for pod in members: + if status.get("updateRevision") and pod["metadata"].get("labels", {}).get("controller-revision-hash") != status["updateRevision"]: + raise RuntimeError(f"{pod['metadata']['name']}: Pod is not at the current StatefulSet revision") + pod_name = pod["metadata"]["name"] + plan["podUIDs"][pod_name] = pod["metadata"]["uid"] + if not any(c.get("type") == "Ready" and c.get("status") == "True" for c in pod.get("status", {}).get("conditions", [])): + raise RuntimeError(f"{pod_name}: Pod must be Ready before migration") + home = "/opt/apache-doris/" + comp + base = "doris_cloud.conf" if comp == "ms" else comp + ".conf" + base_text = kube.read(pod_name, main["name"], home + "/conf/" + base) + if not base_text: + raise RuntimeError(f"{pod_name}: base configuration is missing") + config = parse_conf(base_text) + # An existing explicit path needs individual verification, not relocation. + key = "custom_conf_path" if comp == "ms" else "custom_config_dir" + if config.get(key) and config[key] not in (home + "/conf", "${DORIS_HOME}/conf", home + "/conf/" + base): + raise RuntimeError(f"{pod_name}: custom configuration path already set; verify/migrate it separately") + mount = choose_mount(st, main, comp, config) + filename = comp + "_custom.conf" + if comp == "ms": + # Start with the original ConfigMap, or the entrypoint's image backup. + source = next((e["value"] for e in main.get("env", []) if e["name"] == "CONFIGMAP_PATH" and e.get("value")), "/etc/doris") + baseline = kube.read(pod_name, main["name"], source + "/" + base) + if not baseline: + raise RuntimeError(f"{pod_name}: MS baseline ConfigMap unavailable; export reviewed MS overrides manually before migration") + contents = ms_overrides(config, parse_conf(baseline)) + source_path = home + "/conf/" + base + source_text = base_text + else: + source_path = home + "/conf/" + filename + source_text = kube.read(pod_name, main["name"], source_path) + contents = source_text + if comp == "be" and config.get("deploy_mode") == "cloud": + persisted = dict(config) + persisted.update(parse_conf(contents)) + if not persisted.get("meta_service_endpoint") or not persisted.get("cloud_unique_id"): + raise RuntimeError(f"{pod_name}: persisted cloud heartbeat configuration is incomplete; recover it before migration") + target = mount["mountPath"].rstrip("/") + "/" + STORE + "/" + pod_name + plan["pods"].append({"pod": pod_name, "uid": pod["metadata"]["uid"], "container": main["name"], + "file": pod_name + ".conf", "contents": contents, + "source": source_path, "sourceSHA256": hashlib.sha256(source_text.encode()).hexdigest(), + "target": target, "filename": filename}) + return plan + + +def apply(kube, kind, name, directory, saved): + # Recollect before touching a PVC. This rejects replaced Pods, scale/image + # changes, and dynamic updates that occurred since the backup was reviewed. + live = collect(kube, kind, name) + keys = ("clusterUID", "workloads", "podUIDs") + if any(saved[k] != live[k] for k in keys): + raise RuntimeError("cluster/workload changed since backup; make a fresh backup") + current = {p["pod"]: p for p in live["pods"]} + for pod in saved["pods"]: + now = current.get(pod["pod"]) + if not now or any(pod[k] != now[k] for k in ("uid", "sourceSHA256", "target", "filename")): + raise RuntimeError(f"{pod['pod']}: Pod or runtime configuration changed; make a fresh backup") + if len(current) != len(saved["pods"]): + raise RuntimeError("Pod membership changed since backup") + for pod in saved["pods"]: + contents = (directory / pod["file"]).read_text() + # No shell expansion of file paths or contents. + kube.run(["exec", "-i", pod["pod"], "-c", pod["container"], "--", "bash", "-ec", ''' +set -euo pipefail +umask 077 +mkdir -p -- "$1" +tmp=$(mktemp "$1/.migration.XXXXXX") +trap 'rm -f -- "$tmp"' EXIT +cat > "$tmp" +if [[ -e "$1/$2" ]]; then + cmp -s -- "$tmp" "$1/$2" || exit 1 +else + mv -- "$tmp" "$1/$2" +fi +: > "$1/.migration-complete" +''', "migrate-config", pod["target"], pod["filename"]], data=contents) + # Read back without printing the content. + actual = kube.read(pod["pod"], pod["container"], pod["target"] + "/" + pod["filename"]) + if actual != contents: + raise RuntimeError(f"{pod['pod']}: backup verification failed") + # Make sure no image/replica/Pod change happened while copying files. + final = collect(kube, kind, name) + if saved["workloads"] != final["workloads"] or {(p["pod"], p["uid"], p["sourceSHA256"]) for p in saved["pods"]} != {(p["pod"], p["uid"], p["sourceSHA256"]) for p in final["pods"]}: + raise RuntimeError("cluster changed during migration; persistence was not enabled") + approval = {k: saved[k] for k in keys} + patch = {"metadata": {"annotations": {MIGRATION: json.dumps(approval, separators=(",", ":"))}}, + "spec": {"runtimeConfigPersistence": {"enabled": True}}} + kube.run(["patch", KINDS[kind], name, "--type=merge", "-p", json.dumps(patch)]) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--kind", required=True, choices=KINDS) + parser.add_argument("--name", required=True) + parser.add_argument("--namespace", required=True) + parser.add_argument("--context") + parser.add_argument("--backup-dir", required=True, type=Path) + parser.add_argument("--apply", action="store_true", help="apply a previously reviewed backup and enable persistence") + args = parser.parse_args() + kube = Kubectl(args.namespace, args.context) + manifest = args.backup_dir / "manifest.json" + if args.apply: + saved = json.loads(manifest.read_text()) + if saved.get("kind") != args.kind or saved.get("name") != args.name or saved.get("namespace") != args.namespace: + raise RuntimeError("backup belongs to another cluster") + apply(kube, args.kind, args.name, args.backup_dir, saved) + print("Migration verified; persistence enabled. Watch the normal rolling update.") + else: + if args.backup_dir.exists(): + raise RuntimeError("use a new backup directory; existing backups are never overwritten") + saved = collect(kube, args.kind, args.name) + args.backup_dir.mkdir(mode=0o700, parents=True) + for pod in saved["pods"]: + private_write(args.backup_dir / pod["file"], pod.pop("contents")) + saved.update(kind=args.kind, name=args.name, namespace=args.namespace) + private_write(manifest, json.dumps(saved, indent=2)) + print(f"Backed up {len(saved['pods'])} Pods. Review the private files, then rerun with --apply.") + + +if __name__ == "__main__": + try: + main() + except (RuntimeError, OSError, ValueError, KeyError) as error: + print(f"Migration stopped: {error}", file=sys.stderr) + sys.exit(1)
diff --git a/tools/tests/test_runtime_config_migrate.py b/tools/tests/test_runtime_config_migrate.py new file mode 100644 index 0000000..39213f3 --- /dev/null +++ b/tools/tests/test_runtime_config_migrate.py
@@ -0,0 +1,111 @@ +# 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. + +import copy +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest + +spec = importlib.util.spec_from_file_location("migrate", Path(__file__).parents[1] / "runtime-config-migrate.py") +migrate = importlib.util.module_from_spec(spec) +spec.loader.exec_module(migrate) + + +class FakeKubectl: + def __init__(self): + self.cluster = {"metadata": {"uid": "cluster-uid"}} + self.st = {"metadata": {"name": "cluster-be", "uid": "st-uid", "generation": 2, + "ownerReferences": [{"uid": "cluster-uid"}]}, + "status": {"observedGeneration": 2, "currentRevision": "rev", "updateRevision": "rev"}, + "spec": {"replicas": 1, "template": {"spec": { + "containers": [{"name": "BE", "volumeMounts": [{"name": "storage", "mountPath": "/storage"}]}], + "volumes": [{"name": "storage", "persistentVolumeClaim": {"claimName": "disk"}}]}}}} + self.pod = {"metadata": {"name": "cluster-be-0", "uid": "pod-uid", "labels": {"controller-revision-hash": "rev"}, + "ownerReferences": [{"controller": True, "uid": "st-uid"}]}, + "status": {"conditions": [{"type": "Ready", "status": "True"}]}} + self.files = {"/opt/apache-doris/be/conf/be.conf": "storage_root_path=/storage\n", + "/opt/apache-doris/be/conf/be_custom.conf": "mutable=value\n"} + self.patched = False + + def get(self, kind, name=None): + if kind in migrate.KINDS.values(): + return self.cluster + if kind == "statefulsets": + return {"items": [self.st]} + if kind == "pods": + return {"items": [self.pod]} + raise AssertionError(kind) + + def read(self, pod, container, path): + return self.files.get(path, "") + + def run(self, args, data=None): + if args[0] == "exec": + self.files[args[-2] + "/" + args[-1]] = data + elif args[0] == "patch": + self.patched = json.loads(args[-1]) + return "" + + +class MigrationTest(unittest.TestCase): + def test_ms_excludes_base_and_fdb(self): + result = migrate.ms_overrides({"port": "5000", "mutable": "new", "fdb_cluster": "injected"}, + {"port": "5000", "mutable": "old"}) + self.assertEqual(result, "mutable = new\n") + + def test_cache_only_is_rejected(self): + kube = FakeKubectl() + with self.assertRaisesRegex(RuntimeError, "no safe"): + migrate.choose_mount(kube.st, kube.st["spec"]["template"]["spec"]["containers"][0], + "be", {"file_cache_path": '[{"path":"/storage"}]'}) + + def test_collect_and_apply(self): + kube = FakeKubectl() + plan = migrate.collect(kube, "dcr", "cluster") + self.assertEqual(plan["podUIDs"], {"cluster-be-0": "pod-uid"}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for pod in plan["pods"]: + migrate.private_write(root / pod["file"], pod.pop("contents")) + self.assertEqual((root / pod["file"]).stat().st_mode & 0o777, 0o600) + migrate.apply(kube, "dcr", "cluster", root, plan) + self.assertTrue(kube.patched["spec"]["runtimeConfigPersistence"]["enabled"]) + self.assertNotIn("mutable", kube.patched["metadata"]["annotations"][migrate.MIGRATION]) + + def test_replaced_pod_never_enables(self): + kube = FakeKubectl() + plan = copy.deepcopy(migrate.collect(kube, "dcr", "cluster")) + kube.pod["metadata"]["uid"] = "replacement" + with tempfile.TemporaryDirectory() as directory: + with self.assertRaisesRegex(RuntimeError, "changed"): + migrate.apply(kube, "dcr", "cluster", Path(directory), plan) + self.assertFalse(kube.patched) + + def test_dynamic_change_never_enables(self): + kube = FakeKubectl() + plan = migrate.collect(kube, "dcr", "cluster") + kube.files["/opt/apache-doris/be/conf/be_custom.conf"] = "changed=yes\n" + with tempfile.TemporaryDirectory() as directory: + with self.assertRaisesRegex(RuntimeError, "changed"): + migrate.apply(kube, "dcr", "cluster", Path(directory), plan) + self.assertFalse(kube.patched) + + +if __name__ == "__main__": + unittest.main()