🌱 inmemory: fix watch to continue serving based on resourceVersion parameter #11695
+143
−35
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -20,20 +20,21 @@ import ( | |||||
| "context" | ||||||
| "fmt" | ||||||
| "net/http" | ||||||
| "strconv" | ||||||
| "time" | ||||||
|
|
||||||
| "github.com/emicklei/go-restful/v3" | ||||||
| "github.com/pkg/errors" | ||||||
| "k8s.io/apimachinery/pkg/runtime/schema" | ||||||
| "k8s.io/apimachinery/pkg/watch" | ||||||
| ctrl "sigs.k8s.io/controller-runtime" | ||||||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||||||
| ) | ||||||
|
|
||||||
| // Event records a lifecycle event for a Kubernetes object. | ||||||
| type Event struct { | ||||||
| Type watch.EventType `json:"type,omitempty"` | ||||||
| Object client.Object `json:"object,omitempty"` | ||||||
| } | ||||||
|
|
||||||
| // WatchEventDispatcher dispatches events for a single resourceGroup. | ||||||
|
|
@@ -88,13 +89,15 @@ func (m *WatchEventDispatcher) OnGeneric(resourceGroup string, o client.Object) | |||||
|
|
||||||
| func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful.Response, resourceGroup string, gvk schema.GroupVersionKind) (reterr error) { | ||||||
| ctx := req.Request.Context() | ||||||
| log := h.log.WithValues("resourceGroup", resourceGroup, "gvk", gvk.String()) | ||||||
| ctx = ctrl.LoggerInto(ctx, log) | ||||||
| queryTimeout := req.QueryParameter("timeoutSeconds") | ||||||
| resourceVersion := req.QueryParameter("resourceVersion") | ||||||
| c := h.manager.GetCache() | ||||||
| i, err := c.GetInformerForKind(ctx, gvk) | ||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
| // With an unbuffered event channel RemoveEventHandler could be blocked because it requires a lock on the informer. | ||||||
| // When Run stops reading from the channel the informer could be blocked with an unbuffered chanel and then RemoveEventHandler never goes through. | ||||||
| // 1000 is used to avoid deadlocks in clusters with a higher number of Machines/Nodes. | ||||||
|
|
@@ -115,7 +118,12 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful. | |||||
| L: | ||||||
| for { | ||||||
| select { | ||||||
| case event, ok := <-events: | ||||||
| if !ok { | ||||||
| // End of results. | ||||||
| break L | ||||||
| } | ||||||
| log.V(4).Info("Missed event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion()) | ||||||
| default: | ||||||
| break L | ||||||
| } | ||||||
|
|
@@ -124,11 +132,49 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful. | |||||
| // Note: After we removed the handler, no new events will be written to the events channel. | ||||||
| }() | ||||||
|
|
||||||
| // Get at client to the resource group and list all relevant objects. | ||||||
| inmemoryClient := h.manager.GetResourceGroup(resourceGroup).GetClient() | ||||||
| list, err := h.apiV1list(ctx, req, gvk, inmemoryClient) | ||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
|
|
||||||
| // If resourceVersion was set parse to uint64 which is the representation in the simulated apiserver. | ||||||
| var parsedResourceVersion uint64 | ||||||
| if resourceVersion != "" { | ||||||
| parsedResourceVersion, err = strconv.ParseUint(resourceVersion, 10, 64) | ||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| initialEvents := []Event{} | ||||||
|
|
||||||
| // Loop over all items and fill the list of events with objects which have a newer resourceVersion. | ||||||
| for _, obj := range list.Items { | ||||||
| if resourceVersion != "" { | ||||||
| objResourceVersion, err := strconv.ParseUint(obj.GetResourceVersion(), 10, 64) | ||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
| if objResourceVersion <= parsedResourceVersion { | ||||||
chrischdi marked this conversation as resolved.
Show resolved
Hide resolved
|
||||||
| continue | ||||||
| } | ||||||
| } | ||||||
| eventType := watch.Modified | ||||||
| // kube-apiserver emits all events as ADDED when no resourceVersion is given. | ||||||
| if obj.GetGeneration() == 1 || resourceVersion == "" { | ||||||
| eventType = watch.Added | ||||||
| } | ||||||
| initialEvents = append(initialEvents, Event{Type: eventType, Object: &obj}) | ||||||
| } | ||||||
|
|
||||||
| return watcher.Run(ctx, queryTimeout, initialEvents, resp) | ||||||
| } | ||||||
|
|
||||||
| // Run serves a series of encoded events via HTTP with Transfer-Encoding: chunked. | ||||||
| func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, initialEvents []Event, w http.ResponseWriter) error { | ||||||
| log := ctrl.LoggerFrom(ctx) | ||||||
| flusher, ok := w.(http.Flusher) | ||||||
| if !ok { | ||||||
| return errors.New("can't start Watch: can't get http.Flusher") | ||||||
|
|
@@ -139,6 +185,16 @@ func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, w http.R | |||||
| } | ||||||
| w.Header().Set("Transfer-Encoding", "chunked") | ||||||
| w.WriteHeader(http.StatusOK) | ||||||
|
There was a problem hiding this comment. Is it okay that we don't flush after this like before? |
||||||
|
|
||||||
| // 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()) | ||||||
|
There was a problem hiding this comment.
Suggested change
There was a problem hiding this comment. what about using "initial event" instead of "old event"? |
||||||
| _ = resp.WriteErrorString(http.StatusInternalServerError, err.Error()) | ||||||
| } else { | ||||||
| log.V(4).Info("Wrote old event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion()) | ||||||
| } | ||||||
| } | ||||||
| flusher.Flush() | ||||||
|
|
||||||
| timeoutTimer, seconds, err := setTimer(timeout) | ||||||
|
|
@@ -149,6 +205,18 @@ func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, w http.R | |||||
| ctx, cancel := context.WithTimeout(ctx, seconds) | ||||||
| defer cancel() | ||||||
| defer timeoutTimer.Stop() | ||||||
|
|
||||||
| // 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) | ||||||
|
There was a problem hiding this comment. 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 |
||||||
| if err != nil { | ||||||
| return err | ||||||
| } | ||||||
| minResourceVersion++ | ||||||
| } | ||||||
|
|
||||||
| var objResourceVersion uint64 | ||||||
| for { | ||||||
| select { | ||||||
| case <-ctx.Done(): | ||||||
|
|
@@ -160,8 +228,25 @@ func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, w http.R | |||||
| // End of results. | ||||||
| return nil | ||||||
| } | ||||||
|
|
||||||
| // Parse and check if the object has a higher resource version than we allow. | ||||||
| objResourceVersion, err = strconv.ParseUint(event.Object.GetResourceVersion(), 10, 64) | ||||||
| if err != nil { | ||||||
| log.Error(err, "Parsing object resource version", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion()) | ||||||
| _ = resp.WriteErrorString(http.StatusInternalServerError, err.Error()) | ||||||
| continue | ||||||
| } | ||||||
|
|
||||||
| // Skip objects which were already written. | ||||||
| if objResourceVersion < minResourceVersion { | ||||||
| continue | ||||||
| } | ||||||
|
|
||||||
| if err := resp.WriteEntity(event); err != nil { | ||||||
| log.Error(err, "Writing event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion()) | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| _ = resp.WriteErrorString(http.StatusInternalServerError, err.Error()) | ||||||
| } else { | ||||||
| log.V(4).Info("Wrote event", "eventType", event.Type, "objectName", event.Object.GetName(), "resourceVersion", event.Object.GetResourceVersion()) | ||||||
| } | ||||||
| if len(m.events) == 0 { | ||||||
| flusher.Flush() | ||||||
|
|
||||||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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)