通过Karmada实现应用多Kubernetes集群管理--控制器

func Run(ctx context.Context, opts *options.Options) error {
   klog.Infof("karmada-controller-manager version: %s", version.Get())
   config, err := controllerruntime.GetConfig()
   if err != nil {
      panic(err)
   }
   config.QPS, config.Burst = opts.KubeAPIQPS, opts.KubeAPIBurst
   controllerManager, err := controllerruntime.NewManager(config, controllerruntime.Options{
      Scheme:                     gclient.NewSchema(),
      SyncPeriod:                 &opts.ResyncPeriod.Duration,
      LeaderElection:             opts.LeaderElection.LeaderElect,
      LeaderElectionID:           opts.LeaderElection.ResourceName,
      LeaderElectionNamespace:    opts.LeaderElection.ResourceNamespace,
      LeaderElectionResourceLock: opts.LeaderElection.ResourceLock,
      HealthProbeBindAddress:     net.JoinHostPort(opts.BindAddress, strconv.Itoa(opts.SecurePort)),
      LivenessEndpointName:       "/healthz",
      MetricsBindAddress:         opts.MetricsBindAddress,
      Controller: v1alpha1.ControllerConfigurationSpec{
         GroupKindConcurrency: map[string]int{
            workv1alpha1.SchemeGroupVersion.WithKind("Work").GroupKind().String():                     opts.ConcurrentWorkSyncs,
            workv1alpha2.SchemeGroupVersion.WithKind("ResourceBinding").GroupKind().String():          opts.ConcurrentResourceBindingSyncs,
            workv1alpha2.SchemeGroupVersion.WithKind("ClusterResourceBinding").GroupKind().String():   opts.ConcurrentClusterResourceBindingSyncs,
            clusterv1alpha1.SchemeGroupVersion.WithKind("Cluster").GroupKind().String():               opts.ConcurrentClusterSyncs,
            schema.GroupVersionKind{Group: "", Version: "v1", Kind: "Namespace"}.GroupKind().String(): opts.ConcurrentNamespaceSyncs,
         },
      },
   })
   if err != nil {
      klog.Errorf("failed to build controller manager: %v", err)
      return err
   }

   if err := controllerManager.AddHealthzCheck("ping", healthz.Ping); err != nil {
      klog.Errorf("failed to add health check endpoint: %v", err)
      return err
   }

   setupControllers(controllerManager, opts, ctx.Done())

   // blocks until the context is done.
   if err := controllerManager.Start(ctx); err != nil {
      klog.Errorf("controller manager exits unexpectedly: %v", err)
      return err
   }

   // never reach here
   return nil
}

func setupControllers(mgr controllerruntime.Manager, opts *options.Options, stopChan <-chan struct{}) {
   restConfig := mgr.GetConfig()
   dynamicClientSet := dynamic.NewForConfigOrDie(restConfig)
   discoverClientSet := discovery.NewDiscoveryClientForConfigOrDie(restConfig)

   overrideManager := overridemanager.New(mgr.GetClient())
   skippedResourceConfig := util.NewSkippedResourceConfig()
   if err := skippedResourceConfig.Parse(opts.SkippedPropagatingAPIs); err != nil {
      // The program will never go here because the parameters have been checked
      return
   }

   skippedPropagatingNamespaces := map[string]struct{}{}
   for _, ns := range opts.SkippedPropagatingNamespaces {
      skippedPropagatingNamespaces[ns] = struct{}{}
   }

   controlPlaneInformerManager := informermanager.NewSingleClusterInformerManager(dynamicClientSet, 0, stopChan)

   resourceInterpreter := resourceinterpreter.NewResourceInterpreter("", controlPlaneInformerManager)
   if err := mgr.Add(resourceInterpreter); err != nil {
      klog.Fatalf("Failed to setup custom resource interpreter: %v", err)
   }

   objectWatcher := objectwatcher.NewObjectWatcher(mgr.GetClient(), mgr.GetRESTMapper(), util.NewClusterDynamicClientSet, resourceInterpreter)

   resourceDetector := &detector.ResourceDetector{
      DiscoveryClientSet:              discoverClientSet,
      Client:                          mgr.GetClient(),
      InformerManager:                 controlPlaneInformerManager,
      RESTMapper:                      mgr.GetRESTMapper(),
      DynamicClient:                   dynamicClientSet,
      SkippedResourceConfig:           skippedResourceConfig,
      SkippedPropagatingNamespaces:    skippedPropagatingNamespaces,
      ResourceInterpreter:             resourceInterpreter,
      EventRecorder:                   mgr.GetEventRecorderFor("resource-detector"),
      ConcurrentResourceTemplateSyncs: opts.ConcurrentResourceTemplateSyncs,
      ConcurrentResourceBindingSyncs:  opts.ConcurrentResourceBindingSyncs,
      RateLimiterOptions:              opts.RateLimiterOpts,
   }
   if err := mgr.Add(resourceDetector); err != nil {
      klog.Fatalf("Failed to setup resource detector: %v", err)
   }

   if features.FeatureGate.Enabled(features.PropagateDeps) {
      dependenciesDistributor := &dependenciesdistributor.DependenciesDistributor{
         Client:              mgr.GetClient(),
         DynamicClient:       dynamicClientSet,
         InformerManager:     controlPlaneInformerManager,
         ResourceInterpreter: resourceInterpreter,
         RESTMapper:          mgr.GetRESTMapper(),
      }
      if err := mgr.Add(dependenciesDistributor); err != nil {
         klog.Fatalf("Failed to setup dependencies distributor: %v", err)
      }
   }

   setupClusterAPIClusterDetector(mgr, opts, stopChan)
   controllerContext := controllerscontext.Context{
      Mgr:           mgr,
      ObjectWatcher: objectWatcher,
      Opts: controllerscontext.Options{
         Controllers:                       opts.Controllers,
         ClusterMonitorPeriod:              opts.ClusterMonitorPeriod,
         ClusterMonitorGracePeriod:         opts.ClusterMonitorGracePeriod,
         ClusterStartupGracePeriod:         opts.ClusterStartupGracePeriod,
         ClusterStatusUpdateFrequency:      opts.ClusterStatusUpdateFrequency,
         ClusterLeaseDuration:              opts.ClusterLeaseDuration,
         ClusterLeaseRenewIntervalFraction: opts.ClusterLeaseRenewIntervalFraction,
         ClusterCacheSyncTimeout:           opts.ClusterCacheSyncTimeout,
         ClusterAPIQPS:                     opts.ClusterAPIQPS,
         ClusterAPIBurst:                   opts.ClusterAPIBurst,
         SkippedPropagatingNamespaces:      opts.SkippedPropagatingNamespaces,
         ConcurrentWorkSyncs:               opts.ConcurrentWorkSyncs,
         RateLimiterOptions:                opts.RateLimiterOpts,
      },
      StopChan:                    stopChan,
      DynamicClientSet:            dynamicClientSet,
      OverrideManager:             overrideManager,
      ControlPlaneInformerManager: controlPlaneInformerManager,
      ResourceInterpreter:         resourceInterpreter,
   }

   if err := controllers.StartControllers(controllerContext); err != nil {
      klog.Fatalf("error starting controllers: %v", err)
   }

   // Ensure the InformerManager stops when the stop channel closes
   go func() {
      <-stopChan
      informermanager.StopInstance()
   }()
}