client-go: Add support for API streaming to the reflector #110772

p0lyn0mial · 2022-06-24T14:26:18Z

What type of PR is this?

/kind feature

What this PR does / why we need it:

This pull request introduces new alpha functionality to the reflector, allowing users to enable API streaming.
To activate this feature, users can set the ENABLE_CLIENT_GO_WATCH_LIST_ALPHA environmental variable.
It is important to note that the server must support streaming for this feature to function properly.
If streaming is not supported by the server, the reflector will revert to the previous method
of obtaining data through LIST/WATCH semantics.

Which issue(s) this PR fixes:

xref: kubernetes/enhancements#3157
xref: #115402
xref: #110960

Special notes for your reviewer:

Does this PR introduce a user-facing change?

Introduces new alpha functionality to the reflector, allowing user to enable API streaming.

To activate this feature, users can set the `ENABLE_CLIENT_GO_WATCH_LIST_ALPHA` environmental variable.
It is important to note that the server must support streaming for this feature to function properly.
If streaming is not supported by the server, the reflector will revert to the previous method
of obtaining data through LIST/WATCH semantics.

Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

- [KEP]: https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list

p0lyn0mial · 2022-06-24T14:26:30Z

/assign @wojtek-t

leilajal · 2022-06-30T16:44:26Z

/triage accepted

wojtek-t · 2023-03-09T14:00:55Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+			if isErrorRetriableWithSideEffectsFn(err) {
+				continue
+			}
+			klog.V(2).Infof("%s: watch-list of %v ended with: %v", r.name, r.expectedGVK, err)


remove - this will be logged here:
https://github.com/kubernetes/kubernetes/pull/110772/files#diff-9ccdf713e010f73dbebd01e936cb0077fc63e4f5ab941d865ded42da219d84ecR337

wojtek-t · 2023-03-09T14:01:09Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+		if bookmarkReceived {
+			break // success we got initial data and the bookmark
+		}
+		// try one more time


nit: remove this comment - it doesn't bring any value.

wojtek-t · 2023-03-09T14:02:59Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+		}
+		// try one more time
+	}
+	r.setIsLastSyncResourceVersionUnavailable(false) // watch-list was successful


nit

// We successfully got initial state from watch-list confirmed by the // "k8s.io/initial-events-end" bookmark. initTrace.Step(...) r.setIsLastSyncResource... ...

wojtek-t · 2023-03-09T14:07:01Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

