Repository: kubernetes-sigs/kwok
Pull request #1355 (2025-04-25 03:28:49)

Enable stream watcher to sync pods on node.
<!--  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 #

#### 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 EnablePodsOnNodeSyncStreamWatch option to switch between watch list and stream watch to sync pods on node.
```

#### 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

```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kwok/pull/1355#" title="Author self-approved">lianghao208</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [huang-wei](https://github.com/huang-wei) for approval. For more information see [the Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkwok).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/kwok/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["huang-wei"]} -->
netlify[bot]: ### <span aria-hidden="true">✅</span> Deploy Preview for *k8s-kwok* canceled.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 4d6b994a293e96226a9025c0a7215960eb017a26 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/k8s-kwok/deploys/680b0174149c220008b82068 |
k8s-ci-robot: @lianghao208: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kwok-unit-test-main | 4d6b994a293e96226a9025c0a7215960eb017a26 | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kwok/1355/pull-kwok-unit-test-main/1915608920454336512) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=1355). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Alianghao208). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kwok/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @lianghao208: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kwok-verify-main | 56445f69bd1b3f54ec828301c82a2005a204540d | [link](https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kwok/1355/pull-kwok-verify-main/1915672590483984384) | true | `/test pull-kwok-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=1355). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Alianghao208). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kwok/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/pkg/apis/config/v1alpha1/kwok_configuration_types.go b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
index d052ce4a..2ea8b8fa 100644
--- a/pkg/apis/config/v1alpha1/kwok_configuration_types.go
+++ b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
@@ -157,6 +157,10 @@ type KwokConfigurationOptions struct {
 	// EnablePodsOnNodeSyncListPager enables pager list for workers to sync pods on nodes.
 	// +default=true
 	EnablePodsOnNodeSyncListPager *bool `json:"enablePodsOnNodeSyncListPager"`
+
+	// EnablePodsOnNodeSyncStreamWatch enables stream watch for workers to sync pods on nodes.
+	// +default=false
+	EnablePodsOnNodeSyncStreamWatch *bool `json:"enablePodsOnNodeSyncStreamWatch"`
 }
 
 // TracingConfiguration provides versioned configuration for OpenTelemetry tracing clients.
diff --git a/pkg/apis/config/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/config/v1alpha1/zz_generated.deepcopy.go
index 556a2de5..10ab5f54 100644
--- a/pkg/apis/config/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/config/v1alpha1/zz_generated.deepcopy.go
@@ -232,6 +232,11 @@ func (in *KwokConfigurationOptions) DeepCopyInto(out *KwokConfigurationOptions)
 		*out = new(bool)
 		**out = **in
 	}
+	if in.EnablePodsOnNodeSyncStreamWatch != nil {
+		in, out := &in.EnablePodsOnNodeSyncStreamWatch, &out.EnablePodsOnNodeSyncStreamWatch
+		*out = new(bool)
+		**out = **in
+	}
 	return
 }
 
diff --git a/pkg/apis/config/v1alpha1/zz_generated.defaults.go b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
index 3cd8c7db..3beb1868 100644
--- a/pkg/apis/config/v1alpha1/zz_generated.defaults.go
+++ b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
@@ -74,6 +74,10 @@ func SetObjectDefaults_KwokConfiguration(in *KwokConfiguration) {
 		var ptrVar1 bool = true
 		in.Options.EnablePodsOnNodeSyncListPager = &ptrVar1
 	}
+	if in.Options.EnablePodsOnNodeSyncStreamWatch == nil {
+		var ptrVar1 bool = false
+		in.Options.EnablePodsOnNodeSyncStreamWatch = &ptrVar1
+	}
 }
 
 func SetObjectDefaults_KwokctlConfiguration(in *KwokctlConfiguration) {
diff --git a/pkg/apis/internalversion/kwok_configuration_types.go b/pkg/apis/internalversion/kwok_configuration_types.go
index 423f4081..f96d1f02 100644
--- a/pkg/apis/internalversion/kwok_configuration_types.go
+++ b/pkg/apis/internalversion/kwok_configuration_types.go
@@ -108,6 +108,9 @@ type KwokConfigurationOptions struct {
 
 	// EnablePodsOnNodeSyncListPager enables pager list for workers to sync pods on nodes.
 	EnablePodsOnNodeSyncListPager bool
+
+	// EnablePodsOnNodeSyncStreamWatch enables stream watch for workers to sync pods on nodes.
+	EnablePodsOnNodeSyncStreamWatch bool
 }
 
 // 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 bbae9ac3..778c9d92 100644
--- a/pkg/apis/internalversion/zz_generated.conversion.go
+++ b/pkg/apis/internalversion/zz_generated.conversion.go
@@ -1537,6 +1537,9 @@ func autoConvert_internalversion_KwokConfigurationOptions_To_v1alpha1_KwokConfig
 	if err := v1.Convert_bool_To_Pointer_bool(&in.EnablePodsOnNodeSyncListPager, &out.EnablePodsOnNodeSyncListPager, s); err != nil {
 		return err
 	}
+	if err := v1.Convert_bool_To_Pointer_bool(&in.EnablePodsOnNodeSyncStreamWatch, &out.EnablePodsOnNodeSyncStreamWatch, s); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -1582,6 +1585,9 @@ func autoConvert_v1alpha1_KwokConfigurationOptions_To_internalversion_KwokConfig
 	if err := v1.Convert_Pointer_bool_To_bool(&in.EnablePodsOnNodeSyncListPager, &out.EnablePodsOnNodeSyncListPager, s); err != nil {
 		return err
 	}
+	if err := v1.Convert_Pointer_bool_To_bool(&in.EnablePodsOnNodeSyncStreamWatch, &out.EnablePodsOnNodeSyncStreamWatch, s); err != nil {
+		return err
+	}
 	return nil
 }
 
diff --git a/pkg/kwok/cmd/root.go b/pkg/kwok/cmd/root.go
index 469fcbe1..83307257 100644
--- a/pkg/kwok/cmd/root.go
+++ b/pkg/kwok/cmd/root.go
@@ -294,6 +294,7 @@ func runE(ctx context.Context, flags *flagpole) error {
 		ID:                                    id,
 		PodsOnNodeSyncParallelism:             flags.Options.PodsOnNodeSyncParallelism,
 		EnablePodsOnNodeSyncListPager:         flags.Options.EnablePodsOnNodeSyncListPager,
+		EnablePodsOnNodeSyncStreamWatch:       flags.Options.EnablePodsOnNodeSyncStreamWatch,
 	})
 	if err != nil {
 		return err
diff --git a/pkg/kwok/controllers/controller.go b/pkg/kwok/controllers/controller.go
index caa3b24d..e8476ded 100644
--- a/pkg/kwok/controllers/controller.go
+++ b/pkg/kwok/controllers/controller.go
@@ -124,6 +124,7 @@ type Config struct {
 	NodeLeaseParallelism                  uint
 	PodsOnNodeSyncParallelism             uint
 	EnablePodsOnNodeSyncListPager         bool
+	EnablePodsOnNodeSyncStreamWatch       bool
 	ID                                    string
 	EnableMetrics                         bool
 	EnablePodCache                        bool
@@ -569,8 +570,9 @@ func (c *Controller) podsOnNodeSyncWorker(ctx context.Context) {
 		}
 
 		opt := informer.Option{
-			FieldSelector:   fields.OneTermEqualSelector("spec.nodeName", nodeName).String(),
-			EnableListPager: c.conf.EnablePodsOnNodeSyncListPager,
+			FieldSelector:     fields.OneTermEqualSelector("spec.nodeName", nodeName).String(),
+			EnableListPager:   c.conf.EnablePodsOnNodeSyncListPager,
+			EnableStreamWatch: c.conf.EnablePodsOnNodeSyncStreamWatch,
 		}
 
 		err := c.podsInformer.Sync(ctx, opt, c.podsChan)
diff --git a/pkg/utils/informer/event.go b/pkg/utils/informer/event.go
index 8426563b..4e0fe1ad 100644
--- a/pkg/utils/informer/event.go
+++ b/pkg/utils/informer/event.go
@@ -62,6 +62,7 @@ type Option struct {
 	FieldSelector      string
 	AnnotationSelector string
 	EnableListPager    bool
+	EnableStreamWatch  bool
 	annotationSelector labels.Selector
 }
 
@@ -72,7 +73,7 @@ func (o *Option) setup(opts *metav1.ListOptions) {
 	if o.FieldSelector != "" {
 		opts.FieldSelector = o.FieldSelector
 	}
-	if !o.EnableListPager {
+	if !o.EnableListPager || o.EnableStreamWatch {
 		opts.ResourceVersion = "0"
 	}
 }
diff --git a/pkg/utils/informer/informer.go b/pkg/utils/informer/informer.go
index 588a0731..aa5a05e1 100644
--- a/pkg/utils/informer/informer.go
+++ b/pkg/utils/informer/informer.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"sync"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -28,6 +29,9 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/pager"
+	"k8s.io/client-go/util/watchlist"
+
+	"sigs.k8s.io/kwok/pkg/log"
 )
 
 // Informer is a wrapper around a Get/List/Watch function.
@@ -59,6 +63,19 @@ func (i *Informer[T, L]) Sync(ctx context.Context, opt Option, events chan<- Eve
 		events <- Event[T]{Type: Sync, Object: obj.(T)}
 		return nil
 	}
+
+	if opt.EnableStreamWatch {
+		if watchListOptions, hasWatchListOptionsPrepared, watchListOptionsErr := watchlist.PrepareWatchListOptionsFromListOptions(opt.toListOptions()); watchListOptionsErr != nil {
+			log.FromContext(ctx).Error("Failed preparing watchlist options, falling back to the standard LIST semantics", watchListOptionsErr)
+		} else if hasWatchListOptionsPrepared {
+			watcher, err := i.WatchFunc(ctx, watchListOptions)
+			if err != nil {
+				return err
+			}
+			return handleWatchList(ctx, watcher, fun)
+		}
+	}
+
 	if !opt.EnableListPager {
 		list, err := i.ListFunc(ctx, opt.toListOptions())
 		if err != nil {
@@ -335,3 +352,44 @@ func objType(expectedType runtime.Object) runtime.Object {
 		return &obj
 	}
 }
+
+// handleWatchList copy from watchListClient feature with some modifications to allow streaming handle.
+// Note that this function will close the passed watch.
+func handleWatchList(ctx context.Context, w watch.Interface, eventsFunc func(obj runtime.Object) error) error {
+	defer w.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case event, ok := <-w.ResultChan():
+			if !ok {
+				return fmt.Errorf("unexpected watch close")
+			}
+			if event.Type == watch.Error {
+				return apierrors.FromObject(event.Object)
+			}
+			meta, err := meta.Accessor(event.Object)
+			if err != nil {
+				return fmt.Errorf("failed to parse watch event: %#v", event)
+			}
+
+			switch event.Type {
+			case watch.Added:
+				if err = eventsFunc(event.Object); err != nil {
+					return err
+				}
+			case watch.Bookmark:
+				if meta.GetAnnotations()[metav1.InitialEventsAnnotationKey] == "true" {
+					base64EncodedInitialEventsListBlueprint := meta.GetAnnotations()[metav1.InitialEventsListBlueprintAnnotationKey]
+					if len(base64EncodedInitialEventsListBlueprint) == 0 {
+						return fmt.Errorf("%q annotation is missing content", metav1.InitialEventsListBlueprintAnnotationKey)
+					}
+					return nil
+				}
+			default:
+				return fmt.Errorf("unexpected watch event %#v, expected to only receive watch.Added and watch.Bookmark events", event)
+			}
+		}
+	}
+}
diff --git a/site/content/en/docs/generated/apis.md b/site/content/en/docs/generated/apis.md
index 2b830e57..869f41c5 100644
--- a/site/content/en/docs/generated/apis.md
+++ b/site/content/en/docs/generated/apis.md
@@ -2669,6 +2669,17 @@ bool
 <p>EnablePodsOnNodeSyncListPager enables pager list for workers to sync pods on nodes.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>enablePodsOnNodeSyncStreamWatch</code>
+<em>
+bool
+</em>
+</td>
+<td>
+<p>EnablePodsOnNodeSyncStreamWatch enables stream watch for workers to sync pods on nodes.</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 18158914..e5125972 100644
--- a/pkg/kwok/controllers/controller_test.go
+++ b/pkg/kwok/controllers/controller_test.go
@@ -197,6 +197,29 @@ func TestController(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "node controller test: manage all nodes by stream watch",
+			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:       1,
+				EnablePodsOnNodeSyncListPager:   false,
+				EnablePodsOnNodeSyncStreamWatch: true,
+			},
+			wantNodePhase: map[string]corev1.NodePhase{
+				"node-0": corev1.NodeRunning,
+				"node-1": corev1.NodeRunning,
+				"node-2": corev1.NodeRunning,
+			},
+			wantErr: false,
+		},
 	}
 
 	ctx := context.Background()