🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695

chrischdi · 2025-01-17T08:56:08Z

What this PR does / why we need it:

Watches on remote cluster expires every 10s #8893 leads to watches getting recreated every 10s
When using the inmemory provider, changes during recreation of a watch got lost
- e.g. Node got created in the simulated cluster
- e.g. Pd got created in the simulated cluster
This caused the controllers to require reaching the sync period (or e.g. in the Node missing case, MHC kicked in at 10 mins and the machine got recreated)
This lead to e.g. longer taking scale tests due to e.g. machines not getting ready because an event got missed

This PR has the following changes:

Change the way in-memory provider sets resourceVersions, to have one increasing resourceVersion per simulated cluster (similar to what k8s is doing)
use the resourceVersion parameter on watches (if given) to enqueue events for all objects which have a higher resourceVersion

Which issue(s) this PR fixes (optional, in fixes #<issue number>(, fixes #<issue_number>, ...) format, will close the issue(s) when PR gets merged):
Fixes #

/area provider/infrastructure-in-memory

k8s-ci-robot · 2025-01-17T08:56:36Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please ask for approval from chrischdi. For more information see the Code Review Process.

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

Needs approval from an approver in each of these files:

OWNERS

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

chrischdi · 2025-01-17T08:57:49Z

Kudos to @sbueringer for helping analysing and brainstorming to find the root cause.

chrischdi · 2025-01-17T09:03:10Z

/test help

k8s-ci-robot · 2025-01-17T09:03:13Z

@chrischdi: The specified target(s) for /test were not found.
The following commands are available to trigger required jobs:

/test pull-cluster-api-build-main

/test pull-cluster-api-e2e-blocking-main

/test pull-cluster-api-e2e-conformance-ci-latest-main

/test pull-cluster-api-e2e-conformance-main

/test pull-cluster-api-e2e-latestk8s-main

/test pull-cluster-api-e2e-main

/test pull-cluster-api-e2e-mink8s-main

/test pull-cluster-api-e2e-upgrade-1-32-1-33-main

/test pull-cluster-api-test-main

/test pull-cluster-api-test-mink8s-main

/test pull-cluster-api-verify-main

The following commands are available to trigger optional jobs:

/test pull-cluster-api-apidiff-main

Use /test all to run the following jobs that were automatically triggered:

pull-cluster-api-apidiff-main

pull-cluster-api-build-main

pull-cluster-api-e2e-blocking-main

pull-cluster-api-test-main

pull-cluster-api-verify-main

In response to this:

/test help

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-sigs/prow repository.

chrischdi · 2025-01-17T09:03:23Z

/test pull-cluster-api-e2e-main

test/infrastructure/inmemory/pkg/server/api/watch.go

chrischdi · 2025-01-17T13:39:53Z

/test pull-cluster-api-e2e-main

sbueringer · 2025-01-17T16:13:17Z

test/infrastructure/inmemory/pkg/server/api/watch.go

@@ -139,6 +198,16 @@ func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, w http.R
 	}
 	w.Header().Set("Transfer-Encoding", "chunked")
 	w.WriteHeader(http.StatusOK)


Is it okay that we don't flush after this like before?

sbueringer · 2025-01-17T16:13:59Z

test/infrastructure/inmemory/pkg/server/api/watch.go

