From 842cb9afecd20d812ba69e979de220a658332f80 Mon Sep 17 00:00:00 2001 From: Linying Assad Date: Sun, 20 Sep 2026 16:28:51 +0800 Subject: [PATCH] [Enhancement] Add safe operator upgrade controls --- api/doris/v1/types.go | 7 ++ cmd/operator/conf/flag.go | 26 +++-- cmd/operator/conf/flag_test.go | 54 +++++++++ cmd/operator/main.go | 2 +- config/operator/disaggregated-operator.yaml | 1 + config/operator/operator-sign-ns.yaml | 3 +- config/operator/operator.yaml | 1 + doc/operation/operator_upgrade_control.md | 30 +++++ pkg/controller/doriscluster_controller.go | 6 + .../doriscluster_controller_test.go | 109 ++++++++++++++++++ 10 files changed, 229 insertions(+), 10 deletions(-) create mode 100644 cmd/operator/conf/flag_test.go create mode 100644 doc/operation/operator_upgrade_control.md create mode 100644 pkg/controller/doriscluster_controller_test.go diff --git a/api/doris/v1/types.go b/api/doris/v1/types.go index f533e3fd..f7507fb2 100644 --- a/api/doris/v1/types.go +++ b/api/doris/v1/types.go @@ -28,6 +28,13 @@ var ( AnnotationDebugValue = "debug" ) +const ( + // AnnotationReconcilePaused pauses DorisCluster reconciliation. + AnnotationReconcilePaused = "apache.org.doris/reconcile-paused" + // AnnotationReconcilePausedLegacy preserves compatibility with older SelectDB releases. + AnnotationReconcilePausedLegacy = "selectdb.com.doris/reconcile-paused" +) + // DorisClusterSpec defines the desired state of DorisCluster type DorisClusterSpec struct { //defines the fe cluster state that will be created by operator. diff --git a/cmd/operator/conf/flag.go b/cmd/operator/conf/flag.go index d5d4d800..aa48936e 100644 --- a/cmd/operator/conf/flag.go +++ b/cmd/operator/conf/flag.go @@ -23,33 +23,43 @@ import ( "sigs.k8s.io/controller-runtime/pkg/log/zap" ) +const DefaultLeaderElectionID = "e1370669.selectdb.com" + // definate the start options. type Flag struct { MetricsAddr string ProbeAddr string Namespace string EnableLeaderElection bool + LeaderElectionID string PrintVar bool EnableWebhook bool Opts zap.Options } func ParseFlags() *Flag { + f := newFlag(flag.CommandLine) + flag.Parse() + return f +} + +func newFlag(flagSet *flag.FlagSet) *Flag { f := Flag{} - flag.StringVar(&f.MetricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.") - flag.StringVar(&f.ProbeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") - flag.StringVar(&f.Namespace, "namespace", v12.NamespaceAll, "The namespace to watch for changes.") - flag.BoolVar(&f.EnableLeaderElection, "leader-elect", false, + flagSet.StringVar(&f.MetricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.") + flagSet.StringVar(&f.ProbeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") + flagSet.StringVar(&f.Namespace, "namespace", v12.NamespaceAll, "The namespace to watch for changes.") + flagSet.BoolVar(&f.EnableLeaderElection, "leader-elect", false, "Enable leader election for controller manager. "+ "Enabling this will ensure there is only one active controller manager.") - flag.BoolVar(&f.PrintVar, "version", false, "Prints current version.") + flagSet.StringVar(&f.LeaderElectionID, "leader-election-id", DefaultLeaderElectionID, + "The name of the leader election lease. Operators with non-overlapping watch namespaces may use different values.") + flagSet.BoolVar(&f.PrintVar, "version", false, "Prints current version.") // check switch unnamedwatches on or off, if 'true' passed from console or config in env, will start unnamedwatches operator. - flag.BoolVar(&f.EnableWebhook, "enable-unnamedwatches", true, "start the unnamedwatches.") + flagSet.BoolVar(&f.EnableWebhook, "enable-unnamedwatches", true, "start the unnamedwatches.") f.Opts = zap.Options{ Development: true, } - f.Opts.BindFlags(flag.CommandLine) - flag.Parse() + f.Opts.BindFlags(flagSet) return &f } diff --git a/cmd/operator/conf/flag_test.go b/cmd/operator/conf/flag_test.go new file mode 100644 index 00000000..8bfaa576 --- /dev/null +++ b/cmd/operator/conf/flag_test.go @@ -0,0 +1,54 @@ +// 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 conf + +import ( + "flag" + "testing" +) + +func TestLeaderElectionIDFlag(t *testing.T) { + tests := []struct { + name string + args []string + want string + }{ + { + name: "default", + want: DefaultLeaderElectionID, + }, + { + name: "custom", + args: []string{"--leader-election-id=doris-operator-team-a"}, + want: "doris-operator-team-a", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + flagSet := flag.NewFlagSet(tt.name, flag.ContinueOnError) + parsed := newFlag(flagSet) + if err := flagSet.Parse(tt.args); err != nil { + t.Fatal(err) + } + if parsed.LeaderElectionID != tt.want { + t.Fatalf("LeaderElectionID = %q, want %q", parsed.LeaderElectionID, tt.want) + } + }) + } +} diff --git a/cmd/operator/main.go b/cmd/operator/main.go index 7779fa16..f8f422a2 100644 --- a/cmd/operator/main.go +++ b/cmd/operator/main.go @@ -133,7 +133,7 @@ func main() { }, WebhookServer: webhookServer, LeaderElection: f.EnableLeaderElection, - LeaderElectionID: "e1370669.selectdb.com", + LeaderElectionID: f.LeaderElectionID, //if one reconcile failed, others will not be affected. Controller: controllerconfig.Controller{ RecoverPanic: pointer.Bool(true), diff --git a/config/operator/disaggregated-operator.yaml b/config/operator/disaggregated-operator.yaml index 6641442d..64441e99 100644 --- a/config/operator/disaggregated-operator.yaml +++ b/config/operator/disaggregated-operator.yaml @@ -525,6 +525,7 @@ spec: - /dorisoperator args: - --leader-elect + - --leader-election-id=e1370669.selectdb.com image: apache/doris:operator-latest imagePullPolicy: Always name: dorisoperator diff --git a/config/operator/operator-sign-ns.yaml b/config/operator/operator-sign-ns.yaml index fa2005fd..51bd27bf 100644 --- a/config/operator/operator-sign-ns.yaml +++ b/config/operator/operator-sign-ns.yaml @@ -374,6 +374,7 @@ spec: - /dorisoperator args: - --leader-elect + - --leader-election-id=e1370669.selectdb.com - -namespace=doris image: apache/doris:operator-latest imagePullPolicy: Always @@ -425,4 +426,4 @@ spec: defaultMode: 420 secretName: doris-operator-secret-cert serviceAccountName: doris-operator - terminationGracePeriodSeconds: 10 \ No newline at end of file + terminationGracePeriodSeconds: 10 diff --git a/config/operator/operator.yaml b/config/operator/operator.yaml index 6adf837e..8318231e 100644 --- a/config/operator/operator.yaml +++ b/config/operator/operator.yaml @@ -517,6 +517,7 @@ spec: - /dorisoperator args: - --leader-elect + - --leader-election-id=e1370669.selectdb.com image: apache/doris:operator-latest imagePullPolicy: Always name: dorisoperator diff --git a/doc/operation/operator_upgrade_control.md b/doc/operation/operator_upgrade_control.md new file mode 100644 index 00000000..12260abe --- /dev/null +++ b/doc/operation/operator_upgrade_control.md @@ -0,0 +1,30 @@ +# Controlling DorisCluster reconciliation during an operator upgrade + +The operator supports pausing reconciliation for an individual `DorisCluster`. +Deletion cleanup is not paused, and removing the annotation triggers reconciliation again. + +```shell +kubectl -n annotate doriscluster \ + apache.org.doris/reconcile-paused=true --overwrite + +kubectl -n annotate doriscluster \ + apache.org.doris/reconcile-paused- +``` + +The legacy `selectdb.com.doris/reconcile-paused` annotation is also accepted. +This allows existing clusters to be resumed one at a time after an operator +upgrade while newly created, unannotated clusters continue to reconcile. + +The leader election lease name can be configured with +`--leader-election-id`. Its default remains `e1370669.selectdb.com`. When two +operators run in the same operator namespace, give them different IDs only if +their `--namespace` watch scopes do not overlap: + +```shell +/dorisoperator --leader-elect \ + --leader-election-id=doris-operator-team-a \ + --namespace=team-a +``` + +Overlapping watch scopes with different leader election IDs are unsafe because +both operators can reconcile the same resources. diff --git a/pkg/controller/doriscluster_controller.go b/pkg/controller/doriscluster_controller.go index a5961e04..8322ecaf 100644 --- a/pkg/controller/doriscluster_controller.go +++ b/pkg/controller/doriscluster_controller.go @@ -135,6 +135,12 @@ func (r *DorisClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request return ctrl.Result{}, nil } + if dcr.Annotations[dorisv1.AnnotationReconcilePaused] == "true" || + dcr.Annotations[dorisv1.AnnotationReconcilePausedLegacy] == "true" { + klog.Infof("DorisClusterReconciler skip paused DorisCluster namespace=%s name=%s", dcr.Namespace, dcr.Name) + return ctrl.Result{}, nil + } + if dcr.Spec.EnableRestartWhenConfigChange { coreConfigMaps := resource.GetDorisCoreConfigMapNames(dcr) for componentType := range coreConfigMaps { diff --git a/pkg/controller/doriscluster_controller_test.go b/pkg/controller/doriscluster_controller_test.go new file mode 100644 index 00000000..a29f6485 --- /dev/null +++ b/pkg/controller/doriscluster_controller_test.go @@ -0,0 +1,109 @@ +// 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 controller + +import ( + "context" + "testing" + + dorisv1 "github.com/apache/doris-operator/api/doris/v1" + "github.com/apache/doris-operator/pkg/controller/sub_controller" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +type recordingSubController struct { + syncCalls, clearCalls, statusCalls int +} + +func (r *recordingSubController) Sync(context.Context, *dorisv1.DorisCluster) error { + r.syncCalls++ + return nil +} + +func (r *recordingSubController) ClearResources(context.Context, *dorisv1.DorisCluster) (bool, error) { + r.clearCalls++ + return true, nil +} + +func (r *recordingSubController) GetControllerName() string { return "recording" } + +func (r *recordingSubController) UpdateComponentStatus(*dorisv1.DorisCluster) error { + r.statusCalls++ + return nil +} + +func newTestReconciler(t *testing.T, dcr *dorisv1.DorisCluster, recorder *recordingSubController) *DorisClusterReconciler { + t.Helper() + scheme := runtime.NewScheme() + if err := dorisv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + return &DorisClusterReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(dcr).Build(), + Scs: map[string]sub_controller.SubController{"recording": recorder}, + } +} + +func TestReconcilePausedDorisCluster(t *testing.T) { + for _, annotation := range []string{ + dorisv1.AnnotationReconcilePaused, + dorisv1.AnnotationReconcilePausedLegacy, + } { + t.Run(annotation, func(t *testing.T) { + dcr := &dorisv1.DorisCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "paused", Namespace: "default", + Annotations: map[string]string{annotation: "true"}, + }} + recorder := &recordingSubController{} + reconciler := newTestReconciler(t, dcr, recorder) + result, err := reconciler.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: dcr.Name, Namespace: dcr.Namespace}, + }) + if err != nil || result.Requeue || result.RequeueAfter != 0 { + t.Fatalf("Reconcile() result=%+v error=%v, want no requeue or error", result, err) + } + if recorder.syncCalls != 0 || recorder.clearCalls != 0 || recorder.statusCalls != 0 { + t.Fatalf("paused cluster called subcontrollers: %+v", recorder) + } + }) + } +} + +func TestPausedDorisClusterDeletionStillCleansResources(t *testing.T) { + now := metav1.Now() + dcr := &dorisv1.DorisCluster{ObjectMeta: metav1.ObjectMeta{ + Name: "deleting", Namespace: "default", DeletionTimestamp: &now, + Finalizers: []string{"test-finalizer"}, + Annotations: map[string]string{dorisv1.AnnotationReconcilePaused: "true"}, + }} + recorder := &recordingSubController{} + reconciler := newTestReconciler(t, dcr, recorder) + _, err := reconciler.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: dcr.Name, Namespace: dcr.Namespace}, + }) + if err != nil { + t.Fatal(err) + } + if recorder.clearCalls != 1 || recorder.syncCalls != 0 || recorder.statusCalls != 0 { + t.Fatalf("deleting paused cluster calls: %+v, want only one cleanup", recorder) + } +}