diff --git a/cmd/operator/conf/flag.go b/cmd/operator/conf/flag.go index d5d4d800..688eee73 100644 --- a/cmd/operator/conf/flag.go +++ b/cmd/operator/conf/flag.go @@ -28,6 +28,7 @@ type Flag struct { MetricsAddr string ProbeAddr string Namespace string + NamespaceSelector string EnableLeaderElection bool PrintVar bool EnableWebhook bool @@ -39,6 +40,7 @@ func ParseFlags() *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.StringVar(&f.NamespaceSelector, "namespace-selector", "", "Only reconcile namespaces matching this label selector; cannot be combined with --namespace.") flag.BoolVar(&f.EnableLeaderElection, "leader-elect", false, "Enable leader election for controller manager. "+ "Enabling this will ensure there is only one active controller manager.") diff --git a/cmd/operator/main.go b/cmd/operator/main.go index 7779fa16..613ca06a 100644 --- a/cmd/operator/main.go +++ b/cmd/operator/main.go @@ -41,6 +41,7 @@ import ( "github.com/apache/doris-operator/cmd/operator/conf" "github.com/apache/doris-operator/pkg/common/utils/certificate" "github.com/apache/doris-operator/pkg/controller" + "github.com/apache/doris-operator/pkg/controller/namespacescope" "github.com/apache/doris-operator/pkg/controller/unnamedwatches" "io" "k8s.io/apimachinery/pkg/util/wait" @@ -117,6 +118,14 @@ func main() { webhookServer := webhook.NewServer(webhook.Options{ Port: 9443, }) + if f.NamespaceSelector != "" && (f.Namespace != "" || envs.EnableWebhook) { + setupLog.Error(fmt.Errorf("namespace-selector requires global cache and ENABLE_WEBHOOK=false"), "invalid scope") + os.Exit(1) + } + if _, err := namespacescope.New(nil, f.NamespaceSelector); err != nil { + setupLog.Error(err, "invalid namespace-selector") + os.Exit(1) + } defaultNamespaces := map[string]cache.Config{} if f.Namespace != "" { defaultNamespaces[f.Namespace] = cache.Config{} @@ -157,6 +166,7 @@ func main() { } options := conf.NewControllerOptions(envs) + options.NamespaceScope, _ = namespacescope.New(mgr.GetAPIReader(), f.NamespaceSelector) //initial all controllers for _, c := range controller.Controllers { c.Init(mgr, options) diff --git a/doc/namespace-selector.md b/doc/namespace-selector.md new file mode 100644 index 00000000..8530ad52 --- /dev/null +++ b/doc/namespace-selector.md @@ -0,0 +1,45 @@ +# Sharing controllers across namespaces + +A controller can reconcile a dynamically selected group of namespaces: + +```yaml +# Controller Deployment +args: + - --leader-elect + - --namespace-selector=selectdb.com/operator-group=qa +env: + - name: ENABLE_WEBHOOK + value: "false" +``` + +Label business namespaces before creating Doris CRs: + +```yaml +apiVersion: v1 +kind: Namespace +metadata: + name: qa-run-123 + labels: + selectdb.com/operator-group: qa +``` + +Run another group with a disjoint selector and a separate controller namespace. +Groups must not share the default leader-election Lease. An unfiltered global +controller would still overlap all groups and must not run alongside them. + +`--namespace` and `--namespace-selector` are mutually exclusive. Selector mode +requires webhooks disabled and namespace get/list/watch permission, supplied by +the Helm Chart. The Chart exposes `dorisOperator.namespaceSelector` and disables +webhooks for this mode. Empty configuration preserves the existing global or +single-namespace behavior. + +The controller reads namespace labels before each Doris reconciliation. Read +errors stop reconciliation; removing the group label stops future reconciliations. +Namespace label events enqueue existing CRs, so a newly labeled namespace is +adopted without restarting the controller. Label selectors are Kubernetes selectors; +the platform deploying multiple controllers is responsible for making them disjoint. +Do not transfer namespaces between groups while CRs/finalizers remain. + +The global read cache remains available for explicit dependencies such as another +namespace's FDB connection ConfigMap. This filters reconciliation, not authorization; +use RBAC and separate clusters when read isolation or incompatible CRDs are required. diff --git a/helm-charts/doris-operator/templates/clusterrole.yaml b/helm-charts/doris-operator/templates/clusterrole.yaml index fe9b18db..94fe81df 100644 --- a/helm-charts/doris-operator/templates/clusterrole.yaml +++ b/helm-charts/doris-operator/templates/clusterrole.yaml @@ -21,6 +21,9 @@ metadata: creationTimestamp: null name: {{ template "kube-doris.name" . }}-operator rules: + - apiGroups: [""] + resources: [namespaces] + verbs: [get, list, watch] - apiGroups: - apps resources: diff --git a/helm-charts/doris-operator/templates/deployment.yaml b/helm-charts/doris-operator/templates/deployment.yaml index ea39ed89..bff56230 100644 --- a/helm-charts/doris-operator/templates/deployment.yaml +++ b/helm-charts/doris-operator/templates/deployment.yaml @@ -81,6 +81,9 @@ spec: - /dorisoperator args: - --leader-elect + {{- if .Values.dorisOperator.namespaceSelector }} + - {{ printf "--namespace-selector=%s" .Values.dorisOperator.namespaceSelector | quote }} + {{- end }} image: {{ .Values.dorisOperator.image.repository }}:{{ .Values.dorisOperator.image.tag }} {{- if .Values.dorisOperator.image.imagePullPolicy }} imagePullPolicy: {{ .Values.dorisOperator.image.imagePullPolicy }} @@ -93,7 +96,7 @@ spec: - "ALL" env: - name: ENABLE_WEBHOOK - value: "{{ template "webhook.enable" . }}" + value: {{ if .Values.dorisOperator.namespaceSelector }}"false"{{ else }}"{{ template "webhook.enable" . }}"{{ end }} - name: START_DISAGGREGATED_OPERATOR value: "true" - name: OPERATOR_NAMESPACE diff --git a/helm-charts/doris-operator/values.yaml b/helm-charts/doris-operator/values.yaml index f5b7dce7..9701a5bb 100644 --- a/helm-charts/doris-operator/values.yaml +++ b/helm-charts/doris-operator/values.yaml @@ -17,6 +17,8 @@ dorisOperator: + # Dynamic namespace label selection. Requires a build supporting --namespace-selector. + namespaceSelector: "" # the namespace of doris operator deployed, is not provided be set in values.yaml, please use the helm command specify, example "helm install -n doris" image: repository: apache/doris diff --git a/pkg/controller/disaggregated_cluster_controller.go b/pkg/controller/disaggregated_cluster_controller.go index a29bac96..e67ac923 100644 --- a/pkg/controller/disaggregated_cluster_controller.go +++ b/pkg/controller/disaggregated_cluster_controller.go @@ -27,6 +27,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" + "github.com/apache/doris-operator/pkg/controller/namespacescope" 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" @@ -58,6 +59,7 @@ var ( ) type DisaggregatedClusterReconciler struct { + NamespaceScope *namespacescope.Scope client.Client Recorder record.EventRecorder Scheme *runtime.Scheme @@ -75,9 +77,10 @@ func (dc *DisaggregatedClusterReconciler) Init(mgr ctrl.Manager, options *Option scs[dccsc.GetControllerName()] = dccsc if err := (&DisaggregatedClusterReconciler{ - Client: mgr.GetClient(), - Recorder: mgr.GetEventRecorderFor(disaggregatedClusterController), - Scs: scs, + NamespaceScope: options.NamespaceScope, + Client: mgr.GetClient(), + Recorder: mgr.GetEventRecorderFor(disaggregatedClusterController), + Scs: scs, }).SetupWithManager(mgr); err != nil { klog.Error(err, "unable to create controller ", "disaggregatedClusterReconciler") os.Exit(1) @@ -95,6 +98,9 @@ func (dc *DisaggregatedClusterReconciler) SetupWithManager(mgr ctrl.Manager) err builder := dc.resourceBuilder(ctrl.NewControllerManagedBy(mgr)) builder = dc.watchPodBuilder(builder) builder = dc.watchFDBConfigMapBuilder(builder) + if dc.NamespaceScope.Enabled() { + builder = namespacescope.Watch(builder, mgr.GetClient(), func() client.ObjectList { return &dv1.DorisDisaggregatedClusterList{} }) + } return builder.Complete(dc) } @@ -215,6 +221,9 @@ func (dc *DisaggregatedClusterReconciler) resourceBuilder(builder *ctrl.Builder) // 3. clear need delete resource. // 4. display new status(eorganize status, update cr or status) func (dc *DisaggregatedClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + if allowed, err := dc.NamespaceScope.Allows(ctx, req.Namespace); err != nil || !allowed { + return ctrl.Result{}, err + } var ddc dv1.DorisDisaggregatedCluster err := dc.Get(ctx, req.NamespacedName, &ddc) if apierrors.IsNotFound(err) { diff --git a/pkg/controller/doriscluster_controller.go b/pkg/controller/doriscluster_controller.go index 958d0444..8ed86927 100644 --- a/pkg/controller/doriscluster_controller.go +++ b/pkg/controller/doriscluster_controller.go @@ -39,6 +39,7 @@ import ( "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/namespacescope" "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" @@ -75,6 +76,7 @@ var ( // DorisClusterReconciler reconciles a DorisCluster object type DorisClusterReconciler struct { + NamespaceScope *namespacescope.Scope client.Client Recorder record.EventRecorder Scheme *runtime.Scheme @@ -116,6 +118,9 @@ var ( // For more details, check Reconcile and its Result here: // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.14.4/pkg/reconcile func (r *DorisClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + if allowed, err := r.NamespaceScope.Allows(ctx, req.Namespace); err != nil || !allowed { + return ctrl.Result{}, err + } klog.FromContext(ctx) klog.Info("DorisClusterReconciler reconcile the update crd name ", req.Name, " namespace ", req.Namespace) var edcr dorisv1.DorisCluster @@ -358,6 +363,9 @@ func (r *DorisClusterReconciler) SetupWithManager(mgr ctrl.Manager) error { builder := r.resourceBuilder(ctrl.NewControllerManagedBy(mgr)) builder = r.watchPodBuilder(builder) builder = r.watchConfigMapBuilder(builder) + if r.NamespaceScope.Enabled() { + builder = namespacescope.Watch(builder, mgr.GetClient(), func() client.ObjectList { return &dorisv1.DorisClusterList{} }) + } return builder.Complete(r) } @@ -374,6 +382,7 @@ func (r *DorisClusterReconciler) Init(mgr ctrl.Manager, options *Options) { subcs[brokerControllerName] = brk if err := (&DorisClusterReconciler{ + NamespaceScope: options.NamespaceScope, Client: mgr.GetClient(), Recorder: mgr.GetEventRecorderFor(name), Scs: subcs, diff --git a/pkg/controller/namespacescope/scope.go b/pkg/controller/namespacescope/scope.go new file mode 100644 index 00000000..ebbc8da1 --- /dev/null +++ b/pkg/controller/namespacescope/scope.go @@ -0,0 +1,97 @@ +// 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 namespacescope + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// Scope filters reconciliation, not read permissions. The global cache remains +// available for explicit cross-namespace dependencies such as FDB ConfigMaps. +type Scope struct { + reader client.Reader + selector labels.Selector +} + +func New(reader client.Reader, expression string) (*Scope, error) { + if expression == "" { + return nil, nil + } + selector, err := labels.Parse(expression) + if err != nil { + return nil, err + } + if selector.Empty() { + return nil, fmt.Errorf("namespace selector must not select every namespace") + } + return &Scope{reader: reader, selector: selector}, nil +} + +func (s *Scope) Enabled() bool { return s != nil } + +func (s *Scope) Allows(ctx context.Context, namespace string) (bool, error) { + if s == nil { + return true, nil + } + var ns corev1.Namespace + // Use the API reader: cached namespace labels must not retain ownership after + // an administrator removes the group label. Read failures never permit writes. + if err := s.reader.Get(ctx, types.NamespacedName{Name: namespace}, &ns); err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, err + } + return s.selector.Matches(labels.Set(ns.Labels)), nil +} + +// Watch also adopts existing CRs when a namespace joins the selected group. +func Watch(b *ctrl.Builder, c client.Client, list func() client.ObjectList) *ctrl.Builder { + return b.Watches(&corev1.Namespace{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, ns client.Object) []reconcile.Request { + objects := list() + if err := c.List(ctx, objects, client.InNamespace(ns.GetName())); err != nil { + ctrl.LoggerFrom(ctx).Error(err, "list namespace resources", "namespace", ns.GetName()) + return nil + } + items, err := meta.ExtractList(objects) + if err != nil { + return nil + } + requests := make([]reconcile.Request, 0, len(items)) + for _, item := range items { + obj, err := meta.Accessor(item) + if err == nil { + requests = append(requests, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: obj.GetNamespace(), Name: obj.GetName()}}) + } + } + return requests + }), builder.WithPredicates(predicate.LabelChangedPredicate{})) +} diff --git a/pkg/controller/namespacescope/scope_test.go b/pkg/controller/namespacescope/scope_test.go new file mode 100644 index 00000000..a13b1b0f --- /dev/null +++ b/pkg/controller/namespacescope/scope_test.go @@ -0,0 +1,78 @@ +// 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 namespacescope + +import ( + "context" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func TestNamespaceOwnershipChangesWithoutRestart(t *testing.T) { + scheme := runtime.NewScheme() + _ = corev1.AddToScheme(scheme) + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "run-1", Labels: map[string]string{"selectdb.com/operator-group": "qa"}}} + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ns).Build() + qa, err := New(c, "selectdb.com/operator-group=qa") + if err != nil { + t.Fatal(err) + } + hubspot, _ := New(c, "selectdb.com/operator-group=hubspot") + ctx := context.Background() + assertAllowed := func(scope *Scope, name string, want bool) { + t.Helper() + got, err := scope.Allows(ctx, name) + if err != nil || got != want { + t.Fatalf("Allows(%s) = %v, %v; want %v", name, got, err, want) + } + } + assertAllowed(qa, ns.Name, true) + assertAllowed(hubspot, ns.Name, false) + assertAllowed(qa, "missing", false) + ns.Labels = nil + if err := c.Update(ctx, ns); err != nil { + t.Fatal(err) + } + assertAllowed(qa, ns.Name, false) + ns.Labels = map[string]string{"selectdb.com/operator-group": "hubspot"} + if err := c.Update(ctx, ns); err != nil { + t.Fatal(err) + } + assertAllowed(qa, ns.Name, false) + assertAllowed(hubspot, ns.Name, true) +} + +func TestNamespaceSelectorValidationAndLegacyMode(t *testing.T) { + if _, err := New(nil, " "); err == nil { + t.Fatal("empty expression accepted") + } + if _, err := New(nil, "key in ("); err == nil { + t.Fatal("invalid selector accepted") + } + s, err := New(nil, "") + if err != nil || s.Enabled() { + t.Fatal("legacy mode changed") + } + if allowed, err := s.Allows(context.Background(), "any"); err != nil || !allowed { + t.Fatal("legacy mode must not read Namespace objects") + } +} diff --git a/pkg/controller/options.go b/pkg/controller/options.go index 3434a2db..1d9b8887 100644 --- a/pkg/controller/options.go +++ b/pkg/controller/options.go @@ -17,9 +17,12 @@ package controller +import "github.com/apache/doris-operator/pkg/controller/namespacescope" + // operator start options type Options struct { - EnableWebHook bool + EnableWebHook bool + NamespaceScope *namespacescope.Scope // the operator name Name string //the secret name