add selector for nebula objects

cmd/controller-manager/app/controller-manager.go

@@ -23,6 +23,7 @@ import (
 
 	kruisev1beta1 "github.com/openkruise/kruise-api/apps/v1beta1"
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -142,6 +143,16 @@ func Run(ctx context.Context, opts *options.Options) error {
 			TLSMinVersion: opts.WebhookOpts.TLSMinVersion,
 		})
 	}
+
+	if opts.NebulaSelector != "" {
+		parsedSelector, err := labels.Parse(opts.NebulaSelector)
+		if err != nil {
+			klog.Errorf("couldn't convert selector into a corresponding internal selector object: %v", err)
+			return err
+		}
+		ctrlOptions.Cache.DefaultLabelSelector = parsedSelector
+	}
+
 	mgr, err := ctrlruntime.NewManager(cfg, ctrlOptions)
 	if err != nil {
 		klog.Errorf("Failed to build controller manager: %v", err)

cmd/controller-manager/app/options/options.go

@@ -51,6 +51,9 @@ type Options struct {
 	// Default watches all namespaces
 	Namespaces []string
 
+	// NebulaSelector to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2)
+	NebulaSelector string
+
 	// MetricsBindAddress is the TCP address that the controller should bind to
 	// for serving prometheus metrics.
 	// It can be set to "0" to disable the metrics serving.
@@ -111,6 +114,7 @@ func (o *Options) AddFlags(flags *pflag.FlagSet) {
 
 	flags.DurationVar(&o.SyncPeriod.Duration, "sync-period", 0, "Period at which the controller forces the repopulation of its local object stores.")
 	flags.StringSliceVar(&o.Namespaces, "watch-namespaces", nil, "Namespaces restricts the controller watches for updates to Kubernetes objects. If empty, all namespaces are watched. Multiple namespaces seperated by comma.(e.g. ns1,ns2,ns3).")
+	flags.StringVar(&o.NebulaSelector, "nebula-object-selector", "", "nebula object selector (label query) to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2).")
 	flags.StringVar(&o.MetricsBindAddress, "metrics-bind-address", ":8080", "The TCP address that the controller should bind to for serving prometheus metrics(e.g. 127.0.0.1:8080, :8080). It can be set to \"0\" to disable the metrics serving.")
 	flags.StringVar(&o.HealthProbeBindAddress, "health-probe-bind-address", ":8081", "The TCP address that the controller should bind to for serving health probes.(e.g. 127.0.0.1:8081, :8081). It can be set to \"0\" to disable the health probe serving.")
 	flags.IntVar(&o.ConcurrentNebulaClusterSyncs, "concurrent-nebulacluster-syncs", 5, "The number of NebulaCluster objects that are allowed to sync concurrently.")

pkg/controller/nebularestore/nebula_restore_controller.go

@@ -24,7 +24,6 @@ import (
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/vesoft-inc/nebula-operator/apis/apps/v1alpha1"
@@ -116,6 +115,5 @@ func (r *Reconciler) syncNebulaRestore(restore *v1alpha1.NebulaRestore) error {
 func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.NebulaRestore{}).
-		WithOptions(controller.Options{MaxConcurrentReconciles: 5}).
 		Complete(r)
 }

pkg/kube/rbac.go

@@ -69,9 +69,14 @@ func createClusterRole(ctx context.Context, k8sClient client.Client) error {
 	if err := k8sClient.Get(ctx, client.ObjectKey{Name: v1alpha1.NebulaRoleName}, &rbacv1.ClusterRole{}); err != nil {
 		if apierrors.IsNotFound(err) {
 			if err := k8sClient.Create(ctx, &role); err != nil {
+				if apierrors.IsAlreadyExists(err) {
+					return nil
+				}
 				return fmt.Errorf("failed to create ClusterRole role: %v", err)
 			}
+			return nil
 		}
+		return err
 	}
 	return nil
 }
@@ -90,9 +95,14 @@ func createClusterRoleBinding(ctx context.Context, k8sClient client.Client, name
 	if err := k8sClient.Get(ctx, client.ObjectKey{Name: v1alpha1.NebulaRoleBindingName}, binding); err != nil {
 		if apierrors.IsNotFound(err) {
 			if err := k8sClient.Create(ctx, binding); err != nil {
+				if apierrors.IsAlreadyExists(err) {
+					return nil
+				}
 				return fmt.Errorf("failed to create ClusterRoleBinding: %v", err)
 			}
+			return nil
 		}
+		return err
 	}
 	if !isApplied(binding.Subjects, namespace) {
 		binding.Subjects = append(binding.Subjects, rbacv1.Subject{
@@ -119,9 +129,14 @@ func createServiceAccount(ctx context.Context, k8sClient client.Client, namespac
 	if err := k8sClient.Get(ctx, client.ObjectKey{Name: v1alpha1.NebulaServiceAccountName, Namespace: namespace}, &corev1.ServiceAccount{}); err != nil {
 		if apierrors.IsNotFound(err) {
 			if err := k8sClient.Create(ctx, &serviceAccount); err != nil {
+				if apierrors.IsAlreadyExists(err) {
+					return nil
+				}
 				return fmt.Errorf("failed to create ServiceAccount: %v", err)
 			}
+			return nil
 		}
+		return err
 	}
 	return nil
 }