[Enhancement] Use Streaming Lists when available #244
Conversation
Kubernetes version that has the WatchList version Signed-off-by: Daniel Kiptoon <[email protected]>
Interface to receive updates from Kubernetes on the given API types and publishes them to the broker Signed-off-by: Daniel Kiptoon <[email protected]>
on configuration Signed-off-by: Daniel Kiptoon <[email protected]>
underlying functions Signed-off-by: Daniel Kiptoon <[email protected]>
utilize Signed-off-by: Daniel Kiptoon <[email protected]>
| func New(log logger.Handler, informer dynamicinformer.DynamicSharedInformerFactory, broker broker.Handler, plConfigs map[string]internalconfig.PipelineConfigs, stopChan chan struct{}) *pipeline.Pipeline { | ||
| func New(log logger.Handler, informer dynamicinformer.DynamicSharedInformerFactory, broker broker.Handler, plConfigs map[string]internalconfig.PipelineConfigs, stopChan chan struct{}, dynamicKube dynamic.Interface) *pipeline.Pipeline { | ||
| // TODO: best way to check whether WatchList feature is enabled | ||
| watchList := true |
There was a problem hiding this comment.
I have looked around for ways of determining the enabled features in a Kubernetes cluster, but most knowledge revolves around looking at actual cluster configuration to see what has been enabled and what hasn't been enabled...I would appreciate any pointers to how I can determine programmatically which features have been enabled or not
There was a problem hiding this comment.
@KiptoonKipkurui I encourage you to review this PR (its in Rust), but potentially a great reference - http://gitea.lntu.edu.cn/youling/kube/commit/3af6b7f1df216380dd9c9bd13a163fe5281d67fa#diff-d4e0d370315574d0ee0602792eb3abfd2777d3c7
I assume that you have already seen this as well - https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/3157-watch-list/README.md
There was a problem hiding this comment.
Thank you for this resources, I will check them out
| // Launch the goroutine and pass the channel as an argument | ||
| wg.Add(1) | ||
| go w.backgroundWatchProcessor(watcher.ResultChan(), w.stopChan, &wg) | ||
| data := make(map[string]cache.Store) |
There was a problem hiding this comment.
This data which is essentially a cache used by the informer is used downstream. I would like to understand the potential downfall of not including it as none is accessible while using a watch
| w.log.Info("Received ADD event for: ", obj.(*unstructured.Unstructured).GetName(), "/", obj.(*unstructured.Unstructured).GetNamespace(), " of kind: ", obj.(*unstructured.Unstructured).GroupVersionKind().Kind) | ||
| // when an event is modified... | ||
| case watch.Modified: | ||
| err := w.publishItem(obj.(*unstructured.Unstructured), broker.Update, w.config) |
There was a problem hiding this comment.
For the Update functionality, using informers, one could get the older object and the new object and hence we could compare resource versions to determine whether indeed this object is a valid update or a repeated action. This is not available when using watchers
|
Using streaming lists boils down to adding the flag |
|
@KiptoonKipkurui, great presentation today, sir. |
leecalcote
requested review from
theBeginner86
and removed request for
Revolyssup and
humblenginr
|
@MUzairS15 this isn't ready for a master merge, it will break functionality |
|
Thanks for informing, reverting and reopening you can work on that PR. |
|
I breath again |
Description
This PR Addresses an issue of potential runaway usage of memory due to chucking of reported statuses while watching for Kubernetes resources. Kubernetes 1.27 introduces a WatchList feature still at the alpha stage that allows one to get statuses of objects being watched as a streaming list as opposed to current implementations that uses a list request first before a watch request. This is done by setting
sendInitialEvents=trueparam in a watch request.This PR fixes #221
Notes for Reviewers
This is still WIP and is allow continuous conversation to allow seamless integration with existing infrastructure
Signed commits