@@ -651,17 +801,35 @@ func (r *Reflector) relistResourceVersion() string {
 	r.lastSyncResourceVersionMutex.RLock()
 	defer r.lastSyncResourceVersionMutex.RUnlock()

+	if r.lastSyncResourceVersion == "" {


This isn't correct - I mean, we're changing semantics here.

If isLastSyncResourceVersionUnavailable is true, then previously we would return "", and now we will return '0" if also lastSyncResourceVersion is empty.

Let's not change the semantics for the old path.

[FWIW, I'm not sure why we actually did that, but even if we want to change it, we should do that explicitly in a separate PR and not change it as part of larger complex PR.]

doesn't setIsLastSyncResourceVersionUnavailable imply that r. lastSyncResourceVersion is != "" ? After all we must have provided some RV to the server.

Anyway I can bring it back to the original state just in case.

wojtek-t · 2023-03-09T14:15:46Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+// It begins with synthetic "Added" events of all resources up to the most recent ResourceVersion.
+// It ends with a synthetic "Bookmark" event containing the most recent ResourceVersion.
+// After receiving a "Bookmark" event the reflector is considered to be synchronized.
+// It replaces its internal store with the collected items (syncWith) and reuses the current watch requests for getting further events.


nit: remove "(syncWith)" and split the line a bit :)

wojtek-t · 2023-03-09T14:17:19Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+	var err error
+	var temporaryStore Store
+	var resourceVersion string
+	var bookmarkReceived *bool


nit: This one isn't used in the scope of function - let's define it where it is used only.

wojtek-t · 2023-03-09T14:22:54Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+	var temporaryStore Store
+	var resourceVersion string
+	var bookmarkReceived *bool
+	isErrorRetriableWithSideEffectsFn := func(err error) bool {


nit: I'm not a huge fan of those nested functions - I'm wondering if we can make it an actual reflector function...

maybe in the future, for now it applies only to the watchList method.
I can add a TODO for me - make it a method and find a way to unify error handling in the reflector.

That's not great, but I can probably live with this.

wojtek-t · 2023-03-09T14:25:26Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

@@ -546,16 +682,13 @@ func watchHandler(start time.Time,
 	name string,
 	expectedTypeName string,
 	setLastSyncResourceVersion func(string),
+	exitOnInitEventBookmark *bool,


nit: I'm sorry - that was my mental shortcut, I think we should name it
"exitOnInitialEventsEndBookmark"
(for consistency with annotation name)

wojtek-t · 2023-03-09T14:26:00Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

@@ -609,6 +742,11 @@ loop:
 				}
 			case watch.Bookmark:
 				// A `Bookmark` means watch has synced here, just update the resourceVersion
+				if _, ok := meta.GetAnnotations()["k8s.io/initial-events-end"]; ok {
+					if exitOnInitEventBookmark != nil {
+						exitOnInitEventBookmark = pointer.Bool(true)


*exitOn.. = true ?

right :) - thx.

staging/src/k8s.io/client-go/tools/cache/reflector.go

wojtek-t

I like the test coverage - I added some comments, but there isn't anything big there.

wojtek-t · 2023-03-10T07:13:41Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+			if !apierrors.IsInvalid(err) {
+				return err
+			}
+			klog.V(2).Info("the watch-list feature is not supported by the server, falling back to the previous LIST/WATCH semantic")


make this a warning

wojtek-t · 2023-03-10T07:14:45Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+			}
+			klog.V(2).Info("the watch-list feature is not supported by the server, falling back to the previous LIST/WATCH semantic")
+			fallbackToList = true
+			w = nil // just in case it was actually set


nit;

// Ensure that we won't accidentally pass some garbage down the watch. w = nil

wojtek-t · 2023-03-10T07:16:59Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+	var temporaryStore Store
+	var resourceVersion string
+	var bookmarkReceived *bool
+	isErrorRetriableWithSideEffectsFn := func(err error) bool {


That's not great, but I can probably live with this.

wojtek-t · 2023-03-10T07:24:26Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+			}
+			return nil, err
+		}
+		// if we are here, that means we either hit the case 1 or the case 2


remove this comment - this doesn't bring any value

wojtek-t · 2023-03-10T07:26:07Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+			return nil, err
+		}
+		if *bookmarkReceived {
+			break // success we got initial data and the bookmark


nit: remove this comment - this is obvious from variables naming.

wojtek-t · 2023-03-10T07:46:28Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+	lw.listCounter++
+	lw.requestOptions = append(lw.requestOptions, options)
+	if lw.listCounter == lw.closeAfterListRequests {
+		defer lw.close()


Why it's defer here? Can't we just close it immediately?

wojtek-t · 2023-03-10T07:48:08Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+	if lw.watchCounter == lw.closeAfterWatchRequests {
+		defer lw.close()
+	}
+	if lw.watcherErrorPredicate != nil {


nit: let's call it maybe watchOptionsPredicate

wojtek-t · 2023-03-10T07:51:55Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+		{
+			name: "the reflector can fall back to old LIST/WATCH semantics when a server doesn't support streaming",
+			watchServerErrorPredicate: func(options metav1.ListOptions) error {
+				if options.ResourceVersionMatch == metav1.ResourceVersionMatchNotOlderThan {


It would be more intuitive to check options.SendInitialEvents here...

wojtek-t · 2023-03-10T07:55:17Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+			returnedWatchEvents: []watch.Event{
+				{Type: watch.Added, Object: makePod("p1", "1")},
+				// second request
+				{Type: watch.Added, Object: makePod("p2", "2")},


I would sent the above p1 event again here - so that it looks more like a retry.

wojtek-t · 2023-03-10T07:57:34Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+				}
+			}(),
+			stopAfterWatchEvents:  3,
+			closeAfterWatchEvents: 5,


I don't understand it - there aren't 5 events here...

I must have copy&past it from the previous test cases. Will remove.
It didn't affect the test - was unused.

wojtek-t

Last 3 minor nits - other than that LGTM.

wojtek-t · 2023-03-10T10:16:13Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+			go func() {
+				for i, e := range scenario.watchEvents {
+					listWatcher.fakeWatcher.Action(e.Type, e.Object)
+					i++


I still don't understand it. I played with it:
https://go.dev/play/p/FEuLkFBGb_g

having i++ there and not having it actually doesn't change the output - that's super misleading...

It would be much more intuitive to me if you instead change to:

removing this i++ fline

switching the following ifs to if i+1 == ....

wojtek-t · 2023-03-10T10:21:54Z

staging/src/k8s.io/client-go/tools/cache/reflector_watchlist_test.go

+		actualPods = append(actualPods, *p.(*v1.Pod))
+	}
+
+	for _, expectedPod := range expectedPods {


instead of this, let's just sort the both lists by name and compare the whole lists - it's more intuitive.

[This code in general would return true for the following lists:

[a, a, b]

[a, b, b]

wojtek-t · 2023-03-10T10:26:06Z

staging/src/k8s.io/client-go/tools/cache/reflector.go

+	if r.UseWatchList {
+		w, err = r.watchList(stopCh)
+		if w == nil && err == nil {
+			return nil // stopCh requested


nit:

// stopCh was closed return nil

wojtek-t · 2023-03-10T11:28:43Z

/lgtm
/approve

k8s-ci-robot · 2023-03-10T11:28:49Z

LGTM label has been added.

Git tree hash: 1e1503c42c1a45be2c251bab6cabcfed37b1e119

k8s-ci-robot · 2023-03-10T11:29:14Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: p0lyn0mial, wojtek-t

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~staging/src/k8s.io/apiserver/pkg/storage/OWNERS~~ [wojtek-t]
~~staging/src/k8s.io/client-go/tools/cache/OWNERS~~ [wojtek-t]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

k8s-ci-robot · 2023-03-10T11:47:31Z

@p0lyn0mial: The following tests 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-kubernetes-e2e-gce-ubuntu-containerd	`077c5bd`	link	true	`/test pull-kubernetes-e2e-gce-ubuntu-containerd`
pull-kubernetes-e2e-gce-100-performance	`077c5bd`	link	true	`/test pull-kubernetes-e2e-gce-100-performance`

Full PR test history. Your PR dashboard. Please help us cut down on flakes by linking to an open issue when you hit one in your PR.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.

p0lyn0mial · 2023-03-10T11:57:25Z

/test pull-kubernetes-e2e-kind-ipv6

k8s-ci-robot assigned wojtek-t Jun 24, 2022

k8s-ci-robot requested review from justinsb and lavalamp June 24, 2022 14:27

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from deea5e9 to 8632649 Compare June 28, 2022 20:41

k8s-ci-robot added the area/apiserver label Jun 28, 2022

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from 8632649 to ac00d9e Compare June 29, 2022 12:26

k8s-ci-robot added triage/accepted Indicates an issue or PR is ready to be actively worked on. and removed needs-triage Indicates an issue or PR lacks a `triage/foo` label and requires one. labels Jun 30, 2022

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from ac00d9e to 04bcbfa Compare July 1, 2022 14:54

k8s-ci-robot added size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. and removed size/L Denotes a PR that changes 100-499 lines, ignoring generated files. labels Jul 1, 2022

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch 4 times, most recently from 7b7c78c to 25a8421 Compare July 4, 2022 14:10

k8s-ci-robot removed the size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. label Jul 4, 2022

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from 95f3bd1 to 19fd154 Compare March 9, 2023 13:53

k8s-ci-robot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Mar 9, 2023

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch 2 times, most recently from 5dc3c67 to 718904b Compare March 9, 2023 14:14

wojtek-t reviewed Mar 9, 2023

View reviewed changes

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch 3 times, most recently from cb4b9cb to aac3889 Compare March 9, 2023 16:47

wojtek-t reviewed Mar 10, 2023

View reviewed changes

reflector: introduce watchList

2fed6b8

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from aac3889 to 3a238c3 Compare March 10, 2023 09:34

k8s-ci-robot added release-note Denotes a PR that will be considered when it comes time to generate release notes. and removed do-not-merge/release-note-label-needed Indicates that a PR should not merge because it's missing one of the release note labels. labels Mar 10, 2023

p0lyn0mial changed the title ~~reflector gets a stream of data~~ client-go: Add support for API streaming to the reflector Mar 10, 2023

wojtek-t reviewed Mar 10, 2023

View reviewed changes

reflector: use watchlist

51ec730

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from 3a238c3 to 9d872db Compare March 10, 2023 11:14

p0lyn0mial added 2 commits March 10, 2023 12:16

reflector watchlist tests

194ffd7

cache/controller: Add ENABLE_CLIENT_GO_WATCH_LIST_ALPHA

966b26d

p0lyn0mial force-pushed the upstream-reflector-gets-stream branch from 9d872db to 966b26d Compare March 10, 2023 11:16

k8s-ci-robot added the lgtm "Looks good to me", indicates that a PR is ready to be merged. label Mar 10, 2023

k8s-ci-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Mar 10, 2023

k8s-ci-robot merged commit 90c3232 into kubernetes:master Mar 10, 2023

k8s-ci-robot added this to the v1.27 milestone Mar 10, 2023

wojtek-t mentioned this pull request Mar 14, 2023

Allow informers for getting a stream of data instead of chunking kubernetes/enhancements#3157

Open

8 tasks

client-go: Add support for API streaming to the reflector #110772

client-go: Add support for API streaming to the reflector #110772

Conversation

p0lyn0mial commented Jun 24, 2022 • edited

What type of PR is this?

What this PR does / why we need it:

Which issue(s) this PR fixes:

Special notes for your reviewer:

Does this PR introduce a user-facing change?

Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

p0lyn0mial commented Jun 24, 2022

leilajal commented Jun 30, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wojtek-t left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wojtek-t left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wojtek-t commented Mar 10, 2023

k8s-ci-robot commented Mar 10, 2023

k8s-ci-robot commented Mar 10, 2023

k8s-ci-robot commented Mar 10, 2023 • edited

p0lyn0mial commented Mar 10, 2023

p0lyn0mial commented Jun 24, 2022 •

edited

k8s-ci-robot commented Mar 10, 2023 •

edited