+	// Write all initial events.
+	for _, event := range initialEvents {
+		if err := resp.WriteEntity(event); err != nil {
+			log.Error(err, "Writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())


Suggested change

log.Error(err, "Writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

log.Error(err, "Error writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

what about using "initial event" instead of "old event"?

sbueringer · 2025-01-17T16:17:01Z

test/infrastructure/inmemory/pkg/server/api/watch.go

 			if err := resp.WriteEntity(event); err != nil {
+				log.Error(err, "Writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())


Suggested change

log.Error(err, "Writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

log.Error(err, "Error writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

fabriziopandini

Great job in investigating and fixing this issue, kudos!

fabriziopandini · 2025-01-17T19:39:26Z

test/infrastructure/inmemory/pkg/runtime/cache/client.go

@@ -145,6 +145,9 @@ func (c *cache) List(resourceGroup string, list client.ObjectList, opts ...clien
 	if err := meta.SetList(list, items); err != nil {
 		return apierrors.NewInternalError(err)
 	}
+
+	list.SetResourceVersion(fmt.Sprintf("%d", tracker.lastResourceVersion))


Q: is it correct that list ResourceVersion in the system and not the last resource version in the list of items?
Q: is is correct set set this value non matter of it is a "plain" list or a lis watch?
(from a quick check with kubectl yes to both, but I like a confirmation)

fabriziopandini · 2025-01-17T19:51:39Z

test/infrastructure/inmemory/pkg/server/api/handler.go

+	}
+}
+
+func (h *apiServerHandler) apiV1list(ctx context.Context, req *restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {


might be

Suggested change

func (h *apiServerHandler) apiV1list(ctx context.Context, req *restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {

func (h *apiServerHandler) v1List(ctx context.Context, req *restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {

(easier to distinguish from apiV1List with upper L)

fabriziopandini · 2025-01-17T19:57:49Z

test/infrastructure/inmemory/pkg/server/api/watch.go

+	// Write all initial events.
+	for _, event := range initialEvents {
+		if err := resp.WriteEntity(event); err != nil {
+			log.Error(err, "Writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())


what about using "initial event" instead of "old event"?

fabriziopandini · 2025-01-17T20:03:03Z

test/infrastructure/inmemory/pkg/server/api/watch.go

+	// Determine the highest written resourceVersion so we can filter out duplicated events from the channel.
+	minResourceVersion := uint64(0)
+	if len(initialEvents) > 0 {
+		minResourceVersion, err = strconv.ParseUint(initialEvents[len(initialEvents)-1].Object.GetResourceVersion(), 10, 64)


Instead of assuming an ordering of events by resource version (which I'm not sure we are enforcing somewhere), what about computing minResourceVersion when we go through initialEvents in the for loop above

fix resourceVersion to increment across a tracker

e08e606

chrischdi added the tide/merge-method-squash Denotes a PR that should be squashed by tide when it merges. label Jan 17, 2025

k8s-ci-robot added area/provider/infrastructure-in-memory Issues or PRs related to the in-memory infrastructure provider cncf-cla: yes Indicates the PR's author has signed the CNCF CLA. size/L Denotes a PR that changes 100-499 lines, ignoring generated files. labels Jan 17, 2025

k8s-ci-robot requested review from jackfrancis and richardcase January 17, 2025 08:56

chrischdi force-pushed the pr-inmemory-fix-watch-resourceversion branch from 54e7e43 to 1b644a2 Compare January 17, 2025 08:56

chrischdi commented Jan 17, 2025

View reviewed changes

test/infrastructure/inmemory/pkg/server/api/watch.go Show resolved Hide resolved

sbueringer reviewed Jan 17, 2025

View reviewed changes

test/infrastructure/inmemory/pkg/server/api/watch.go Show resolved Hide resolved

test/infrastructure/inmemory/pkg/server/api/watch.go Outdated Show resolved Hide resolved

inmemory: implement continuing watch based on resourceVersion

3bac618

chrischdi force-pushed the pr-inmemory-fix-watch-resourceversion branch from 1b644a2 to 3bac618 Compare January 17, 2025 13:18

test fixes

cce5bde

chrischdi added 3 commits January 17, 2025 15:18

inmemory: add some debug log lines for watches

b4ed0fc

stream all events if resourceVersion is not given

42309df

fixes

3054216

sbueringer reviewed Jan 17, 2025

View reviewed changes

chrischdi added 3 commits January 17, 2025 17:31

set proper resourceVersion on list calls

caf66a6

fixup tests

e2d871c

another fixes

36fa992

fabriziopandini reviewed Jan 17, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695

🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695

chrischdi commented Jan 17, 2025

k8s-ci-robot commented Jan 17, 2025

chrischdi commented Jan 17, 2025

chrischdi commented Jan 17, 2025

k8s-ci-robot commented Jan 17, 2025

chrischdi commented Jan 17, 2025

chrischdi commented Jan 17, 2025

sbueringer Jan 17, 2025

sbueringer Jan 17, 2025

fabriziopandini Jan 17, 2025

sbueringer Jan 17, 2025

fabriziopandini left a comment

fabriziopandini Jan 17, 2025

fabriziopandini Jan 17, 2025

fabriziopandini Jan 17, 2025

fabriziopandini Jan 17, 2025

	log.Error(err, "Writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())
	log.Error(err, "Error writing old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

		if err := resp.WriteEntity(event); err != nil {
		log.Error(err, "Writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

	log.Error(err, "Writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())
	log.Error(err, "Error writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion())

	func (h apiServerHandler) apiV1list(ctx context.Context, req restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {
	func (h apiServerHandler) v1List(ctx context.Context, req restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {

🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695

Are you sure you want to change the base?

🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695

Conversation

chrischdi commented Jan 17, 2025

k8s-ci-robot commented Jan 17, 2025

chrischdi commented Jan 17, 2025

chrischdi commented Jan 17, 2025

k8s-ci-robot commented Jan 17, 2025

chrischdi commented Jan 17, 2025

chrischdi commented Jan 17, 2025

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

fabriziopandini 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