Documentation
¶
Overview ¶
Package framework implements all the grunt work involved in running a simple controller.
Example ¶
package main import ( "fmt" "time" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/util/sets" ) func main() { // source simulates an apiserver object endpoint. source := framework.NewFakeControllerSource() // This will hold the downstream state, as we know it. downstream := cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc) // This will hold incoming changes. Note how we pass downstream in as a // KeyLister, that way resync operations will result in the correct set // of update/delete deltas. fifo := cache.NewDeltaFIFO(cache.MetaNamespaceKeyFunc, nil, downstream) // Let's do threadsafe output to get predictable test results. deletionCounter := make(chan string, 1000) cfg := &framework.Config{ Queue: fifo, ListerWatcher: source, ObjectType: &api.Pod{}, FullResyncPeriod: time.Millisecond * 100, RetryOnError: false, // Let's implement a simple controller that just deletes // everything that comes in. Process: func(obj interface{}) error { // Obj is from the Pop method of the Queue we make above. newest := obj.(cache.Deltas).Newest() if newest.Type != cache.Deleted { // Update our downstream store. err := downstream.Add(newest.Object) if err != nil { return err } // Delete this object. source.Delete(newest.Object.(runtime.Object)) } else { // Update our downstream store. err := downstream.Delete(newest.Object) if err != nil { return err } // fifo's KeyOf is easiest, because it handles // DeletedFinalStateUnknown markers. key, err := fifo.KeyOf(newest.Object) if err != nil { return err } // Report this deletion. deletionCounter <- key } return nil }, } // Create the controller and run it until we close stop. stop := make(chan struct{}) defer close(stop) go framework.New(cfg).Run(stop) // Let's add a few objects to the source. testIDs := []string{"a-hello", "b-controller", "c-framework"} for _, name := range testIDs { // Note that these pods are not valid-- the fake source doesn't // call validation or anything. source.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: name}}) } // Let's wait for the controller to process the things we just added. outputSet := sets.String{} for i := 0; i < len(testIDs); i++ { outputSet.Insert(<-deletionCounter) } for _, key := range outputSet.List() { fmt.Println(key) } }
Output: a-hello b-controller c-framework
Index ¶
- func DeletionHandlingMetaNamespaceKeyFunc(obj interface{}) (string, error)
- type Config
- type Controller
- type ControllerInterface
- type FakeControllerSource
- func (f *FakeControllerSource) Add(obj runtime.Object)
- func (f *FakeControllerSource) AddDropWatch(obj runtime.Object)
- func (f *FakeControllerSource) Change(e watch.Event, watchProbability float64)
- func (f *FakeControllerSource) Delete(lastValue runtime.Object)
- func (f *FakeControllerSource) DeleteDropWatch(lastValue runtime.Object)
- func (f *FakeControllerSource) List(options api.ListOptions) (runtime.Object, error)
- func (f *FakeControllerSource) Modify(obj runtime.Object)
- func (f *FakeControllerSource) ModifyDropWatch(obj runtime.Object)
- func (f *FakeControllerSource) Shutdown()
- func (f *FakeControllerSource) Watch(options api.ListOptions) (watch.Interface, error)
- type FakePVCControllerSource
- type FakePVControllerSource
- type ProcessFunc
- type ResourceEventHandler
- type ResourceEventHandlerFuncs
- type SharedIndexInformer
- type SharedInformer
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DeletionHandlingMetaNamespaceKeyFunc ¶
DeletionHandlingMetaNamespaceKeyFunc checks for cache.DeletedFinalStateUnknown objects before calling cache.MetaNamespaceKeyFunc.
Types ¶
type Config ¶
type Config struct { // The queue for your objects; either a cache.FIFO or // a cache.DeltaFIFO. Your Process() function should accept // the output of this Oueue's Pop() method. cache.Queue // Something that can list and watch your objects. cache.ListerWatcher // Something that can process your objects. Process ProcessFunc // The type of your objects. ObjectType runtime.Object // Reprocess everything at least this often. // Note that if it takes longer for you to clear the queue than this // period, you will end up processing items in the order determined // by cache.FIFO.Replace(). Currently, this is random. If this is a // problem, we can change that replacement policy to append new // things to the end of the queue instead of replacing the entire // queue. FullResyncPeriod time.Duration // 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. RetryOnError bool }
Config contains all the settings for a Controller.
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller is a generic controller framework.
func NewIndexerInformer ¶ added in v0.18.2
func NewIndexerInformer( lw cache.ListerWatcher, objType runtime.Object, resyncPeriod time.Duration, h ResourceEventHandler, indexers cache.Indexers, ) (cache.Indexer, *Controller)
NewIndexerInformer returns a cache.Indexer and a controller for populating the index while also providing event notifications. You should only used the returned cache.Index for Get/List operations; Add/Modify/Deletes will cause the event notifications to be faulty.
Parameters:
- lw is list and watch functions for the source of the resource you want to be informed of.
- objType is an object of the type that you expect to receive.
- resyncPeriod: if non-zero, will re-list this often (you will get OnUpdate calls, even if nothing changed). Otherwise, re-list will be delayed as long as possible (until the upstream source closes the watch or times out, or you stop the controller).
- h is the object you want notifications sent to.
func NewInformer ¶
func NewInformer( lw cache.ListerWatcher, objType runtime.Object, resyncPeriod time.Duration, h ResourceEventHandler, ) (cache.Store, *Controller)
NewInformer returns a cache.Store and a controller for populating the store while also providing event notifications. You should only used the returned cache.Store for Get/List operations; Add/Modify/Deletes will cause the event notifications to be faulty.
Parameters:
- lw is list and watch functions for the source of the resource you want to be informed of.
- objType is an object of the type that you expect to receive.
- resyncPeriod: if non-zero, will re-list this often (you will get OnUpdate calls, even if nothing changed). Otherwise, re-list will be delayed as long as possible (until the upstream source closes the watch or times out, or you stop the controller).
- h is the object you want notifications sent to.
func (*Controller) HasSynced ¶ added in v0.18.2
func (c *Controller) HasSynced() bool
Returns true once this controller has completed an initial resource listing
func (*Controller) Requeue ¶ added in v1.2.3
func (c *Controller) Requeue(obj interface{}) error
Requeue adds the provided object back into the queue if it does not already exist.
func (*Controller) Run ¶
func (c *Controller) Run(stopCh <-chan struct{})
Run begins processing items, and will continue until a value is sent down stopCh. It's an error to call Run more than once. Run blocks; call via go.
type ControllerInterface ¶ added in v1.3.0
type ControllerInterface interface { Run(stopCh <-chan struct{}) HasSynced() bool }
TODO make the "Controller" private, and convert all references to use ControllerInterface instead
type FakeControllerSource ¶
type FakeControllerSource struct { Items map[nnu]runtime.Object Broadcaster *watch.Broadcaster // contains filtered or unexported fields }
FakeControllerSource implements listing/watching for testing.
func NewFakeControllerSource ¶
func NewFakeControllerSource() *FakeControllerSource
func (*FakeControllerSource) Add ¶
func (f *FakeControllerSource) Add(obj runtime.Object)
Add adds an object to the set and sends an add event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) AddDropWatch ¶
func (f *FakeControllerSource) AddDropWatch(obj runtime.Object)
AddDropWatch adds an object to the set but forgets to send an add event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) Change ¶
func (f *FakeControllerSource) Change(e watch.Event, watchProbability float64)
Change records the given event (setting the object's resource version) and sends a watch event with the specified probability.
func (*FakeControllerSource) Delete ¶
func (f *FakeControllerSource) Delete(lastValue runtime.Object)
Delete deletes an object from the set and sends a delete event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) DeleteDropWatch ¶
func (f *FakeControllerSource) DeleteDropWatch(lastValue runtime.Object)
DeleteDropWatch deletes an object from the set but forgets to send a delete event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) List ¶
func (f *FakeControllerSource) List(options api.ListOptions) (runtime.Object, error)
List returns a list object, with its resource version set.
func (*FakeControllerSource) Modify ¶
func (f *FakeControllerSource) Modify(obj runtime.Object)
Modify updates an object in the set and sends a modified event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) ModifyDropWatch ¶
func (f *FakeControllerSource) ModifyDropWatch(obj runtime.Object)
ModifyDropWatch updates an object in the set but forgets to send a modify event to watchers. obj's ResourceVersion is set.
func (*FakeControllerSource) Shutdown ¶ added in v1.2.3
func (f *FakeControllerSource) Shutdown()
Shutdown closes the underlying broadcaster, waiting for events to be delivered. It's an error to call any method after calling shutdown. This is enforced by Shutdown() leaving f locked.
func (*FakeControllerSource) Watch ¶
func (f *FakeControllerSource) Watch(options api.ListOptions) (watch.Interface, error)
Watch returns a watch, which will be pre-populated with all changes after resourceVersion.
type FakePVCControllerSource ¶ added in v1.3.0
type FakePVCControllerSource struct {
FakeControllerSource
}
func NewFakePVCControllerSource ¶ added in v1.3.0
func NewFakePVCControllerSource() *FakePVCControllerSource
func (*FakePVCControllerSource) List ¶ added in v1.3.0
func (f *FakePVCControllerSource) List(options api.ListOptions) (runtime.Object, error)
List returns a list object, with its resource version set.
type FakePVControllerSource ¶ added in v1.3.0
type FakePVControllerSource struct {
FakeControllerSource
}
func NewFakePVControllerSource ¶ added in v1.3.0
func NewFakePVControllerSource() *FakePVControllerSource
func (*FakePVControllerSource) List ¶ added in v1.3.0
func (f *FakePVControllerSource) List(options api.ListOptions) (runtime.Object, error)
List returns a list object, with its resource version set.
type ProcessFunc ¶
type ProcessFunc func(obj interface{}) error
ProcessFunc processes a single object.
type ResourceEventHandler ¶
type ResourceEventHandler interface { OnAdd(obj interface{}) OnUpdate(oldObj, newObj interface{}) OnDelete(obj interface{}) }
ResourceEventHandler can handle notifications for events that happen to a resource. The events are informational only, so you can't return an error.
- OnAdd is called when an object is added.
- OnUpdate is called when an object is modified. Note that oldObj is the last known state of the object-- it is possible that several changes were combined together, so you can't use this to see every single change. OnUpdate is also called when a re-list happens, and it will get called even if nothing changed. This is useful for periodically evaluating or syncing something.
- OnDelete will get the final state of the item if it is known, otherwise it will get an object of type cache.DeletedFinalStateUnknown. This can happen if the watch is closed and misses the delete event and we don't notice the deletion until the subsequent re-list.
type ResourceEventHandlerFuncs ¶
type ResourceEventHandlerFuncs struct { AddFunc func(obj interface{}) UpdateFunc func(oldObj, newObj interface{}) DeleteFunc func(obj interface{}) }
ResourceEventHandlerFuncs is an adaptor to let you easily specify as many or as few of the notification functions as you want while still implementing ResourceEventHandler.
func (ResourceEventHandlerFuncs) OnAdd ¶
func (r ResourceEventHandlerFuncs) OnAdd(obj interface{})
OnAdd calls AddFunc if it's not nil.
func (ResourceEventHandlerFuncs) OnDelete ¶
func (r ResourceEventHandlerFuncs) OnDelete(obj interface{})
OnDelete calls DeleteFunc if it's not nil.
func (ResourceEventHandlerFuncs) OnUpdate ¶
func (r ResourceEventHandlerFuncs) OnUpdate(oldObj, newObj interface{})
OnUpdate calls UpdateFunc if it's not nil.
type SharedIndexInformer ¶ added in v1.3.0
type SharedIndexInformer interface { SharedInformer // AddIndexers add indexers to the informer before it starts. }
func NewSharedIndexInformer ¶ added in v1.3.0
func NewSharedIndexInformer(lw cache.ListerWatcher, objType runtime.Object, resyncPeriod time.Duration, indexers cache.Indexers) SharedIndexInformer
NewSharedIndexInformer creates a new instance for the listwatcher. TODO: create a cache/factory of these at a higher level for the list all, watch all of a given resource that can be shared amongst all consumers.
type SharedInformer ¶ added in v1.3.0
type SharedInformer interface { // events to a single handler are delivered sequentially, but there is no coordination between different handlers // You may NOT add a handler *after* the SharedInformer is running. That will result in an error being returned. // TODO we should try to remove this restriction eventually. // GetController gives back a synthetic interface that "votes" to start the informer }
if you use this, there is one behavior change compared to a standard Informer. When you receive a notification, the cache will be AT LEAST as fresh as the notification, but it MAY be more fresh. You should NOT depend on the contents of the cache exactly matching the notification you've received in handler functions. If there was a create, followed by a delete, the cache may NOT have your item. This has advantages over the broadcaster since it allows us to share a common cache across many controllers. Extending the broadcaster would have required us keep duplicate caches for each watch.
func NewSharedInformer ¶ added in v1.3.0
func NewSharedInformer(lw cache.ListerWatcher, objType runtime.Object, resyncPeriod time.Duration) SharedInformer
NewSharedInformer creates a new instance for the listwatcher. TODO: create a cache/factory of these at a higher level for the list all, watch all of a given resource that can be shared amongst all consumers.