[Enhancement] Add safe operator upgrade controls

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.

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
 }

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

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),

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

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
+      terminationGracePeriodSeconds: 10

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

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 <namespace> annotate doriscluster <name> \
+  apache.org.doris/reconcile-paused=true --overwrite
+
+kubectl -n <namespace> annotate doriscluster <name> \
+  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.

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 {

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