Merge pull request #1014 from vincepri/prepare057

Prepare for v0.5.7

Author: k8s-ci-robot

go.mod

@@ -31,3 +31,5 @@ require (
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
 	sigs.k8s.io/yaml v1.1.0
 )
+
+replace github.com/evanphx/json-patch => github.com/evanphx/json-patch v0.0.0-20190815234213-e83c0a1c26c8

go.sum

@@ -53,7 +53,7 @@ func GVKForObject(obj runtime.Object, scheme *runtime.Scheme) (schema.GroupVersi
 		return schema.GroupVersionKind{}, err
 	}
 	if isUnversioned {
-		return schema.GroupVersionKind{}, fmt.Errorf("cannot create a new informer for the unversioned type %T", obj)
+		return schema.GroupVersionKind{}, fmt.Errorf("cannot create group-version-kind for unversioned type %T", obj)
 	}
 
 	if len(gvks) < 1 {

pkg/client/apiutil/apimachinery.go

@@ -22,6 +22,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
+	"sigs.k8s.io/controller-runtime/pkg/internal/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/ratelimiter"
@@ -99,6 +100,7 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		},
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		Name:                    name,
+		Log:                     log.RuntimeLog.WithName("controller").WithValues("controller", name),
 	}
 
 	// Add the controller as a Manager components

pkg/controller/controller.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -29,15 +30,12 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
-	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
-var log = logf.RuntimeLog.WithName("controller")
-
 var _ inject.Injector = &Controller{}
 
 // Controller implements controller.Controller
@@ -88,6 +86,9 @@ type Controller struct {
 
 	// watches maintains a list of sources, handlers, and predicates to start when the controller is started.
 	watches []watchDescription
+
+	// Log is used to log messages to users during reconciliation, or for example when a watch is started.
+	Log logr.Logger
 }
 
 // watchDescription contains all the information necessary to start a watch.
@@ -122,7 +123,7 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc
 
 	c.watches = append(c.watches, watchDescription{src: src, handler: evthdler, predicates: prct})
 	if c.Started {
-		log.Info("Starting EventSource", "controller", c.Name, "source", src)
+		c.Log.Info("Starting EventSource", "source", src)
 		return src.Start(evthdler, c.Queue, prct...)
 	}
 
@@ -148,14 +149,14 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		// caches to sync so that they have a chance to register their intendeded
 		// caches.
 		for _, watch := range c.watches {
-			log.Info("Starting EventSource", "controller", c.Name, "source", watch.src)
+			c.Log.Info("Starting EventSource", "source", watch.src)
 			if err := watch.src.Start(watch.handler, c.Queue, watch.predicates...); err != nil {
 				return err
 			}
 		}
 
 		// Start the SharedIndexInformer factories to begin populating the SharedIndexInformer caches
-		log.Info("Starting Controller", "controller", c.Name)
+		c.Log.Info("Starting Controller")
 
 		for _, watch := range c.watches {
 			syncingSource, ok := watch.src.(source.SyncingSource)
@@ -166,7 +167,7 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 				// This code is unreachable in case of kube watches since WaitForCacheSync will never return an error
 				// Leaving it here because that could happen in the future
 				err := fmt.Errorf("failed to wait for %s caches to sync: %w", c.Name, err)
-				log.Error(err, "Could not wait for Cache to sync", "controller", c.Name)
+				c.Log.Error(err, "Could not wait for Cache to sync")
 				return err
 			}
 		}
@@ -176,7 +177,7 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		}
 
 		// Launch workers to process resources
-		log.Info("Starting workers", "controller", c.Name, "worker count", c.MaxConcurrentReconciles)
+		c.Log.Info("Starting workers", "worker count", c.MaxConcurrentReconciles)
 		for i := 0; i < c.MaxConcurrentReconciles; i++ {
 			// Process work items
 			go wait.Until(c.worker, c.JitterPeriod, stop)
@@ -190,7 +191,7 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 	}
 
 	<-stop
-	log.Info("Stopping workers", "controller", c.Name)
+	c.Log.Info("Stopping workers")
 	return nil
 }
 
