diff --git a/README-CN.md b/README-CN.md index 6ad1dbea..7e3a990b 100644 --- a/README-CN.md +++ b/README-CN.md @@ -52,7 +52,11 @@ kubectl apply -f https://raw.githubusercontent.com/apache/doris-operator/$(curl - [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 f54213a3..7b949f56 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,11 @@ kubectl apply -f https://raw.githubusercontent.com/apache/doris-operator/$(curl - [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 002968ff..99c5f5a9 100644 --- a/api/disaggregated/v1/types.go +++ b/api/disaggregated/v1/types.go @@ -23,6 +23,11 @@ import ( ) 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 @@ type DorisDisaggregatedClusterList struct { 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 02162b73..bd829f4a 100644 --- a/api/disaggregated/v1/zz_generated.deepcopy.go +++ b/api/disaggregated/v1/zz_generated.deepcopy.go @@ -289,6 +289,11 @@ func (in *DorisDisaggregatedClusterList) DeepCopyObject() runtime.Object { // 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 { @@ -566,6 +571,21 @@ func (in *ReadinessProbePolicy) DeepCopy() *ReadinessProbePolicy { return out } +// 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 f533e3fd..cb691198 100644 --- a/api/doris/v1/types.go +++ b/api/doris/v1/types.go @@ -30,6 +30,11 @@ var ( // 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 @@ type DorisClusterList struct { 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 fdcc5be8..0436d776 100644 --- a/api/doris/v1/zz_generated.deepcopy.go +++ b/api/doris/v1/zz_generated.deepcopy.go @@ -417,6 +417,11 @@ func (in *DorisClusterList) DeepCopyObject() runtime.Object { // 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) @@ -938,6 +943,21 @@ func (in *ResourceMetricSource) DeepCopy() *ResourceMetricSource { return out } +// 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 86cc968f..bdbd96f8 100644 --- a/config/crd/bases/crds.yaml +++ b/config/crd/bases/crds.yaml @@ -9239,6 +9239,21 @@ spec: 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 @@ spec: 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 0547f69d..ad81c18c 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 @@ spec: 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 eb9b96e6..c02e110d 100644 --- a/config/crd/bases/doris.apache.com_dorisclusters.yaml +++ b/config/crd/bases/doris.apache.com_dorisclusters.yaml @@ -9239,6 +9239,21 @@ spec: 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 eb9b96e6..c02e110d 100644 --- a/config/crd/bases/doris.selectdb.com_dorisclusters.yaml +++ b/config/crd/bases/doris.selectdb.com_dorisclusters.yaml @@ -9239,6 +9239,21 @@ spec: 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 00000000..51def1e2 --- /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/` 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/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml b/helm-charts/doris-operator/crds/disaggregated.cluster.doris.com_dorisdisaggregatedclusters.yaml index 0547f69d..ad81c18c 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 @@ spec: 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 eb9b96e6..c02e110d 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 @@ spec: 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/runtimeconfig/pod.go b/pkg/common/utils/runtimeconfig/pod.go new file mode 100644 index 00000000..c24e5cf3 --- /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 00000000..97d79149 --- /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 00000000..4102bd27 --- /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 00000000..553027c4 --- /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 69b753d6..a29bac96 100644 --- a/pkg/controller/disaggregated_cluster_controller.go +++ b/pkg/controller/disaggregated_cluster_controller.go @@ -26,6 +26,7 @@ import ( 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 @@ func (dc *DisaggregatedClusterReconciler) Reconcile(ctx context.Context, req rec 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 a5961e04..958d0444 100644 --- a/pkg/controller/doriscluster_controller.go +++ b/pkg/controller/doriscluster_controller.go @@ -38,6 +38,7 @@ import ( 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 @@ func (r *DorisClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request 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 ae6772b3..84fab24f 100644 --- a/pkg/controller/sub_controller/be/controller.go +++ b/pkg/controller/sub_controller/be/controller.go @@ -22,6 +22,7 @@ import ( "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 @@ func (be *Controller) Sync(ctx context.Context, dcr *v1.DorisCluster) error { } 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 bb3bdd32..5561e6c0 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 @@ import ( "encoding/json" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "regexp" "strconv" "strings" @@ -151,6 +152,9 @@ func (dcgs *DisaggregatedComputeGroupsController) computeGroupSync(ctx context.C } 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) @@ -689,6 +693,13 @@ func (dcgs *DisaggregatedComputeGroupsController) ClearStatefulsetUnusedPVCs(ctx 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 87249a4f..c4e15f6f 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 @@ import ( 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 newTestStatefulSet(namespace, name, storage string) *appv1.StatefulSet { }, } } + +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/disaggregated_fe/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go index f9c60d44..eecd92b9 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 @@ import ( "context" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "strconv" "strings" @@ -87,6 +88,9 @@ func (dfc *DisaggregatedFEController) Sync(ctx context.Context, obj client.Objec 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 a7a3936a..407dc9d0 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 @@ import ( "context" "errors" "fmt" + "github.com/apache/doris-operator/pkg/common/utils/runtimeconfig" "strconv" "strings" @@ -169,6 +170,9 @@ func (dms *DisaggregatedMSController) Sync(ctx context.Context, obj client.Objec 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 6b10d8b0..cebc800e 100644 --- a/pkg/controller/sub_controller/fe/controller.go +++ b/pkg/controller/sub_controller/fe/controller.go @@ -22,6 +22,7 @@ import ( 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 @@ func (fc *Controller) Sync(ctx context.Context, cluster *v1.DorisCluster) error } 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 00000000..e7748cf3 --- /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 00000000..39213f38 --- /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()