Repository: kubernetes-sigs/kwok (go)

Issue:
Customize the number of workers to sync nodes in parallel
### What would you like to be added?

add NodeSyncWorkerParallelism in kwok-contorller to customize number of workers to sync nodes in parallel.

### Why is this needed?

In large-scale cluster stress testing, we use kwok to simulate the following scenario: after a large number of kubelet restarts, there are concurrent requests sent to the apiserver to list pods by node name. 
Currently, the kwok-controller can only simulate a scenario with a concurrency of 1. We hope to add a parameter that allows users to specify the concurrency for the list pods by node name requests to better simulate the real-world scenario of a large number of kubelet restarts.

Pull request:
Customize the number of workers to sync nodes in parallel
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?

/kind feature

<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:



#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #1346 

#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
add NodeSyncWorkerParallelism to customize number of workers to sync nodes in parallel.
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


Patch:
diff --git a/pkg/apis/config/v1alpha1/kwok_configuration_types.go b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
index 48f69750..e6b64fde 100644
--- a/pkg/apis/config/v1alpha1/kwok_configuration_types.go
+++ b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
@@ -149,6 +149,10 @@ type KwokConfigurationOptions struct {
 	// NodeLeaseParallelism is the number of NodeLeases that are allowed to be processed in parallel.
 	// +default=4
 	NodeLeaseParallelism uint `json:"nodeLeaseParallelism,omitempty"`
+
+	// PodsOnNodeSyncParallelism is the number of workers to sync pods on nodes in parallel.
+	// +default=1
+	PodsOnNodeSyncParallelism uint `json:"podsOnNodeSyncParallelism,omitempty"`
 }
 
 // TracingConfiguration provides versioned configuration for OpenTelemetry tracing clients.
diff --git a/pkg/apis/config/v1alpha1/zz_generated.defaults.go b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
index dd6d1c6b..45b676a9 100644
--- a/pkg/apis/config/v1alpha1/zz_generated.defaults.go
+++ b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
@@ -67,6 +67,9 @@ func SetObjectDefaults_KwokConfiguration(in *KwokConfiguration) {
 	if in.Options.NodeLeaseParallelism == 0 {
 		in.Options.NodeLeaseParallelism = 4
 	}
+	if in.Options.PodsOnNodeSyncParallelism == 0 {
+		in.Options.PodsOnNodeSyncParallelism = 1
+	}
 }
 
 func SetObjectDefaults_KwokctlConfiguration(in *KwokctlConfiguration) {
diff --git a/pkg/apis/internalversion/kwok_configuration_types.go b/pkg/apis/internalversion/kwok_configuration_types.go
index 2d0f46fe..d201835c 100644
--- a/pkg/apis/internalversion/kwok_configuration_types.go
+++ b/pkg/apis/internalversion/kwok_configuration_types.go
@@ -102,6 +102,9 @@ type KwokConfigurationOptions struct {
 
 	// NodeLeaseParallelism is the number of NodeLeases that are allowed to be processed in parallel.
 	NodeLeaseParallelism uint
+
+	// PodsOnNodeSyncParallelism is the number of workers to sync pods on nodes in parallel.
+	PodsOnNodeSyncParallelism uint
 }
 
 // TracingConfiguration provides versioned configuration for OpenTelemetry tracing clients.
diff --git a/pkg/apis/internalversion/zz_generated.conversion.go b/pkg/apis/internalversion/zz_generated.conversion.go
index c45034b7..e8cc4dd1 100644
--- a/pkg/apis/internalversion/zz_generated.conversion.go
+++ b/pkg/apis/internalversion/zz_generated.conversion.go
@@ -1533,6 +1533,7 @@ func autoConvert_internalversion_KwokConfigurationOptions_To_v1alpha1_KwokConfig
 	out.NodePlayStageParallelism = in.NodePlayStageParallelism
 	out.NodeLeaseDurationSeconds = in.NodeLeaseDurationSeconds
 	out.NodeLeaseParallelism = in.NodeLeaseParallelism
+	out.PodsOnNodeSyncParallelism = in.PodsOnNodeSyncParallelism
 	return nil
 }
 
@@ -1574,6 +1575,7 @@ func autoConvert_v1alpha1_KwokConfigurationOptions_To_internalversion_KwokConfig
 	out.NodePlayStageParallelism = in.NodePlayStageParallelism
 	out.NodeLeaseDurationSeconds = in.NodeLeaseDurationSeconds
 	out.NodeLeaseParallelism = in.NodeLeaseParallelism
