forked from kubernetes/perf-tests
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Increase page size in slo-monitor to 5000 (attempt kubernetes#2).
- Loading branch information
Showing
4 changed files
with
214 additions
and
22 deletions.
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
197 changes: 197 additions & 0 deletions
197
slo-monitor/src/third_party/forked/k8s.io/client-go/tools/cache/controller.go
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 |
|---|---|---|
| @@ -0,0 +1,197 @@ | ||
| /* | ||
| Copyright 2020 The Kubernetes Authors. | ||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package cache | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "k8s.io/apimachinery/pkg/runtime" | ||
| utilruntime "k8s.io/apimachinery/pkg/util/runtime" | ||
| "k8s.io/apimachinery/pkg/util/wait" | ||
| "k8s.io/client-go/tools/cache" | ||
| ) | ||
|
|
||
| ////////////////////////////////////////////////// | ||
| // This is a fork of Config and Controller structs from k8s.io/client-go/tools/cache/controller.go with | ||
| // cherry pick of https://github.com/kubernetes/kubernetes/pull/94363. | ||
| // TODO(maciejborsz): Get rid of it once https://github.com/kubernetes/kubernetes/pull/94363 is available | ||
| // in version of client-go used in slo-monitor. | ||
| ///////////////////////////////////////////////// | ||
|
|
||
| // Config contains all the settings for a Controller. | ||
| type Config struct { | ||
| // The queue for your objects - has to be a DeltaFIFO due to | ||
| // assumptions in the implementation. Your Process() function | ||
| // should accept the output of this Queue's Pop() method. | ||
| cache.Queue | ||
|
|
||
| // Something that can list and watch your objects. | ||
| cache.ListerWatcher | ||
|
|
||
| // Something that can process a popped Deltas. | ||
| Process ProcessFunc | ||
|
|
||
| // ObjectType is an example object of the type this controller is | ||
| // expected to handle. Only the type needs to be right, except | ||
| // that when that is `unstructured.Unstructured` the object's | ||
| // `"apiVersion"` and `"kind"` must also be right. | ||
| ObjectType runtime.Object | ||
|
|
||
| // FullResyncPeriod is the period at which ShouldResync is considered. | ||
| FullResyncPeriod time.Duration | ||
|
|
||
| // ShouldResync is periodically used by the reflector to determine | ||
| // whether to Resync the Queue. If ShouldResync is `nil` or | ||
| // returns true, it means the reflector should proceed with the | ||
| // resync. | ||
| ShouldResync ShouldResyncFunc | ||
|
|
||
| // If true, when Process() returns an error, re-enqueue the object. | ||
| // TODO: add interface to let you inject a delay/backoff or drop | ||
| // the object completely if desired. Pass the object in | ||
| // question to this interface as a parameter. This is probably moot | ||
| // now that this functionality appears at a higher level. | ||
| RetryOnError bool | ||
|
|
||
| // XXX: Not possible to implement in a fork. | ||
| // Called whenever the ListAndWatch drops the connection with an error. | ||
| // WatchErrorHandler cache.WatchErrorHandler | ||
|
|
||
| // WatchListPageSize is the requested chunk size of initial and relist watch lists. | ||
| WatchListPageSize int64 | ||
| } | ||
|
|
||
| // ShouldResyncFunc is a type of function that indicates if a reflector should perform a | ||
| // resync or not. It can be used by a shared informer to support multiple event handlers with custom | ||
| // resync periods. | ||
| type ShouldResyncFunc func() bool | ||
|
|
||
| // ProcessFunc processes a single object. | ||
| type ProcessFunc func(obj interface{}) error | ||
|
|
||
| // `*controller` implements Controller | ||
| type controller struct { | ||
| config Config | ||
| reflector *cache.Reflector | ||
| reflectorMutex sync.RWMutex | ||
| // XXX: Not possible to implement in a fork. | ||
| // clock clock.Clock | ||
| } | ||
|
|
||
| // Controller is a low-level controller that is parameterized by a | ||
| // Config and used in sharedIndexInformer. | ||
| type Controller interface { | ||
| // Run does two things. One is to construct and run a Reflector | ||
| // to pump objects/notifications from the Config's ListerWatcher | ||
| // to the Config's Queue and possibly invoke the occasional Resync | ||
| // on that Queue. The other is to repeatedly Pop from the Queue | ||
| // and process with the Config's ProcessFunc. Both of these | ||
| // continue until `stopCh` is closed. | ||
| Run(stopCh <-chan struct{}) | ||
|
|
||
| // HasSynced delegates to the Config's Queue | ||
| HasSynced() bool | ||
|
|
||
| // LastSyncResourceVersion delegates to the Reflector when there | ||
| // is one, otherwise returns the empty string | ||
| LastSyncResourceVersion() string | ||
| } | ||
|
|
||
| // New makes a new Controller from the given Config. | ||
| func New(c *Config) Controller { | ||
| ctlr := &controller{ | ||
| config: *c, | ||
| // XXX: Not possible to implement in a fork. | ||
| // clock: &clock.RealClock{}, | ||
| } | ||
| return ctlr | ||
| } | ||
|
|
||
| // Run begins processing items, and will continue until a value is sent down stopCh or it is closed. | ||
| // It's an error to call Run more than once. | ||
| // Run blocks; call via go. | ||
| func (c *controller) Run(stopCh <-chan struct{}) { | ||
| defer utilruntime.HandleCrash() | ||
| go func() { | ||
| <-stopCh | ||
| c.config.Queue.Close() | ||
| }() | ||
| r := cache.NewReflector( | ||
| c.config.ListerWatcher, | ||
| c.config.ObjectType, | ||
| c.config.Queue, | ||
| c.config.FullResyncPeriod, | ||
| ) | ||
| r.ShouldResync = c.config.ShouldResync | ||
| r.WatchListPageSize = c.config.WatchListPageSize | ||
|
|
||
| // XXX: Not possible to implement in a fork. | ||
| // r.clock = c.clock | ||
| // if c.config.WatchErrorHandler != nil { | ||
| // r.watchErrorHandler = c.config.WatchErrorHandler | ||
| // } | ||
|
|
||
| c.reflectorMutex.Lock() | ||
| c.reflector = r | ||
| c.reflectorMutex.Unlock() | ||
|
|
||
| var wg wait.Group | ||
| defer wg.Wait() | ||
|
|
||
| wg.StartWithChannel(stopCh, r.Run) | ||
|
|
||
| wait.Until(c.processLoop, time.Second, stopCh) | ||
| } | ||
|
|
||
| // Returns true once this controller has completed an initial resource listing | ||
| func (c *controller) HasSynced() bool { | ||
| return c.config.Queue.HasSynced() | ||
| } | ||
|
|
||
| func (c *controller) LastSyncResourceVersion() string { | ||
| c.reflectorMutex.RLock() | ||
| defer c.reflectorMutex.RUnlock() | ||
| if c.reflector == nil { | ||
| return "" | ||
| } | ||
| return c.reflector.LastSyncResourceVersion() | ||
| } | ||
|
|
||
| // processLoop drains the work queue. | ||
| // TODO: Consider doing the processing in parallel. This will require a little thought | ||
| // to make sure that we don't end up processing the same object multiple times | ||
| // concurrently. | ||
| // | ||
| // TODO: Plumb through the stopCh here (and down to the queue) so that this can | ||
| // actually exit when the controller is stopped. Or just give up on this stuff | ||
| // ever being stoppable. Converting this whole package to use Context would | ||
| // also be helpful. | ||
| func (c *controller) processLoop() { | ||
| for { | ||
| obj, err := c.config.Queue.Pop(cache.PopProcessFunc(c.config.Process)) | ||
| if err != nil { | ||
| if err == cache.FIFOClosedError { | ||
| return | ||
| } | ||
| if c.config.RetryOnError { | ||
| // This is the safe way to re-enqueue. | ||
| c.config.Queue.AddIfNotPresent(obj) | ||
| } | ||
| } | ||
| } | ||
| } |