@@ -28,11 +28,13 @@ import (
2828 "sigs.k8s.io/controller-runtime/pkg/healthz"
2929 "sigs.k8s.io/controller-runtime/pkg/log/zap"
3030 "sigs.k8s.io/controller-runtime/pkg/manager"
31+ metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
3132 "sigs.k8s.io/controller-runtime/pkg/webhook"
3233 gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
3334 gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
3435 gatewayv1alpha3 "sigs.k8s.io/gateway-api/apis/v1alpha3"
3536 mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
37+ "sigs.k8s.io/multicluster-runtime/pkg/manager/coordinator/sharded"
3638 "sigs.k8s.io/multicluster-runtime/pkg/multicluster"
3739 mcsingle "sigs.k8s.io/multicluster-runtime/providers/single"
3840
@@ -76,6 +78,12 @@ func main() {
7678 var enableLeaderElection bool
7779 var leaderElectionNamespace string
7880 var probeAddr string
81+ var enableClusterSharding bool
82+ var clusterShardingLeaseNamespace string
83+ var clusterShardingLeasePrefix string
84+ var clusterShardingPeerWeight uint
85+ var singletonControllersLeaderElection bool
86+ var singletonControllersLeaderElectionID string
7987
8088 var serverConfigFile string
8189
@@ -84,6 +92,42 @@ func main() {
8492 "Enable leader election for controller manager. " +
8593 "Enabling this will ensure there is only one active controller manager." )
8694 flag .StringVar (& leaderElectionNamespace , "leader-elect-namespace" , "" , "The namespace to use for leader election." )
95+ flag .BoolVar (
96+ & enableClusterSharding ,
97+ "cluster-sharding-enabled" ,
98+ false ,
99+ "Enable multicluster controller sharding via per-cluster coordination leases." ,
100+ )
101+ flag .StringVar (
102+ & clusterShardingLeaseNamespace ,
103+ "cluster-sharding-lease-namespace" ,
104+ "kube-system" ,
105+ "Namespace for controller cluster sharding leases." ,
106+ )
107+ flag .StringVar (
108+ & clusterShardingLeasePrefix ,
109+ "cluster-sharding-lease-prefix" ,
110+ "mcr-shard" ,
111+ "Lease name prefix for controller cluster sharding." ,
112+ )
113+ flag .UintVar (
114+ & clusterShardingPeerWeight ,
115+ "cluster-sharding-peer-weight" ,
116+ 1 ,
117+ "Relative shard weight for this controller instance." ,
118+ )
119+ flag .BoolVar (
120+ & singletonControllersLeaderElection ,
121+ "singleton-controllers-leader-elect" ,
122+ true ,
123+ "Enable leader election for singleton downstream controllers (Challenge and GatewayDownstreamCertificateSolver)." ,
124+ )
125+ flag .StringVar (
126+ & singletonControllersLeaderElectionID ,
127+ "singleton-controllers-leader-election-id" ,
128+ "6a7d51cc.datumapis.com-singleton" ,
129+ "Leader election ID for singleton downstream controllers." ,
130+ )
87131
88132 opts := zap.Options {
89133 Development : true ,
@@ -157,12 +201,57 @@ func main() {
157201 renewDeadline := serverConfig .LeaderElection .RenewDeadline .Duration
158202 retryPeriod := serverConfig .LeaderElection .RetryPeriod .Duration
159203
204+ mcManagerOptions := []mcmanager.Option {}
205+ if enableClusterSharding {
206+ setupLog .Info (
207+ "enabling cluster sharding coordinator" ,
208+ "leaseNamespace" ,
209+ clusterShardingLeaseNamespace ,
210+ "leasePrefix" ,
211+ clusterShardingLeasePrefix ,
212+ "peerWeight" ,
213+ clusterShardingPeerWeight ,
214+ )
215+
216+ clusterShardingOptions := []sharded.Option {
217+ sharded .WithShardLease (clusterShardingLeaseNamespace , clusterShardingLeasePrefix ),
218+ sharded .WithPerClusterLease (true ),
219+ }
220+ if clusterShardingPeerWeight > 0 {
221+ clusterShardingOptions = append (
222+ clusterShardingOptions ,
223+ sharded .WithPeerWeight (uint32 (clusterShardingPeerWeight )),
224+ )
225+ }
226+
227+ mcManagerOptions = append (
228+ mcManagerOptions ,
229+ mcmanager .WithCoordinator (
230+ sharded .New (
231+ deploymentCluster .GetClient (),
232+ ctrl .Log .WithName ("cluster-sharding-coordinator" ),
233+ clusterShardingOptions ... ,
234+ ),
235+ ),
236+ )
237+ }
238+
239+ primaryManagerLeaderElection := enableLeaderElection
240+ if enableClusterSharding && enableLeaderElection {
241+ setupLog .Info (
242+ "disabling primary manager leader election while cluster sharding is enabled" ,
243+ "singletonControllersLeaderElection" ,
244+ singletonControllersLeaderElection ,
245+ )
246+ primaryManagerLeaderElection = false
247+ }
248+
160249 mgr , err := mcmanager .New (cfg , provider , ctrl.Options {
161250 Scheme : scheme ,
162251 Metrics : metricsServerOptions ,
163252 WebhookServer : webhookServer ,
164253 HealthProbeBindAddress : probeAddr ,
165- LeaderElection : enableLeaderElection ,
254+ LeaderElection : primaryManagerLeaderElection ,
166255 LeaderElectionID : "6a7d51cc.datumapis.com" ,
167256 LeaderElectionNamespace : leaderElectionNamespace ,
168257 LeaseDuration : & leaseDuration ,
@@ -179,7 +268,7 @@ func main() {
179268 // if you are doing or is intended to do any operation such as perform cleanups
180269 // after the manager stops then its usage might be unsafe.
181270 // LeaderElectionReleaseOnCancel: true,
182- })
271+ }, mcManagerOptions ... )
183272 if err != nil {
184273 setupLog .Error (err , "unable to start manager" )
185274 os .Exit (1 )
@@ -205,6 +294,28 @@ func main() {
205294 os .Exit (1 )
206295 }
207296
297+ var singletonMgr manager.Manager
298+ singletonControllerMgr := mgr .GetLocalManager ()
299+ if enableClusterSharding {
300+ singletonMgr , err = manager .New (cfg , manager.Options {
301+ Scheme : scheme ,
302+ Metrics : metricsserver.Options {BindAddress : "0" },
303+ WebhookServer : webhook .NewServer (webhook.Options {Port : 0 }),
304+ HealthProbeBindAddress : "0" ,
305+ LeaderElection : singletonControllersLeaderElection ,
306+ LeaderElectionID : singletonControllersLeaderElectionID ,
307+ LeaderElectionNamespace : leaderElectionNamespace ,
308+ LeaseDuration : & leaseDuration ,
309+ RenewDeadline : & renewDeadline ,
310+ RetryPeriod : & retryPeriod ,
311+ })
312+ if err != nil {
313+ setupLog .Error (err , "unable to create singleton controller manager" )
314+ os .Exit (1 )
315+ }
316+ singletonControllerMgr = singletonMgr
317+ }
318+
208319 if err := (& controller.NetworkReconciler {}).SetupWithManager (mgr ); err != nil {
209320 setupLog .Error (err , "unable to create controller" , "controller" , "Network" )
210321 os .Exit (1 )
@@ -283,7 +394,7 @@ func main() {
283394 if err := (& controller.GatewayDownstreamCertificateSolverReconciler {
284395 Config : serverConfig ,
285396 DownstreamCluster : downstreamCluster ,
286- }).SetupWithManager (mgr ); err != nil {
397+ }).SetupWithManager (singletonControllerMgr ); err != nil {
287398 setupLog .Error (err , "unable to create controller" , "controller" , "GatewayDownstreamCertificateSolver" )
288399 os .Exit (1 )
289400 }
@@ -311,7 +422,7 @@ func main() {
311422 if err := (& controller.ChallengeReconciler {
312423 Config : serverConfig ,
313424 DownstreamCluster : downstreamCluster ,
314- }).SetupWithManager (mgr ); err != nil {
425+ }).SetupWithManager (singletonControllerMgr ); err != nil {
315426 setupLog .Error (err , "unable to create controller" , "controller" , "Challenge" )
316427 os .Exit (1 )
317428 }
@@ -399,6 +510,13 @@ func main() {
399510 return ignoreCanceled (mgr .Start (ctx ))
400511 })
401512
513+ if singletonMgr != nil {
514+ setupLog .Info ("starting singleton controller manager" )
515+ g .Go (func () error {
516+ return ignoreCanceled (singletonMgr .Start (ctx ))
517+ })
518+ }
519+
402520 if err := g .Wait (); err != nil {
403521 setupLog .Error (err , "unable to start" )
404522 os .Exit (1 )
0 commit comments