+	out.PodsOnNodeSyncParallelism = in.PodsOnNodeSyncParallelism
 	return nil
 }
 
diff --git a/pkg/kwok/cmd/root.go b/pkg/kwok/cmd/root.go
index 99dfcbb9..8e90512a 100644
--- a/pkg/kwok/cmd/root.go
+++ b/pkg/kwok/cmd/root.go
@@ -292,6 +292,7 @@ func runE(ctx context.Context, flags *flagpole) error {
 		NodeLeaseParallelism:                  flags.Options.NodeLeaseParallelism,
 		NodeLeaseDurationSeconds:              flags.Options.NodeLeaseDurationSeconds,
 		ID:                                    id,
+		PodsOnNodeSyncParallelism:             flags.Options.PodsOnNodeSyncParallelism,
 	})
 	if err != nil {
 		return err
diff --git a/pkg/kwok/controllers/controller.go b/pkg/kwok/controllers/controller.go
index 1b39c798..e5e39d67 100644
--- a/pkg/kwok/controllers/controller.go
+++ b/pkg/kwok/controllers/controller.go
@@ -122,6 +122,7 @@ type Config struct {
 	NodePlayStageParallelism              uint
 	NodeLeaseDurationSeconds              uint
 	NodeLeaseParallelism                  uint
+	PodsOnNodeSyncParallelism             uint
 	ID                                    string
 	EnableMetrics                         bool
 	EnablePodCache                        bool
@@ -339,7 +340,9 @@ func (c *Controller) startStageController(ctx context.Context, ref internalversi
 			return fmt.Errorf("failed to init pod controller: %w", err)
 		}
 
-		go c.podsOnNodeSyncWorker(ctx)
+		for i := uint(0); i < c.conf.PodsOnNodeSyncParallelism; i++ {
+			go c.podsOnNodeSyncWorker(ctx)
+		}
 
 	case nodeRef:
 		err := c.initNodeLeaseController(ctx)
diff --git a/site/content/en/docs/generated/apis.md b/site/content/en/docs/generated/apis.md
index 5a77cd27..e9b5456f 100644
--- a/site/content/en/docs/generated/apis.md
+++ b/site/content/en/docs/generated/apis.md
@@ -2647,6 +2647,17 @@ uint
 <p>NodeLeaseParallelism is the number of NodeLeases that are allowed to be processed in parallel.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>podsOnNodeSyncParallelism</code>
+<em>
+uint
+</em>
+</td>
+<td>
+<p>PodsOnNodeSyncParallelism is the number of workers to sync pods on nodes in parallel.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="config.kwok.x-k8s.io/v1alpha1.KwokctlConfigurationOptions">


Test patch:
diff --git a/pkg/kwok/controllers/controller_test.go b/pkg/kwok/controllers/controller_test.go
index 7766377b..805bb0e8 100644
--- a/pkg/kwok/controllers/controller_test.go
+++ b/pkg/kwok/controllers/controller_test.go
@@ -151,6 +151,27 @@ func TestController(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "node controller test: manage all nodes in parallel",
+			conf: Config{
+				TypedClient:    fake.NewSimpleClientset(nodes...),
+				ManageAllNodes: true,
+				LocalStages: map[internalversion.StageResourceRef][]*internalversion.Stage{
+					podRef:  podStages,
+					nodeRef: nodeStages,
+				},
+				CIDR:                      "10.0.0.1/24",
+				NodePlayStageParallelism:  1,
+				PodPlayStageParallelism:   1,
+				PodsOnNodeSyncParallelism: 3,
+			},
+			wantNodePhase: map[string]corev1.NodePhase{
+				"node-0": corev1.NodeRunning,
+				"node-1": corev1.NodeRunning,
+				"node-2": corev1.NodeRunning,
+			},
+			wantErr: false,
+		},
 	}
 
 	ctx := context.Background()
diff --git a/test/kwok/kwok.test.sh b/test/kwok/kwok.test.sh
index 6bf79f09..3111c9f7 100755
--- a/test/kwok/kwok.test.sh
+++ b/test/kwok/kwok.test.sh
@@ -183,7 +183,7 @@ function main() {
   test_pod_running || failed+=("pod_running_again")
   test_check_pod_status || failed+=("check_pod_status_again")
 
-  sleep 40
+  sleep 45
 
   test_check_node_lease_transitions 1 || failed+=("check_node_lease_transitions_again")