Skip to content

Commit

Permalink
Add separate concurrency flag for cluster cache tracker
Browse files Browse the repository at this point in the history
  • Loading branch information
chrischdi committed Aug 4, 2023
1 parent 54dfd42 commit fb958c0
Show file tree
Hide file tree
Showing 4 changed files with 87 additions and 70 deletions.
47 changes: 26 additions & 21 deletions bootstrap/kubeadm/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -76,26 +76,27 @@ func init() {
}

var (
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchFilterValue string
watchNamespace string
profilerAddress string
enableContentionProfiling bool
clusterConcurrency int
kubeadmConfigConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
webhookPort int
webhookCertDir string
healthAddr string
tokenTTL time.Duration
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchFilterValue string
watchNamespace string
profilerAddress string
enableContentionProfiling bool
clusterConcurrency int
clusterCacheTrackerConcurrency int
kubeadmConfigConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
webhookPort int
webhookCertDir string
healthAddr string
tokenTTL time.Duration
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
)

// InitFlags initializes this manager's flags.
Expand Down Expand Up @@ -128,6 +129,10 @@ func InitFlags(fs *pflag.FlagSet) {

fs.IntVar(&clusterConcurrency, "cluster-concurrency", 10,
"Number of clusters to process simultaneously")
fs.MarkDeprecated("cluster-concurrency", "This flag has no function anymore and is going to be removed in a next release. Use \"--clustercachetracker-concurrency\" insteaed.")

fs.IntVar(&clusterCacheTrackerConcurrency, "clustercachetracker-concurrency", 10,
"Number of clusters to process simultaneously")

fs.IntVar(&kubeadmConfigConcurrency, "kubeadmconfig-concurrency", 10,
"Number of kubeadm configs to process simultaneously")
Expand Down Expand Up @@ -307,7 +312,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
Client: mgr.GetClient(),
Tracker: tracker,
WatchFilterValue: watchFilterValue,
}).SetupWithManager(ctx, mgr, concurrency(clusterConcurrency)); err != nil {
}).SetupWithManager(ctx, mgr, concurrency(clusterCacheTrackerConcurrency)); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
os.Exit(1)
}
Expand Down
6 changes: 5 additions & 1 deletion controlplane/kubeadm/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@ var (
profilerAddress string
enableContentionProfiling bool
kubeadmControlPlaneConcurrency int
clusterCacheTrackerConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
Expand Down Expand Up @@ -134,6 +135,9 @@ func InitFlags(fs *pflag.FlagSet) {
fs.IntVar(&kubeadmControlPlaneConcurrency, "kubeadmcontrolplane-concurrency", 10,
"Number of kubeadm control planes to process simultaneously")

fs.IntVar(&clusterCacheTrackerConcurrency, "clustercachetracker-concurrency", 10,
"Number of clusters to process simultaneously")

fs.DurationVar(&syncPeriod, "sync-period", 10*time.Minute,
"The minimum interval at which watched resources are reconciled (e.g. 15m)")

Expand Down Expand Up @@ -320,7 +324,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
Client: mgr.GetClient(),
Tracker: tracker,
WatchFilterValue: watchFilterValue,
}).SetupWithManager(ctx, mgr, concurrency(kubeadmControlPlaneConcurrency)); err != nil {
}).SetupWithManager(ctx, mgr, concurrency(clusterCacheTrackerConcurrency)); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
os.Exit(1)
}
Expand Down
62 changes: 33 additions & 29 deletions main.go
Original file line number Diff line number Diff line change
Expand Up @@ -82,34 +82,35 @@ var (
controllerName = "cluster-api-controller-manager"

// flags.
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchNamespace string
watchFilterValue string
profilerAddress string
enableContentionProfiling bool
clusterTopologyConcurrency int
clusterClassConcurrency int
clusterConcurrency int
extensionConfigConcurrency int
machineConcurrency int
machineSetConcurrency int
machineDeploymentConcurrency int
machinePoolConcurrency int
clusterResourceSetConcurrency int
machineHealthCheckConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
nodeDrainClientTimeout time.Duration
webhookPort int
webhookCertDir string
healthAddr string
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchNamespace string
watchFilterValue string
profilerAddress string
enableContentionProfiling bool
clusterTopologyConcurrency int
clusterCacheTrackerConcurrency int
clusterClassConcurrency int
clusterConcurrency int
extensionConfigConcurrency int
machineConcurrency int
machineSetConcurrency int
machineDeploymentConcurrency int
machinePoolConcurrency int
clusterResourceSetConcurrency int
machineHealthCheckConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
nodeDrainClientTimeout time.Duration
webhookPort int
webhookCertDir string
healthAddr string
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
)

func init() {
Expand Down Expand Up @@ -177,6 +178,9 @@ func InitFlags(fs *pflag.FlagSet) {
fs.IntVar(&clusterConcurrency, "cluster-concurrency", 10,
"Number of clusters to process simultaneously")

fs.IntVar(&clusterCacheTrackerConcurrency, "clustercachetracker-concurrency", 10,
"Number of clusters to process simultaneously")

fs.IntVar(&extensionConfigConcurrency, "extensionconfig-concurrency", 10,
"Number of extension configs to process simultaneously")

Expand Down Expand Up @@ -394,7 +398,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
Client: mgr.GetClient(),
Tracker: tracker,
WatchFilterValue: watchFilterValue,
}).SetupWithManager(ctx, mgr, concurrency(clusterConcurrency)); err != nil {
}).SetupWithManager(ctx, mgr, concurrency(clusterCacheTrackerConcurrency)); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
os.Exit(1)
}
Expand Down
42 changes: 23 additions & 19 deletions test/infrastructure/docker/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -68,24 +68,25 @@ var (
controllerName = "cluster-api-docker-controller-manager"

// flags.
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchNamespace string
watchFilterValue string
profilerAddress string
enableContentionProfiling bool
concurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
webhookPort int
webhookCertDir string
healthAddr string
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
metricsBindAddr string
enableLeaderElection bool
leaderElectionLeaseDuration time.Duration
leaderElectionRenewDeadline time.Duration
leaderElectionRetryPeriod time.Duration
watchNamespace string
watchFilterValue string
profilerAddress string
enableContentionProfiling bool
concurrency int
clusterCacheTrackerConcurrency int
syncPeriod time.Duration
restConfigQPS float32
restConfigBurst int
webhookPort int
webhookCertDir string
healthAddr string
tlsOptions = flags.TLSOptions{}
logOptions = logs.NewOptions()
)

func init() {
Expand Down Expand Up @@ -135,6 +136,9 @@ func initFlags(fs *pflag.FlagSet) {
fs.IntVar(&concurrency, "concurrency", 10,
"The number of docker machines to process simultaneously")

fs.IntVar(&clusterCacheTrackerConcurrency, "clustercachetracker-concurrency", 10,
"Number of clusters to process simultaneously")

fs.DurationVar(&syncPeriod, "sync-period", 10*time.Minute,
"The minimum interval at which watched resources are reconciled (e.g. 15m)")

Expand Down Expand Up @@ -316,7 +320,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
Tracker: tracker,
WatchFilterValue: watchFilterValue,
}).SetupWithManager(ctx, mgr, controller.Options{
MaxConcurrentReconciles: concurrency,
MaxConcurrentReconciles: clusterCacheTrackerConcurrency,
}); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "ClusterCacheReconciler")
os.Exit(1)
Expand Down

0 comments on commit fb958c0

Please sign in to comment.