@@ -228,23 +229,23 @@ func (c *Controller) reconcileHandler(obj interface{}) bool {
 		c.updateMetrics(time.Since(reconcileStartTS))
 	}()
 
-	var req reconcile.Request
-	var ok bool
-	if req, ok = obj.(reconcile.Request); !ok {
+	// Make sure that the the object is a valid request.
+	req, ok := obj.(reconcile.Request)
+	if !ok {
 		// As the item in the workqueue is actually invalid, we call
 		// Forget here else we'd go into a loop of attempting to
 		// process a work item that is invalid.
 		c.Queue.Forget(obj)
-		log.Error(nil, "Queue item was not a Request",
-			"controller", c.Name, "type", fmt.Sprintf("%T", obj), "value", obj)
+		c.Log.Error(nil, "Queue item was not a Request", "type", fmt.Sprintf("%T", obj), "value", obj)
 		// Return true, don't take a break
 		return true
 	}
+
 	// RunInformersAndControllers the syncHandler, passing it the namespace/Name string of the
 	// resource to be synced.
 	if result, err := c.Do.Reconcile(req); err != nil {
 		c.Queue.AddRateLimited(req)
-		log.Error(err, "Reconciler error", "controller", c.Name, "request", req)
+		c.Log.Error(err, "Reconciler error", "name", req.Name, "namespace", req.Namespace)
 		ctrlmetrics.ReconcileErrors.WithLabelValues(c.Name).Inc()
 		ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, "error").Inc()
 		return false
@@ -268,7 +269,7 @@ func (c *Controller) reconcileHandler(obj interface{}) bool {
 	c.Queue.Forget(obj)
 
 	// TODO(directxman12): What does 1 mean?  Do we want level constants?  Do we want levels at all?
-	log.V(1).Info("Successfully Reconciled", "controller", c.Name, "request", req)
+	c.Log.V(1).Info("Successfully Reconciled", "name", req.Name, "namespace", req.Namespace)
 
 	ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, "success").Inc()
 	// Return true, don't take a break

pkg/internal/controller/controller.go

@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
+	"sigs.k8s.io/controller-runtime/pkg/internal/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -65,6 +66,7 @@ var _ = Describe("controller", func() {
 			MaxConcurrentReconciles: 1,
 			Do:                      fakeReconcile,
 			MakeQueue:               func() workqueue.RateLimitingInterface { return queue },
+			Log:                     log.RuntimeLog.WithName("controller").WithName("test"),
 		}
 		Expect(ctrl.InjectFunc(func(interface{}) error { return nil })).To(Succeed())
 	})

pkg/internal/controller/controller_test.go

@@ -5,6 +5,7 @@ import (
 	"net/url"
 
 	"k8s.io/apimachinery/pkg/runtime/serializer"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 
@@ -42,17 +43,20 @@ func (f *ControlPlane) Start() error {
 
 // Stop will stop your control plane processes, and clean up their data.
 func (f *ControlPlane) Stop() error {
+	var errList []error
+
 	if f.APIServer != nil {
 		if err := f.APIServer.Stop(); err != nil {
-			return err
+			errList = append(errList, err)
 		}
 	}
 	if f.Etcd != nil {
 		if err := f.Etcd.Stop(); err != nil {
-			return err
+			errList = append(errList, err)
 		}
 	}
-	return nil
+
+	return utilerrors.NewAggregate(errList)
 }
 
 // APIURL returns the URL you should connect to to talk to your API.

pkg/internal/testing/integration/control_plane.go

@@ -33,7 +33,7 @@ type Cache interface {
 }
 
 // CacheInto will set informers on i and return the result if it implements Cache.  Returns
-//// false if i does not implement Cache.
+// false if i does not implement Cache.
 func CacheInto(c cache.Cache, i interface{}) (bool, error) {
 	if s, ok := i.(Cache); ok {
 		return true, s.InjectCache(c)
@@ -62,7 +62,7 @@ type Config interface {
 }
 
 // ConfigInto will set config on i and return the result if it implements Config.  Returns
-//// false if i does not implement Config.
+// false if i does not implement Config.
 func ConfigInto(config *rest.Config, i interface{}) (bool, error) {
 	if s, ok := i.(Config); ok {
 		return true, s.InjectConfig(config)