Merge pull request #576 from Garrybest/enhance-lease

Leader election: clean up and add leader election to agent

cmd/agent/app/agent.go

@@ -67,10 +67,12 @@ func run(ctx context.Context, karmadaConfig karmadactl.KarmadaConfig, opts *opti
 	}
 
 	controllerManager, err := controllerruntime.NewManager(controlPlaneRestConfig, controllerruntime.Options{
-		Scheme:           gclient.NewSchema(),
+		Scheme:                     gclient.NewSchema(),
-		Namespace:        executionSpace,
+		Namespace:                  executionSpace,
-		LeaderElection:   false,
+		LeaderElection:             opts.LeaderElection.LeaderElect,
-		LeaderElectionID: "agent.karmada.io",
+		LeaderElectionID:           fmt.Sprintf("karmada-agent-%s", opts.ClusterName),
+		LeaderElectionNamespace:    opts.LeaderElection.ResourceNamespace,
+		LeaderElectionResourceLock: opts.LeaderElection.ResourceLock,
 	})
 	if err != nil {
 		klog.Errorf("failed to build controller manager: %v", err)

cmd/agent/app/options/options.go

@@ -5,10 +5,15 @@ import (
 
 	"github.com/spf13/pflag"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	componentbaseconfig "k8s.io/component-base/config"
+
+	"github.com/karmada-io/karmada/pkg/util"
 )
 
 // Options contains everything necessary to create and run controller-manager.
 type Options struct {
+	LeaderElection    componentbaseconfig.LeaderElectionConfiguration
 	KarmadaKubeConfig string
 	// ClusterContext is the name of the cluster context in control plane KUBECONFIG file.
 	// Default value is the current-context.
@@ -28,7 +33,13 @@ type Options struct {
 
 // NewOptions builds an default scheduler options.
 func NewOptions() *Options {
-	return &Options{}
+	return &Options{
+		LeaderElection: componentbaseconfig.LeaderElectionConfiguration{
+			LeaderElect:       true,
+			ResourceLock:      resourcelock.LeasesResourceLock,
+			ResourceNamespace: util.NamespaceKarmadaSystem,
+		},
+	}
 }
 
 // AddFlags adds flags of scheduler to the specified FlagSet
@@ -37,6 +48,7 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 		return
 	}
 
+	fs.BoolVar(&o.LeaderElection.LeaderElect, "leader-elect", true, "Start a leader election client and gain leadership before executing the main loop. Enable this when running replicated components for high availability.")
 	fs.StringVar(&o.KarmadaKubeConfig, "karmada-kubeconfig", o.KarmadaKubeConfig, "Path to karmada control plane kubeconfig file.")
 	fs.StringVar(&o.KarmadaContext, "karmada-context", "", "Name of the cluster context in karmada control plane kubeconfig file.")
 	fs.StringVar(&o.ClusterName, "cluster-name", o.ClusterName, "Name of member cluster that the agent serves for.")

cmd/controller-manager/app/controllermanager.go

@@ -44,7 +44,6 @@ func NewControllerManagerCommand(ctx context.Context) *cobra.Command {
 		Use:  "controller-manager",
 		Long: `The controller manager runs a bunch of controllers`,
 		Run: func(cmd *cobra.Command, args []string) {
-			opts.Complete()
 			if err := Run(ctx, opts); err != nil {
 				fmt.Fprintf(os.Stderr, "%v\n", err)
 				os.Exit(1)
@@ -64,11 +63,13 @@ func Run(ctx context.Context, opts *options.Options) error {
 		panic(err)
 	}
 	controllerManager, err := controllerruntime.NewManager(config, controllerruntime.Options{
-		Scheme:                 gclient.NewSchema(),
+		Scheme:                     gclient.NewSchema(),
-		LeaderElection:         opts.LeaderElection.LeaderElect,
+		LeaderElection:             opts.LeaderElection.LeaderElect,
-		LeaderElectionID:       "karmada-controller-manager",
+		LeaderElectionID:           opts.LeaderElection.ResourceName,
-		HealthProbeBindAddress: fmt.Sprintf("%s:%d", opts.BindAddress, opts.SecurePort),
+		LeaderElectionNamespace:    opts.LeaderElection.ResourceNamespace,
-		LivenessEndpointName:   "/healthz",
+		LeaderElectionResourceLock: opts.LeaderElection.ResourceLock,
+		HealthProbeBindAddress:     fmt.Sprintf("%s:%d", opts.BindAddress, opts.SecurePort),
+		LivenessEndpointName:       "/healthz",
 	})
 	if err != nil {
 		klog.Errorf("failed to build controller manager: %v", err)

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

@@ -7,13 +7,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	componentbaseconfig "k8s.io/component-base/config"
-	"k8s.io/klog/v2"
-)
 
-var (
+	"github.com/karmada-io/karmada/pkg/util"
-	defaultElectionLeaseDuration = metav1.Duration{Duration: 15 * time.Second}
-	defaultElectionRenewDeadline = metav1.Duration{Duration: 10 * time.Second}
-	defaultElectionRetryPeriod   = metav1.Duration{Duration: 2 * time.Second}
 )
 
 const (
@@ -23,7 +18,6 @@ const (
 
 // Options contains everything necessary to create and run controller-manager.
 type Options struct {
-	HostNamespace  string
 	LeaderElection componentbaseconfig.LeaderElectionConfiguration
 	// BindAddress is the IP address on which to listen for the --secure-port port.
 	BindAddress string
@@ -57,34 +51,13 @@ type Options struct {
 
 // NewOptions builds an empty options.
 func NewOptions() *Options {
-	return &Options{}
+	return &Options{
-}
+		LeaderElection: componentbaseconfig.LeaderElectionConfiguration{
-
+			LeaderElect:       true,
-// Complete fills in any fields not set that are required to have valid data. It's mutating the receiver.
+			ResourceLock:      resourcelock.LeasesResourceLock,
-func (o *Options) Complete() {
+			ResourceNamespace: util.NamespaceKarmadaSystem,
-	if len(o.HostNamespace) == 0 {
+			ResourceName:      "karmada-controller-manager",
-		o.HostNamespace = "default"
+		},
-		klog.Infof("Set default value: Options.HostNamespace = %s", "default")
-	}
-
-	if len(o.LeaderElection.ResourceLock) == 0 {
-		o.LeaderElection.ResourceLock = resourcelock.EndpointsLeasesResourceLock
-		klog.Infof("Set default value: Options.LeaderElection.ResourceLock = %s", resourcelock.EndpointsLeasesResourceLock)
-	}
-
-	if o.LeaderElection.LeaseDuration.Duration.Seconds() == 0 {
-		o.LeaderElection.LeaseDuration = defaultElectionLeaseDuration
-		klog.Infof("Set default value: Options.LeaderElection.LeaseDuration = %s", defaultElectionLeaseDuration.Duration.String())
-	}
-
-	if o.LeaderElection.RenewDeadline.Duration.Seconds() == 0 {
-		o.LeaderElection.RenewDeadline = defaultElectionRenewDeadline
-		klog.Infof("Set default value: Options.LeaderElection.RenewDeadline = %s", defaultElectionRenewDeadline.Duration.String())
-	}
-
-	if o.LeaderElection.RetryPeriod.Duration.Seconds() == 0 {
-		o.LeaderElection.RetryPeriod = defaultElectionRetryPeriod
-		klog.Infof("Set default value: Options.LeaderElection.RetryPeriod = %s", defaultElectionRetryPeriod.Duration.String())
 	}
 }
 

cmd/scheduler/app/options/options.go

@@ -7,6 +7,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	componentbaseconfig "k8s.io/component-base/config"
+
+	"github.com/karmada-io/karmada/pkg/util"
 )
 
 const (
@@ -38,11 +40,13 @@ type Options struct {
 func NewOptions() *Options {
 	return &Options{
 		LeaderElection: componentbaseconfig.LeaderElectionConfiguration{
-			LeaderElect:   false,
+			LeaderElect:       true,
-			ResourceLock:  resourcelock.LeasesResourceLock,
+			ResourceLock:      resourcelock.LeasesResourceLock,
-			LeaseDuration: defaultElectionLeaseDuration,
+			ResourceNamespace: util.NamespaceKarmadaSystem,
-			RenewDeadline: defaultElectionRenewDeadline,
+			ResourceName:      "karmada-scheduler",
-			RetryPeriod:   defaultElectionRetryPeriod,
+			LeaseDuration:     defaultElectionLeaseDuration,
+			RenewDeadline:     defaultElectionRenewDeadline,
+			RetryPeriod:       defaultElectionRetryPeriod,
 		},
 	}
 }
@@ -53,8 +57,7 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 		return
 	}
 
-	fs.BoolVar(&o.LeaderElection.LeaderElect, "leader-elect", false, "Enable leader election, which must be true when running multi instances.")
+	fs.BoolVar(&o.LeaderElection.LeaderElect, "leader-elect", true, "Enable leader election, which must be true when running multi instances.")
-	fs.StringVar(&o.LeaderElection.ResourceNamespace, "lock-namespace", "", "Define the namespace of the lock object.")
 	fs.StringVar(&o.KubeConfig, "kubeconfig", o.KubeConfig, "Path to a KubeConfig. Only required if out-of-cluster.")
 	fs.StringVar(&o.Master, "master", o.Master, "The address of the Kubernetes API server. Overrides any value in KubeConfig. Only required if out-of-cluster.")
 	fs.StringVar(&o.BindAddress, "bind-address", defaultBindAddress, "The IP address on which to listen for the --secure-port port.")

cmd/scheduler/app/scheduler.go

@@ -80,7 +80,7 @@ func run(opts *options.Options, stopChan <-chan struct{}) error {
 
 	rl, err := resourcelock.New(opts.LeaderElection.ResourceLock,
 		opts.LeaderElection.ResourceNamespace,
-		"karmada-scheduler",
+		opts.LeaderElection.ResourceName,
 		leaderElectionClient.CoreV1(),
 		leaderElectionClient.CoordinationV1(),
 		resourcelock.ResourceLockConfig{

cmd/webhook/app/options/options.go

@@ -1,19 +1,7 @@
 package options
 
 import (
-	"time"
-
 	"github.com/spf13/pflag"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/tools/leaderelection/resourcelock"
-	componentbaseconfig "k8s.io/component-base/config"
-	"k8s.io/klog/v2"
-)
-
-var (
-	defaultElectionLeaseDuration = metav1.Duration{Duration: 15 * time.Second}
-	defaultElectionRenewDeadline = metav1.Duration{Duration: 10 * time.Second}
-	defaultElectionRetryPeriod   = metav1.Duration{Duration: 2 * time.Second}
 )
 
 const (
@@ -33,8 +21,7 @@ type Options struct {
 	// CertDir is the directory that contains the server key and certificate.
 	// if not set, webhook server would look up the server key and certificate in {TempDir}/k8s-webhook-server/serving-certs.
 	// The server key and certificate must be named `tls.key` and `tls.crt`, respectively.
-	CertDir        string
+	CertDir string
-	LeaderElection componentbaseconfig.LeaderElectionConfiguration
 }
 
 // NewOptions builds an empty options.
@@ -42,29 +29,6 @@ func NewOptions() *Options {
 	return &Options{}
 }
 
-// Complete fills in any fields not set that are required to have valid data. It's mutating the receiver.
-func (o *Options) Complete() {
-	if len(o.LeaderElection.ResourceLock) == 0 {
-		o.LeaderElection.ResourceLock = resourcelock.EndpointsLeasesResourceLock
-		klog.Infof("Set default value: Options.LeaderElection.ResourceLock = %s", resourcelock.EndpointsLeasesResourceLock)
-	}
-
-	if o.LeaderElection.LeaseDuration.Duration.Seconds() == 0 {
-		o.LeaderElection.LeaseDuration = defaultElectionLeaseDuration
-		klog.Infof("Set default value: Options.LeaderElection.LeaseDuration = %s", defaultElectionLeaseDuration.Duration.String())
-	}
-
-	if o.LeaderElection.RenewDeadline.Duration.Seconds() == 0 {
-		o.LeaderElection.RenewDeadline = defaultElectionRenewDeadline
-		klog.Infof("Set default value: Options.LeaderElection.RenewDeadline = %s", defaultElectionRenewDeadline.Duration.String())
-	}
-
-	if o.LeaderElection.RetryPeriod.Duration.Seconds() == 0 {
-		o.LeaderElection.RetryPeriod = defaultElectionRetryPeriod
-		klog.Infof("Set default value: Options.LeaderElection.RetryPeriod = %s", defaultElectionRetryPeriod.Duration.String())
-	}
-}
-
 // AddFlags adds flags to the specified FlagSet.
 func (o *Options) AddFlags(flags *pflag.FlagSet) {
 	flags.StringVar(&o.BindAddress, "bind-address", defaultBindAddress,

cmd/webhook/app/webhook.go

@@ -8,7 +8,6 @@ import (
 	"os"
 
 	"github.com/spf13/cobra"
-	"k8s.io/component-base/logs"
 	"k8s.io/klog/v2"
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
@@ -31,7 +30,6 @@ func NewWebhookCommand(ctx context.Context) *cobra.Command {
 		Use:  "webhook",
 		Long: `Start a webhook server`,
 		Run: func(cmd *cobra.Command, args []string) {
-			opts.Complete()
 			if err := Run(ctx, opts); err != nil {
 				fmt.Fprintf(os.Stderr, "%v\n", err)
 				os.Exit(1)
@@ -47,20 +45,16 @@ func NewWebhookCommand(ctx context.Context) *cobra.Command {
 
 // Run runs the webhook server with options. This should never exit.
 func Run(ctx context.Context, opts *options.Options) error {
-	logs.InitLogs()
-	defer logs.FlushLogs()
-
 	config, err := controllerruntime.GetConfig()
 	if err != nil {
 		panic(err)
 	}
 	hookManager, err := controllerruntime.NewManager(config, controllerruntime.Options{
-		Scheme:           gclient.NewSchema(),
+		Scheme:         gclient.NewSchema(),
-		Host:             opts.BindAddress,
+		Host:           opts.BindAddress,
-		Port:             opts.SecurePort,
+		Port:           opts.SecurePort,
-		CertDir:          opts.CertDir,
+		CertDir:        opts.CertDir,
-		LeaderElection:   false,
+		LeaderElection: false,
-		LeaderElectionID: "webhook.karmada.io",
 	})
 	if err != nil {
 		klog.Errorf("failed to build webhook server: %v", err)

pkg/util/constants.go

@@ -96,3 +96,8 @@ const (
 	// PropagationInstructionSuppressed indicates that the resource should not be propagated.
 	PropagationInstructionSuppressed = "suppressed"
 )
+
+const (
+	// NamespaceKarmadaSystem is the karmada system namespace.
+	NamespaceKarmadaSystem = "karmada-system"
+)