Merge pull request #165 from pwittrock/master

Polish Informers for Kubernetes API types

cmd/apiregister-gen/generators/controller_generator.go

@@ -95,7 +95,8 @@ func New{{.Target.Kind}}Controller(config *rest.Config, si *sharedinformers.Shar
 		c.Informers.WorkerQueues = map[string]*controller.QueueWorker{}
 	}
 	c.Informers.WorkerQueues["{{.Target.Kind}}"] = queue
-	si.Factory.{{title .Target.Group}}().{{title .Target.Version}}().{{title .Resource}}().Informer().AddEventHandler(&controller.QueueingEventHandler{q, nil})
+	si.Factory.{{title .Target.Group}}().{{title .Target.Version}}().{{title .Resource}}().Informer().
+        AddEventHandler(&controller.QueueingEventHandler{q, nil, false})
 	return c
 }
 

cmd/apiserver-boot/boot/create/resource.go

@@ -157,6 +157,9 @@ func createResource(boilerplate string) {
 		}
 	}
 
+	path = filepath.Join(dir, "pkg", "controller", "sharedinformers", "informers.go")
+	created = util.WriteIfNotFound(path, "sharedinformer-template", sharedInformersTemplate, a)
+
 	if found {
 		os.Exit(-1)
 	}
@@ -343,6 +346,28 @@ var _ = Describe("{{.Kind}}", func() {
 })
 `
 
+var sharedInformersTemplate = `
+{{.BoilerPlate}}
+
+package sharedinformers
+
+// SetupKubernetesTypes registers the config for watching Kubernetes types
+func (si *SharedInformers) SetupKubernetesTypes() bool {
+    // Set this to true to initial the ClientSet and InformerFactory for
+    // Kubernetes APIs (e.g. Deployment)
+	return false
+}
+
+// StartAdditionalInformers starts watching Deployments
+func (si *SharedInformers) StartAdditionalInformers(shutdown <-chan struct{}) {
+    // Start specific Kubernetes API informers here.  Note, it is only necessary
+    // to start 1 informer for each Kind. (e.g. only 1 Deployment informer)
+
+    // Uncomment this to start listening for Deployment Create / Update / Deletes
+    // go si.KubernetesFactory.Apps().V1beta1().Deployments().Informer().Run(shutdown)
+}
+`
+
 var resourceControllerTemplate = `
 {{.BoilerPlate}}
 

docs/watching_kubernetes_resources.md

@@ -0,0 +1,100 @@
+# Watching Kubernetes resources
+
+## Run the sharedinformers to watch the types
+
+Setup the informers and related objects that are shared across all controllers
+run by the controller manager.  The following steps are done once per-controller-manager
+(go binary), not once per-controller (go package).
+
+**Step 1:** If `pkg/controller/sharedinformer/informers.go` does not already exist, create it.
+
+This is automatically created for you by `apiserver-boot` when creating a resource for
+releases alpha.18+.
+
+**Step 2:** Enable watching Kubernetes types and initializing the ClientSet
+
+This will require that the controller-manager and apiserver are run as aggregated
+with a core apiserver, as the controller-manager will be watch both core and custom
+types.
+
+Setting this to true will also allow your controller to read/write to the core apiserver using the
+`ClientSet` passed into your controller's `Init` function through `si.KubernetesClientSet`.
+
+```go
+// SetupKubernetesTypes registers the config for watching Kubernetes types
+func (si *SharedInformers) SetupKubernetesTypes() bool {
+	return true
+}
+```
+
+**Step 3:** Start the informers for the resources you want to watch
+
+For each Kind you that want to get notified about when it is created/updated/deleted,
+`Run` a corresponding Informer in `StartAdditionalInformers`.
+
+**Note:** If you want to watch Deployments, you do not need to start informers for all
+group/versions.  You only need to watch 1 group/version and other Deployment group/versions
+will be converted to the group/version you are watching.
+
+```go
+// StartAdditionalInformers starts watching Deployments
+func (si *SharedInformers) StartAdditionalInformers(shutdown <-chan struct{}) {
+	go si.KubernetesFactory.Apps().V1beta1().Deployments().Informer().Run(shutdown)
+}
+```
+
+## Register your controller with the informer
+
+**Step 1:** Map the Kubernetes resource instance to the key of an instance of your resource
+
+When receiving a notification for the Kubernetes resource (e.g. Deployment) you want
+to run the `Reconcile` loop for your resource instance that owns the Kubernetes resource.
+If you write an `OwnerReference` for each of the Kubernetes resources created by your resource,
+you may look at that field to find the key for your resource.
+
+- Cast the argument to the type
+  - **Note:** double check the group/version are consistent between
+    - the informer you started in pkg/controller/sharedinformers/informers.go
+    - the type you cast the argument to
+    - the informer with which you register your reconcile loop (Step 2)
+
+```go
+func (c *FooControllerImpl) DeploymentToFoo(i interface{}) (string, error) {
+	d, _ := i.(*v1beta1.Deployment)
+	log.Printf("Deployment update: %v", d.Name)
+	if len(d.OwnerReferences) == 1 && d.OwnerReferences[0].Kind == "Foo" {
+		return d.Namespace + "/" + d.OwnerReferences[0].Name, nil
+	} else {
+		// Not owned
+		return "", nil
+	}
+}
+```
+
+**Step 2:** Register your resource Reconcile with the informer
+
+In your controller init function tie it all together:
+- The sharedinformer you started that watches for events
+- The conversion from a Deployment to the key of your resource
+- The reconcile function that takes the key of one of your resources
+
+```go
+func (c *FooControllerImpl) Init(
+	config *rest.Config,
+	si *sharedinformers.SharedInformers,
+	reconcileKey func(key string) error) {
+    ...
+	si.Watch(
+	    "FooDeployment",
+	    si.KubernetesFactory.Extensions().V1beta1().Deployments().Informer(),
+	    c.DeploymentToFoo, reconcileKey)
+}
+```
+
+### Create/delete/update the Kubernetes from your controller reconcile loop
+
+Use the si.KubernetesClientSet from within your controllers `Reconcile` function
+to update Kubernetes objects.
+
+**Note**: Consider using a `Lister` for reading and indexing cached objects to reduce load
+on the apiserver.

pkg/controller/controller.go

@@ -31,8 +31,9 @@ import (
 
 // QueueingEventHandler queues the key for the object on add and update events
 type QueueingEventHandler struct {
-	Queue    workqueue.RateLimitingInterface
-	ObjToKey func(obj interface{}) (string, error)
+	Queue         workqueue.RateLimitingInterface
+	ObjToKey      func(obj interface{}) (string, error)
+	EnqueueDelete bool
 }
 
 func (c *QueueingEventHandler) enqueue(obj interface{}) {
@@ -60,6 +61,9 @@ func (c *QueueingEventHandler) OnUpdate(oldObj, newObj interface{}) {
 
 func (c *QueueingEventHandler) OnDelete(obj interface{}) {
 	glog.V(6).Infof("Delete event for %+v\n", obj)
+	if c.EnqueueDelete {
+		c.enqueue(obj)
+	}
 }
 
 // QueueWorker continuously runs a Reconcile function against a message Queue

pkg/controller/informers.go

@@ -73,7 +73,7 @@ func (c *SharedInformersDefaults) Watch(
 		c.WorkerQueues = map[string]*QueueWorker{}
 	}
 	c.WorkerQueues[name] = queue
-	i.AddEventHandler(&QueueingEventHandler{q, f})
+	i.AddEventHandler(&QueueingEventHandler{q, f, true})
 }
 
 func NewConfig(kubeconfig string) (*rest.Config, error) {