implement InjectRecorder

Author: lichuqiang

pkg/client/config/config.go

@@ -102,12 +102,22 @@ func GetKubernetesClientSet() (*kubernetes.Clientset, error) {
 // in cluster and use the cluster provided kubeconfig.
 //
 // Will log.Fatal if KubernetesInformers cannot be created
-func GetKubernetesClientSetOrDie() (*kubernetes.Clientset, error) {
+func GetKubernetesClientSetOrDie() *kubernetes.Clientset {
 	cs, err := GetKubernetesClientSet()
 	if err != nil {
 		log.Fatalf("%v", err)
 	}
-	return cs, nil
+	return cs
+}
+
+// GetKubernetesClientSetOrDieWithConfig creates a *kubernetes.ClientSet with given config,
+// for talking to a Kubernetes apiserver.
+func GetKubernetesClientSetOrDieWithConfig(config *rest.Config) *kubernetes.Clientset {
+	cs, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Fatalf("%v", err)
+	}
+	return cs
 }
 
 // GetKubernetesInformers creates a informers.SharedInformerFactory for talking to a Kubernetes apiserver.

pkg/controller/controller.go

@@ -85,6 +85,11 @@ func New(name string, mrg manager.Manager, options Options) (Controller, error)
 		Name: name,
 	}
 
+	// Inject dependencies into the controller.
+	if err := mrg.SetFields(c); err != nil {
+		return nil, err
+	}
+
 	// Add the controller as a Manager components
 	return c, mrg.Add(c)
 }

pkg/internal/controller/controller.go

@@ -26,6 +26,7 @@ import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/handler"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/predicate"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/reconcile"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/recorder"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/runtime/inject"
 	logf "github.com/kubernetes-sigs/controller-runtime/pkg/runtime/log"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/source"
@@ -34,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
 	toolscache "k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 )
 
@@ -87,6 +89,10 @@ type Controller struct {
 	// Started is true if the Controller has been Started
 	Started bool
 
+	// recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	recorder record.EventRecorder
+
 	// TODO(community): Consider initializing a logger with the Controller Name as the tag
 }
 
@@ -234,3 +240,21 @@ func (c *Controller) InjectFunc(f inject.Func) error {
 	c.SetFields = f
 	return nil
 }
+
+var _ inject.Recorder = &Controller{}
+
+// InjectRecorder is internal should be called only by the Controller.
+// InjectRecorder is used to inject the event recorder.
+func (c *Controller) InjectRecorder(provider recorder.Provider) error {
+	if c.recorder != nil {
+		return nil
+	}
+
+	eventRecorder, err := provider.GetEventRecorderFor(c.Name)
+	if err != nil {
+		return err
+	}
+	c.recorder = eventRecorder
+
+	return nil
+}

pkg/manager/internal.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/kubernetes-sigs/controller-runtime/pkg/cache"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/recorder"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/runtime/inject"
 	logf "github.com/kubernetes-sigs/controller-runtime/pkg/runtime/log"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -50,6 +51,10 @@ type controllerManager struct {
 	// which can later be consumed via field selectors from the injected client.
 	fieldIndexes client.FieldIndexer
 
+	// recorderProvider is used to generate event recorders that will be injected into Controllers
+	// (and EventHandlers, Sources and Predicates).
+	recorderProvider recorder.Provider
+
 	mu      sync.Mutex
 	started bool
 	errChan chan error
@@ -93,6 +98,9 @@ func (cm *controllerManager) SetFields(i interface{}) error {
 	if _, err := inject.CacheInto(cm.cache, i); err != nil {
 		return err
 	}
+	if _, err := inject.RecorderInto(cm.recorderProvider, i); err != nil {
+		return err
+	}
 	if _, err := inject.InjectorInto(cm.SetFields, i); err != nil {
 		return err
 	}

pkg/manager/manager.go

@@ -22,6 +22,8 @@ import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/cache"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client/apiutil"
+	clientconfig "github.com/kubernetes-sigs/controller-runtime/pkg/client/config"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/recorder"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -131,5 +133,9 @@ func New(config *rest.Config, options Options) (Manager, error) {
 
 	cm.fieldIndexes = cm.cache
 	cm.client = client.DelegatingClient{ReadInterface: cm.cache, WriteInterface: writeObj}
+
+	cm.recorderProvider = recorder.NewProvider()
+	cm.recorderProvider.SetScheme(cm.scheme)
+	cm.recorderProvider.SetClientSet(clientconfig.GetKubernetesClientSetOrDieWithConfig(config))
 	return cm, nil
 }

pkg/recorder/recorder.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2018 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 recorder
+
+import (
+	"fmt"
+
+	"github.com/golang/glog"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
+)
+
+// Provider knows how to generate new event recorders with given name.
+type Provider interface {
+	// NewRecorder returns an EventRecorder with given name.
+	GetEventRecorderFor(name string) (record.EventRecorder, error)
+	// SetClientSet specifies the Clientset of the provider.
+	SetClientSet(cs *kubernetes.Clientset)
+	// SetScheme specifies the Scheme of the provider.
+	SetScheme(scheme *runtime.Scheme)
+}
+
+type provider struct {
+	clientSet        *kubernetes.Clientset
+	scheme           *runtime.Scheme
+	eventBroadcaster record.EventBroadcaster
+}
+
+// NewProvider create a new Provider instance.
+func NewProvider() Provider {
+	return &provider{}
+}
+
+func (p *provider) SetClientSet(cs *kubernetes.Clientset) {
+	if p.clientSet == nil {
+		p.clientSet = cs
+	}
+}
+
+func (p *provider) SetScheme(scheme *runtime.Scheme) {
+	if p.scheme == nil {
+		p.scheme = scheme
+	}
+}
+
+func (p *provider) GetEventRecorderFor(name string) (record.EventRecorder, error) {
+	if p.scheme == nil {
+		return nil, fmt.Errorf("must call SetScheme to specify Scheme before getting recorder")
+	}
+
+	if p.eventBroadcaster == nil {
+		if p.clientSet == nil {
+			return nil, fmt.Errorf("must call SetClientSet to specify ClientSet before getting recorder")
+		}
+		p.eventBroadcaster = record.NewBroadcaster()
+		p.eventBroadcaster.StartLogging(glog.Infof)
+		p.eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: p.clientSet.CoreV1().Events("")})
+	}
+
+	return p.eventBroadcaster.NewRecorder(p.scheme, corev1.EventSource{Component: name}), nil
+}

pkg/runtime/inject/inject.go

@@ -19,6 +19,7 @@ package inject
 import (
 	"github.com/kubernetes-sigs/controller-runtime/pkg/cache"
 	"github.com/kubernetes-sigs/controller-runtime/pkg/client"
+	"github.com/kubernetes-sigs/controller-runtime/pkg/recorder"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
 )
@@ -98,6 +99,21 @@ func StopChannelInto(stop <-chan struct{}, i interface{}) (bool, error) {
 	return false, nil
 }
 
+// Recorder is used by the ControllerManager to inject recorder into Sources, EventHandlers, Predicates, and
+// Reconciles
+type Recorder interface {
+	InjectRecorder(provider recorder.Provider) error
+}
+
+// RecorderInto will set recorder and return the result on i if it implements Recorder.  Returns
+// false if i does not implement Recorder.
+func RecorderInto(provider recorder.Provider, i interface{}) (bool, error) {
+	if is, ok := i.(Recorder); ok {
+		return true, is.InjectRecorder(provider)
+	}
+	return false, nil
+}
+
 // Func injects dependencies into i.
 type Func func(i interface{}) error