Repository: knative/pkg
Pull request #1423 (2020-06-18 08:50:39)

Abort reconcilation on finalizer patch errors.
This aborts reconcilation if finalizers could not be patched correctly with an error and thus it retries.

We shouldn't start a reconcilation if we haven't been able to correctly add the finalizer first. Otherwise we could get into a weird spot where the resources are created before the finalizer and in a very degenerate case the finalizer wouldn't even be executed. Also, the current code has a bug where if the patch fails the resource coming back from the patch (essentially an empty object) is passed into ReconcileKind, causing headaches.

The same imo goes for removing a finalizer. Failing to remove a finalizer blocks the resource forever, thus we should retry if that happens too.

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1423#" title="Author self-approved">markusthoemmes</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mattmoor**
You can assign the PR to them by writing `/assign @mattmoor` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[codegen/OWNERS](https://github.com/knative/pkg/blob/master/codegen/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["mattmoor"]} -->
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1423#" title="Author self-approved">markusthoemmes</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson**
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["evankanderson"]} -->
markusthoemmes: /assign @n3wscott 
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1423#" title="Author self-approved">markusthoemmes</a>*, *<a href="https://github.com/knative/pkg/pull/1423#pullrequestreview-433357020" title="Approved">n3wscott</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson**
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["evankanderson"]} -->

Patch:
diff --git a/client/injection/apiextensions/reconciler/apiextensions/v1beta1/customresourcedefinition/reconciler.go b/client/injection/apiextensions/reconciler/apiextensions/v1beta1/customresourcedefinition/reconciler.go
index ae55b3e32..30d0fa9cb 100644
--- a/client/injection/apiextensions/reconciler/apiextensions/v1beta1/customresourcedefinition/reconciler.go
+++ b/client/injection/apiextensions/reconciler/apiextensions/v1beta1/customresourcedefinition/reconciler.go
@@ -21,6 +21,7 @@ package customresourcedefinition
 import (
 	context "context"
 	json "encoding/json"
+	fmt "fmt"
 	reflect "reflect"
 
 	zap "go.uber.org/zap"
@@ -161,7 +162,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 		// Set and update the finalizer on resource if r.reconciler
 		// implements Finalizer.
 		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
-			logger.Warnw("Failed to set finalizers", zap.Error(err))
+			return fmt.Errorf("failed to set finalizers: %w", err)
 		}
 
 		// Reconcile this copy of the resource and then write back any status
@@ -176,7 +177,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 		// and reconciled cleanly (nil or normal event), remove the finalizer.
 		reconcileEvent = fin.FinalizeKind(ctx, resource)
 		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
-			logger.Warnw("Failed to clear finalizers", zap.Error(err))
+			return fmt.Errorf("failed to clear finalizers: %w", err)
 		}
 	}
 
diff --git a/client/injection/kube/reconciler/core/v1/namespace/reconciler.go b/client/injection/kube/reconciler/core/v1/namespace/reconciler.go
index 4b7c037e3..00c761a1d 100644
--- a/client/injection/kube/reconciler/core/v1/namespace/reconciler.go
+++ b/client/injection/kube/reconciler/core/v1/namespace/reconciler.go
@@ -21,6 +21,7 @@ package namespace
 import (
 	context "context"
 	json "encoding/json"
+	fmt "fmt"
 	reflect "reflect"
 
 	zap "go.uber.org/zap"
@@ -160,7 +161,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 		// Set and update the finalizer on resource if r.reconciler
 		// implements Finalizer.
 		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
-			logger.Warnw("Failed to set finalizers", zap.Error(err))
+			return fmt.Errorf("failed to set finalizers: %w", err)
 		}
 
 		// Reconcile this copy of the resource and then write back any status
@@ -175,7 +176,7 @@ func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
 		// and reconciled cleanly (nil or normal event), remove the finalizer.
 		reconcileEvent = fin.FinalizeKind(ctx, resource)
 		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
-			logger.Warnw("Failed to clear finalizers", zap.Error(err))
+			return fmt.Errorf("failed to clear finalizers: %w", err)
 		}
 	}
 
diff --git a/codegen/cmd/injection-gen/generators/reconciler_reconciler.go b/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
index 647b82fd0..df4d595ab 100644
--- a/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
+++ b/codegen/cmd/injection-gen/generators/reconciler_reconciler.go
@@ -147,6 +147,7 @@ func (g *reconcilerReconcilerGenerator) GenerateType(c *generator.Context, t *ty
 			Package: "context",
 			Name:    "Context",
 		}),
+		"fmtErrorf":           c.Universe.Package("fmt").Function("Errorf"),
 		"reflectDeepEqual":    c.Universe.Package("reflect").Function("DeepEqual"),
 		"equalitySemantic":    c.Universe.Package("k8s.io/apimachinery/pkg/api/equality").Variable("Semantic"),
 		"jsonMarshal":         c.Universe.Package("encoding/json").Function("Marshal"),
@@ -308,7 +309,7 @@ func (r *reconcilerImpl) Reconcile(ctx {{.contextContext|raw}}, key string) erro
 		// Set and update the finalizer on resource if r.reconciler
 		// implements Finalizer.
 		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
-			logger.Warnw("Failed to set finalizers", zap.Error(err))
+			return {{.fmtErrorf|raw}}("failed to set finalizers: %w", err)
 		}
 
 		{{if .isKRShaped}}
@@ -320,7 +321,7 @@ func (r *reconcilerImpl) Reconcile(ctx {{.contextContext|raw}}, key string) erro
 		reconcileEvent = r.reconciler.ReconcileKind(ctx, resource)
 
 		{{if .isKRShaped}}
-		reconciler.PostProcessReconcile(ctx, resource)
+		reconciler.PostProcessReconcile(ctx, resource, original)
 		{{end}}
 	} else if fin, ok := r.reconciler.(Finalizer); ok {
 		// Append the target method to the logger.
@@ -330,7 +331,7 @@ func (r *reconcilerImpl) Reconcile(ctx {{.contextContext|raw}}, key string) erro
 		// and reconciled cleanly (nil or normal event), remove the finalizer.
 		reconcileEvent = fin.FinalizeKind(ctx, resource)
 		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
-			logger.Warnw("Failed to clear finalizers", zap.Error(err))
+			return {{.fmtErrorf|raw}}("failed to clear finalizers: %w", err)
 		}
 	}
 
diff --git a/configmap/parse.go b/configmap/parse.go
index 210d2114e..1ef675818 100644
--- a/configmap/parse.go
+++ b/configmap/parse.go
@@ -77,6 +77,20 @@ func AsInt64(key string, target *int64) ParseFunc {
 	}
 }
 
+// AsUint32 parses the value at key as an uint32 into the target, if it exists.
+func AsUint32(key string, target *uint32) ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			val, err := strconv.ParseUint(raw, 10, 32)
+			if err != nil {
+				return fmt.Errorf("failed to parse %q: %w", key, err)
+			}
+			*target = uint32(val)
+		}
+		return nil
+	}
+}
+
 // AsFloat64 parses the value at key as a float64 into the target, if it exists.
 func AsFloat64(key string, target *float64) ParseFunc {
 	return func(data map[string]string) error {
diff --git a/controller/controller.go b/controller/controller.go
index 3981c5820..6024ec253 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -36,8 +36,10 @@ import (
 	"k8s.io/client-go/util/workqueue"
 
 	"knative.dev/pkg/kmeta"
+	kle "knative.dev/pkg/leaderelection"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/logging/logkey"
+	"knative.dev/pkg/reconciler"
 )
 
 const (
@@ -176,6 +178,10 @@ func FilterWithNameAndNamespace(namespace, name string) func(obj interface{}) bo
 // Impl is our core controller implementation.  It handles queuing and feeding work
 // from the queue to an implementation of Reconciler.
 type Impl struct {
+	// Name is the unique name for this controller workqueue within this process.
+	// This is used for surfacing metrics, and per-controller leader election.
+	Name string
+
 	// Reconciler is the workhorse of this controller, it is fed the keys
 	// from the workqueue to process.  Public for testing.
 	Reconciler Reconciler
@@ -205,7 +211,9 @@ func NewImpl(r Reconciler, logger *zap.SugaredLogger, workQueueName string) *Imp
 }
 
 func NewImplWithStats(r Reconciler, logger *zap.SugaredLogger, workQueueName string, reporter StatsReporter) *Impl {
+	logger = logger.Named(workQueueName)
 	return &Impl{
+		Name:       workQueueName,
 		Reconciler: r,
 		WorkQueue: workqueue.NewNamedRateLimitingQueue(
 			workqueue.DefaultControllerRateLimiter(),
@@ -341,6 +349,14 @@ func (c *Impl) EnqueueKey(key types.NamespacedName) {
 	c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.WorkQueue.Len())
 }
 
+// MaybeEnqueueBucketKey takes a Bucket and namespace/name string and puts it onto the work queue.
+func (c *Impl) MaybeEnqueueBucketKey(bkt reconciler.Bucket, key types.NamespacedName) {
+	if bkt.Has(key) {
+		c.WorkQueue.Add(key)
+		c.logger.Debugf("Adding to queue %s (depth: %d)", safeKey(key), c.WorkQueue.Len())
+	}
+}
+
 // EnqueueKeyAfter takes a namespace/name string and schedules its execution in
 // the work queue after given delay.
 func (c *Impl) EnqueueKeyAfter(key types.NamespacedName, delay time.Duration) {
@@ -349,10 +365,12 @@ func (c *Impl) EnqueueKeyAfter(key types.NamespacedName, delay time.Duration) {
 }
 
 // RunContext starts the controller's worker threads, the number of which is threadiness.
+// If the context has been decorated for LeaderElection, then an elector is built and run.
 // It then blocks until the context is cancelled, at which point it shuts down its
 // internal work queue and waits for workers to finish processing their current
 // work items.
 func (c *Impl) RunContext(ctx context.Context, threadiness int) error {
+	logger := c.logger
 	defer runtime.HandleCrash()
 	sg := sync.WaitGroup{}
 	defer sg.Wait()
@@ -363,8 +381,20 @@ func (c *Impl) RunContext(ctx context.Context, threadiness int) error {
 		}
 	}()
 
+	if la, ok := c.Reconciler.(reconciler.LeaderAware); ok {
+		// Build and execute an elector.
+		le, err := kle.BuildElector(ctx, la, c.Name, c.MaybeEnqueueBucketKey)
+		if err != nil {
+			return err
+		}
+		sg.Add(1)
+		go func() {
+			defer sg.Done()
+			le.Run(ctx)
+		}()
+	}
+
 	// Launch workers to process resources that get enqueued to our workqueue.
-	logger := c.logger
 	logger.Info("Starting controller and workers")
 	for i := 0; i < threadiness; i++ {
 		sg.Add(1)
diff --git a/go.mod b/go.mod
index e23d408b6..601674ec9 100644
--- a/go.mod
+++ b/go.mod
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/common v0.9.1
 	github.com/spf13/pflag v1.0.5
 	github.com/tsenart/vegeta v12.7.1-0.20190725001342-b5f4fca92137+incompatible
-	go.opencensus.io v0.22.3
+	go.opencensus.io v0.22.4
 	go.uber.org/multierr v1.5.0
 	go.uber.org/zap v1.14.1
 	golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e
@@ -47,7 +47,7 @@ require (
 	k8s.io/code-generator v0.18.0
 	k8s.io/gengo v0.0.0-20200205140755-e0e292d8aa12
 	k8s.io/klog v1.0.0
-	knative.dev/test-infra v0.0.0-20200615231324-3a016f44102c
+	knative.dev/test-infra v0.0.0-20200617235125-6382dba95484
 	sigs.k8s.io/boskos v0.0.0-20200530174753-71e795271860
 )
 
diff --git a/go.sum b/go.sum
index 7e85df170..1d1a0d0ea 100644
--- a/go.sum
+++ b/go.sum
@@ -898,6 +898,8 @@ go.opencensus.io v0.22.2 h1:75k/FF0Q2YM8QYo07VPddOLBslDt1MZOdEslOHvmzAs=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3 h1:8sGtKOrtQqkN1bp2AtX+misvLIlOmsEsNd+9NIcPEm8=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+go.opencensus.io v0.22.4 h1:LYy1Hy3MJdrCdMwwzxA/dRok4ejH+RwNGbuoD9fCjto=
+go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
@@ -1401,8 +1403,8 @@ knative.dev/test-infra v0.0.0-20200522180958-6a0a9b9d893a h1:c0qTABRcNoxZVu5gsry
 knative.dev/test-infra v0.0.0-20200522180958-6a0a9b9d893a/go.mod h1:n9eQkzmSNj8BiqNFl1lzoz68D09uMeJfyOjc132Gbik=
 knative.dev/test-infra v0.0.0-20200606045118-14ebc4a42974 h1:CrZmlbB+j3ZF/aTrfyypY5ulX2w7XrkfeXKQsbkqzTg=
 knative.dev/test-infra v0.0.0-20200606045118-14ebc4a42974/go.mod h1://I6IZIF0QDgs5wotU243ZZ5cTpm6/GthayjUenBBc0=
-knative.dev/test-infra v0.0.0-20200615231324-3a016f44102c h1:pzn7d3gVWX6p10CpdSFAYlgFhLwI6hGQ8H4sxQfvob4=
-knative.dev/test-infra v0.0.0-20200615231324-3a016f44102c/go.mod h1:+BfrTJpc++rH30gX/C0QY6NT2eYVzycll52uw6CrQnc=
+knative.dev/test-infra v0.0.0-20200617235125-6382dba95484 h1:5D1Fm6aA1T1QQXLb1HkJ5t8gB9pTkhLYak1CCqIP+pE=
+knative.dev/test-infra v0.0.0-20200617235125-6382dba95484/go.mod h1:+BfrTJpc++rH30gX/C0QY6NT2eYVzycll52uw6CrQnc=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/injection/README.md b/injection/README.md
index d30b090bc..e23ff32f5 100644
--- a/injection/README.md
+++ b/injection/README.md
@@ -471,7 +471,7 @@ reconciler.PreProcessReconcile(ctx, resource)
 
 reconcileEvent = r.reconciler.ReconcileKind(ctx, resource)
 
-reconciler.PostProcessReconcile(ctx, resource)
+reconciler.PostProcessReconcile(ctx, resource, oldResource)
 ```
 
 #### Stubs
diff --git a/injection/sharedmain/main.go b/injection/sharedmain/main.go
index 1410138e1..5c2ba94a3 100644
--- a/injection/sharedmain/main.go
+++ b/injection/sharedmain/main.go
@@ -52,6 +52,7 @@ import (
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/profiling"
+	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/system"
 	"knative.dev/pkg/version"
@@ -399,6 +400,11 @@ func SecretFetcher(ctx context.Context) metrics.SecretFetcher {
 func ControllersAndWebhooksFromCtors(ctx context.Context,
 	cmw *configmap.InformedWatcher,
 	ctors ...injection.ControllerConstructor) ([]*controller.Impl, []interface{}) {
+
+	// Check whether the context has been infused with a leader elector builder.
+	// If it has, then every reconciler we plan to start MUST implement LeaderAware.
+	leEnabled := kle.HasLeaderElection(ctx)
+
 	controllers := make([]*controller.Impl, 0, len(ctors))
 	webhooks := make([]interface{}, 0)
 	for _, cf := range ctors {
@@ -410,6 +416,12 @@ func ControllersAndWebhooksFromCtors(ctx context.Context,
 		case webhook.AdmissionController, webhook.ConversionController:
 			webhooks = append(webhooks, c)
 		}
+
+		if leEnabled {
+			if _, ok := ctrl.Reconciler.(reconciler.LeaderAware); !ok {
+				log.Fatalf("%T is not leader-aware, all reconcilers must be leader-aware to enable fine-grained leader election.", ctrl.Reconciler)
+			}
+		}
 	}
 
 	return controllers, webhooks
diff --git a/leaderelection/config.go b/leaderelection/config.go
index 06f0874e2..44378d30c 100644
--- a/leaderelection/config.go
+++ b/leaderelection/config.go
@@ -30,6 +30,10 @@ import (
 
 const configMapNameEnv = "CONFIG_LEADERELECTION_NAME"
 
+// MaxBuckets is the maximum number of buckets to allow users to define.
+// This is a variable so that it may be customized in the binary entrypoint.
+var MaxBuckets uint32 = 10
+
 var validResourceLocks = sets.NewString("leases", "configmaps", "endpoints")
 
 // NewConfigFromMap returns a Config for the given map, or an error.
@@ -43,6 +47,8 @@ func NewConfigFromMap(data map[string]string) (*Config, error) {
 		cm.AsDuration("renewDeadline", &config.RenewDeadline),
 		cm.AsDuration("retryPeriod", &config.RetryPeriod),
 
+		cm.AsUint32("buckets", &config.Buckets),
+
 		// enabledComponents are not validated here, because they are dependent on
 		// the component. Components should provide additional validation for this
 		// field.
@@ -51,6 +57,9 @@ func NewConfigFromMap(data map[string]string) (*Config, error) {
 		return nil, err
 	}
 
+	if config.Buckets < 1 || config.Buckets > MaxBuckets {
+		return nil, fmt.Errorf("buckets: value must be between %d <= %d <= %d", 1, config.Buckets, MaxBuckets)
+	}
 	if !validResourceLocks.Has(config.ResourceLock) {
 		return nil, fmt.Errorf(`resourceLock: invalid value %q: valid values are "leases","configmaps","endpoints"`, config.ResourceLock)
 	}
@@ -72,6 +81,7 @@ func NewConfigFromConfigMap(configMap *corev1.ConfigMap) (*Config, error) {
 // single source repository, viz: serving or eventing.
 type Config struct {
 	ResourceLock      string
+	Buckets           uint32
 	LeaseDuration     time.Duration
 	RenewDeadline     time.Duration
 	RetryPeriod       time.Duration
@@ -83,6 +93,7 @@ func (c *Config) GetComponentConfig(name string) ComponentConfig {
 		return ComponentConfig{
 			Component:     name,
 			LeaderElect:   true,
+			Buckets:       c.Buckets,
 			ResourceLock:  c.ResourceLock,
 			LeaseDuration: c.LeaseDuration,
 			RenewDeadline: c.RenewDeadline,
@@ -96,6 +107,7 @@ func (c *Config) GetComponentConfig(name string) ComponentConfig {
 func defaultConfig() *Config {
 	return &Config{
 		ResourceLock:      "leases",
+		Buckets:           1,
 		LeaseDuration:     15 * time.Second,
 		RenewDeadline:     10 * time.Second,
 		RetryPeriod:       2 * time.Second,
@@ -107,6 +119,7 @@ func defaultConfig() *Config {
 type ComponentConfig struct {
 	Component     string
 	LeaderElect   bool
+	Buckets       uint32
 	ResourceLock  string
 	LeaseDuration time.Duration
 	RenewDeadline time.Duration
diff --git a/leaderelection/context.go b/leaderelection/context.go
new file mode 100644
index 000000000..a0d8555f9
--- /dev/null
+++ b/leaderelection/context.go
@@ -0,0 +1,223 @@
+/*
+Copyright 2020 The Knative 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 leaderelection
+
+import (
+	"context"
+	"fmt"
+	"hash/fnv"
+	"strings"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/system"
+)
+
+// WithStandardLeaderElectorBuilder infuses a context with the ability to build
+// LeaderElectors with the provided component configuration acquiring resource
+// locks via the provided kubernetes client.
+func WithStandardLeaderElectorBuilder(ctx context.Context, kc kubernetes.Interface, cc ComponentConfig) context.Context {
+	return context.WithValue(ctx, builderKey{}, &standardBuilder{
+		kc:  kc,
+		lec: cc,
+	})
+}
+
+// HasLeaderElection returns whether there is leader election configuration
+// associated with the context
+func HasLeaderElection(ctx context.Context) bool {
+	val := ctx.Value(builderKey{})
+	return val != nil
+}
+
+// Elector is the interface for running a leader elector.
+type Elector interface {
+	Run(context.Context)
+}
+
+// BuildElector builds a leaderelection.LeaderElector for the named LeaderAware
+// reconciler using a builder added to the context via WithStandardLeaderElectorBuilder.
+func BuildElector(ctx context.Context, la reconciler.LeaderAware, name string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
+	if val := ctx.Value(builderKey{}); val != nil {
+		switch builder := val.(type) {
+		case *standardBuilder:
+			return builder.BuildElector(ctx, la, name, enq)
+		}
+		// TODO(mattmoor): Add a flavor of builder that relies on StatefulSet to partition the key space.
+	}
+
+	return &unopposedElector{
+		la:  la,
+		bkt: reconciler.UniversalBucket(),
+		enq: enq,
+	}, nil
+}
+
+type builderKey struct{}
+
+type standardBuilder struct {
+	kc  kubernetes.Interface
+	lec ComponentConfig
+}
+
+func (b *standardBuilder) BuildElector(ctx context.Context, la reconciler.LeaderAware, name string, enq func(reconciler.Bucket, types.NamespacedName)) (Elector, error) {
+	logger := logging.FromContext(ctx)
+
+	id, err := UniqueID()
+	if err != nil {
+		return nil, err
+	}
+
+	buckets := make([]Elector, 0, b.lec.Buckets)
+	for i := uint32(0); i < b.lec.Buckets; i++ {
+		bkt := &bucket{
+			component: b.lec.Component,
+			name:      name,
+			index:     i,
+			total:     b.lec.Buckets,
+		}
+
+		rl, err := resourcelock.New(b.lec.ResourceLock,
+			system.Namespace(), // use namespace we are running in
+			bkt.Name(),
+			b.kc.CoreV1(),
+			b.kc.CoordinationV1(),
+			resourcelock.ResourceLockConfig{
+				Identity: id,
+			})
+		if err != nil {
+			return nil, err
+		}
+		logger.Infof("%s will run in leader-elected mode with id %q", bkt.Name(), rl.Identity())
+
+		le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+			Lock:          rl,
+			LeaseDuration: b.lec.LeaseDuration,
+			RenewDeadline: b.lec.RenewDeadline,
+			RetryPeriod:   b.lec.RetryPeriod,
+			Callbacks: leaderelection.LeaderCallbacks{
+				OnStartedLeading: func(context.Context) {
+					logger.Infof("%q has started leading %q", rl.Identity(), bkt.Name())
+					if err := la.Promote(bkt, enq); err != nil {
+						// TODO(mattmoor): We expect this to effectively never happen,
+						// but if it does, we should support wrapping `le` in an elector
+						// we can cancel here.
+						logger.Fatalf("%q failed to Promote: %v", rl.Identity(), err)
+					}
+				},
+				OnStoppedLeading: func() {
+					logger.Infof("%q has stopped leading %q", rl.Identity(), bkt.Name())
+					la.Demote(bkt)
+				},
+			},
+			ReleaseOnCancel: true,
+
+			Name: rl.Identity(),
+		})
+		if err != nil {
+			return nil, err
+		}
+		// TODO: use health check watchdog, knative/pkg#1048
+		// if lec.WatchDog != nil {
+		// 	lec.WatchDog.SetLeaderElection(le)
+		// }
+		buckets = append(buckets, &runUntilCancelled{Elector: le})
+	}
+	return &runAll{les: buckets}, nil
+}
+
+// unopposedElector promotes when run without needing to be elected.
+type unopposedElector struct {
+	bkt reconciler.Bucket
+	la  reconciler.LeaderAware
+	enq func(reconciler.Bucket, types.NamespacedName)
+}
+
+// Run implements Elector
+func (ue *unopposedElector) Run(ctx context.Context) {
+	ue.la.Promote(ue.bkt, ue.enq)
+}
+
+type runAll struct {
+	les []Elector
+}
+
+// Run implements Elector
+func (ra *runAll) Run(ctx context.Context) {
+	sg := sync.WaitGroup{}
+	defer sg.Wait()
+
+	for _, le := range ra.les {
+		sg.Add(1)
+		go func(le Elector) {
+			defer sg.Done()
+			le.Run(ctx)
+		}(le)
+	}
+}
+
+// runUntilCancelled wraps a single-term Elector into one that runs until
+// the passed context is cancelled.
+type runUntilCancelled struct {
+	// Elector is a single-term elector as we get from K8s leaderelection package.
+	Elector
+}
+
+// Run implements Elector
+func (ruc *runUntilCancelled) Run(ctx context.Context) {
+	// Turn the single-term elector into a continuous election cycle.
+	for {
+		ruc.Elector.Run(ctx)
+		select {
+		case <-ctx.Done():
+			return // Run quit because context was cancelled, we are done!
+		default:
+			// Context wasn't cancelled, start over.
+		}
+	}
+}
+
+type bucket struct {
+	component string
+	name      string
+
+	// We are bucket {index} of {total}
+	index uint32
+	total uint32
+}
+
+var _ reconciler.Bucket = (*bucket)(nil)
+
+// Name implements reconciler.Bucket
+func (b *bucket) Name() string {
+	// The resource name is the lowercase:
+	//   {component}.{workqueue}.{index}-of-{total}
+	return strings.ToLower(fmt.Sprintf("%s.%s.%02d-of-%02d", b.component, b.name, b.index, b.total))
+}
+
+// Has implements reconciler.Bucket
+func (b *bucket) Has(nn types.NamespacedName) bool {
+	h := fnv.New32a()
+	h.Write([]byte(nn.Namespace + "." + nn.Name))
+	ii := h.Sum32() % b.total
+	return b.index == ii
+}
diff --git a/reconciler/leader.go b/reconciler/leader.go
new file mode 100644
index 000000000..3d5f68fdc
--- /dev/null
+++ b/reconciler/leader.go
@@ -0,0 +1,118 @@
+/*
+Copyright 2020 The Knative 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 reconciler
+
+import (
+	"sync"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// Bucket is an opaque type used to scope leadership.
+type Bucket interface {
+	// Name returns a string representing this bucket, which uniquely
+	// identifies the bucket and is suitable for use as a resource lock name.
+	Name() string
+
+	// Has determines whether this Bucket contains a particular key.
+	Has(key types.NamespacedName) bool
+}
+
+// UniversalBucket returns a Bucket that "Has()" all keys.
+func UniversalBucket() Bucket {
+	return &bucket{}
+}
+
+type bucket struct{}
+
+var _ Bucket = (*bucket)(nil)
+
+// Name implements Bucket
+func (b *bucket) Name() string {
+	return ""
+}
+
+// Has implements Bucket
+func (b *bucket) Has(nn types.NamespacedName) bool {
+	return true
+}
+
+// LeaderAware is implemented by Reconcilers that are aware of their leader status.
+type LeaderAware interface {
+	// Promote is called when we become the leader of a given Bucket.  It must be
+	// supplied with an enqueue function through which a Bucket resync may be triggered.
+	Promote(b Bucket, enq func(Bucket, types.NamespacedName)) error
+
+	// Demote is called when we stop being the leader for the specified Bucket.
+	Demote(Bucket)
+}
+
+// LeaderAwareFuncs implements LeaderAware using the given functions for handling
+// promotion and demotion.
+type LeaderAwareFuncs struct {
+	sync.RWMutex
+	buckets map[string]Bucket
+
+	PromoteFunc func(b Bucket, enq func(Bucket, types.NamespacedName)) error
+	DemoteFunc  func(b Bucket)
+}
+
+var _ LeaderAware = (*LeaderAwareFuncs)(nil)
+
+// IsLeaderFor implements LeaderAware
+func (laf *LeaderAwareFuncs) IsLeaderFor(key types.NamespacedName) bool {
+	laf.RLock()
+	defer laf.RUnlock()
+
+	for _, bkt := range laf.buckets {
+		if bkt.Has(key) {
+			return true
+		}
+	}
+	return false
+}
+
+// Promote implements LeaderAware
+func (laf *LeaderAwareFuncs) Promote(b Bucket, enq func(Bucket, types.NamespacedName)) error {
+	func() {
+		laf.Lock()
+		defer laf.Unlock()
+		if laf.buckets == nil {
+			laf.buckets = make(map[string]Bucket, 1)
+		}
+		laf.buckets[b.Name()] = b
+		return
+	}()
+
+	if promote := laf.PromoteFunc; promote != nil {
+		return promote(b, enq)
+	}
+	return nil
+}
+
+// Demote implements LeaderAware
+func (laf *LeaderAwareFuncs) Demote(b Bucket) {
+	func() {
+		laf.Lock()
+		defer laf.Unlock()
+		delete(laf.buckets, b.Name())
+	}()
+
+	if demote := laf.DemoteFunc; demote != nil {
+		demote(b)
+	}
+}
diff --git a/reconciler/reconcile_common.go b/reconciler/reconcile_common.go
index c9f8b5cb8..a807edfb6 100644
--- a/reconciler/reconcile_common.go
+++ b/reconciler/reconcile_common.go
@@ -18,6 +18,10 @@ package reconciler
 
 import (
 	"context"
+	"reflect"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -50,7 +54,7 @@ func PreProcessReconcile(ctx context.Context, resource duckv1.KRShaped) {
 }
 
 // PostProcessReconcile contains logic to apply after reconciliation of a resource.
-func PostProcessReconcile(ctx context.Context, resource duckv1.KRShaped) {
+func PostProcessReconcile(ctx context.Context, resource, oldResource duckv1.KRShaped) {
 	logger := logging.FromContext(ctx)
 	status := resource.GetStatus()
 	mgr := resource.GetConditionSet().Manage(status)
@@ -64,4 +68,26 @@ func PostProcessReconcile(ctx context.Context, resource duckv1.KRShaped) {
 	} else if rc.Reason == failedGenerationBump {
 		logger.Warn("A reconciler observed a new generation without updating the resource status")
 	}
+
+	groomConditionsTransitionTime(resource, oldResource)
+}
+
+// groomConditionsTransitionTime ensures that the LastTransitionTime only advances for resources
+// where the condition has changed during reconciliation. This also ensures that all advanced
+// conditions share the same timestamp.
+func groomConditionsTransitionTime(resource, oldResource duckv1.KRShaped) {
+	now := apis.VolatileTime{Inner: metav1.NewTime(time.Now())}
+	sts := resource.GetStatus()
+	for i := range sts.Conditions {
+		cond := &sts.Conditions[i]
+
+		if oldCond := oldResource.GetStatus().GetCondition(cond.Type); oldCond != nil {
+			cond.LastTransitionTime = oldCond.LastTransitionTime
+			if reflect.DeepEqual(cond, oldCond) {
+				continue
+			}
+		}
+
+		cond.LastTransitionTime = now
+	}
 }
diff --git a/vendor/go.opencensus.io/Makefile b/vendor/go.opencensus.io/Makefile
index 457866cb1..b3ce3df30 100644
--- a/vendor/go.opencensus.io/Makefile
+++ b/vendor/go.opencensus.io/Makefile
@@ -8,7 +8,7 @@ ALL_PKGS := $(shell go list $(sort $(dir $(ALL_SRC))))
 GOTEST_OPT?=-v -race -timeout 30s
 GOTEST_OPT_WITH_COVERAGE = $(GOTEST_OPT) -coverprofile=coverage.txt -covermode=atomic
 GOTEST=go test
-GOFMT=gofmt
+GOIMPORTS=goimports
 GOLINT=golint
 GOVET=go vet
 EMBEDMD=embedmd
@@ -17,14 +17,14 @@ TRACE_ID_LINT_EXCEPTION="type name will be used as trace.TraceID by other packag
 TRACE_OPTION_LINT_EXCEPTION="type name will be used as trace.TraceOptions by other packages"
 README_FILES := $(shell find . -name '*README.md' | sort | tr '\n' ' ')
 
-.DEFAULT_GOAL := fmt-lint-vet-embedmd-test
+.DEFAULT_GOAL := imports-lint-vet-embedmd-test
 
-.PHONY: fmt-lint-vet-embedmd-test
-fmt-lint-vet-embedmd-test: fmt lint vet embedmd test
+.PHONY: imports-lint-vet-embedmd-test
+imports-lint-vet-embedmd-test: imports lint vet embedmd test
 
 # TODO enable test-with-coverage in tavis
 .PHONY: travis-ci
-travis-ci: fmt lint vet embedmd test test-386
+travis-ci: imports lint vet embedmd test test-386
 
 all-pkgs:
 	@echo $(ALL_PKGS) | tr ' ' '\n' | sort
@@ -44,15 +44,15 @@ test-386:
 test-with-coverage:
 	$(GOTEST) $(GOTEST_OPT_WITH_COVERAGE) $(ALL_PKGS)
 
-.PHONY: fmt
-fmt:
-	@FMTOUT=`$(GOFMT) -s -l $(ALL_SRC) 2>&1`; \
-	if [ "$$FMTOUT" ]; then \
-		echo "$(GOFMT) FAILED => gofmt the following files:\n"; \
-		echo "$$FMTOUT\n"; \
+.PHONY: imports
+imports:
+	@IMPORTSOUT=`$(GOIMPORTS) -l $(ALL_SRC) 2>&1`; \
+	if [ "$$IMPORTSOUT" ]; then \
+		echo "$(GOIMPORTS) FAILED => goimports the following files:\n"; \
+		echo "$$IMPORTSOUT\n"; \
 		exit 1; \
 	else \
-	    echo "Fmt finished successfully"; \
+	    echo "Imports finished successfully"; \
 	fi
 
 .PHONY: lint
@@ -91,6 +91,7 @@ embedmd:
 
 .PHONY: install-tools
 install-tools:
-	go get -u golang.org/x/tools/cmd/cover
 	go get -u golang.org/x/lint/golint
+	go get -u golang.org/x/tools/cmd/cover
+	go get -u golang.org/x/tools/cmd/goimports
 	go get -u github.com/rakyll/embedmd
diff --git a/vendor/go.opencensus.io/plugin/ocgrpc/client.go b/vendor/go.opencensus.io/plugin/ocgrpc/client.go
index 28fddb844..2063b6f76 100644
--- a/vendor/go.opencensus.io/plugin/ocgrpc/client.go
+++ b/vendor/go.opencensus.io/plugin/ocgrpc/client.go
@@ -16,8 +16,8 @@ package ocgrpc
 
 import (
 	"context"
-	"go.opencensus.io/trace"
 
+	"go.opencensus.io/trace"
 	"google.golang.org/grpc/stats"
 )
 
diff --git a/vendor/go.opencensus.io/plugin/ocgrpc/server.go b/vendor/go.opencensus.io/plugin/ocgrpc/server.go
index 15ada839d..8a53e0972 100644
--- a/vendor/go.opencensus.io/plugin/ocgrpc/server.go
+++ b/vendor/go.opencensus.io/plugin/ocgrpc/server.go
@@ -16,9 +16,10 @@ package ocgrpc
 
 import (
 	"context"
-	"go.opencensus.io/trace"
 
 	"google.golang.org/grpc/stats"
+
+	"go.opencensus.io/trace"
 )
 
 // ServerHandler implements gRPC stats.Handler recording OpenCensus stats and
diff --git a/vendor/go.opencensus.io/plugin/ocgrpc/trace_common.go b/vendor/go.opencensus.io/plugin/ocgrpc/trace_common.go
index fef582756..61bc543d0 100644
--- a/vendor/go.opencensus.io/plugin/ocgrpc/trace_common.go
+++ b/vendor/go.opencensus.io/plugin/ocgrpc/trace_common.go
@@ -15,16 +15,16 @@
 package ocgrpc
 
 import (
+	"context"
 	"strings"
 
 	"google.golang.org/grpc/codes"
-
-	"context"
-	"go.opencensus.io/trace"
-	"go.opencensus.io/trace/propagation"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
+
+	"go.opencensus.io/trace"
+	"go.opencensus.io/trace/propagation"
 )
 
 const traceContextKey = "grpc-trace-bin"
diff --git a/vendor/go.opencensus.io/plugin/ochttp/propagation/b3/b3.go b/vendor/go.opencensus.io/plugin/ochttp/propagation/b3/b3.go
index 2f1c7f006..9ad885219 100644
--- a/vendor/go.opencensus.io/plugin/ochttp/propagation/b3/b3.go
+++ b/vendor/go.opencensus.io/plugin/ochttp/propagation/b3/b3.go
@@ -68,7 +68,7 @@ func ParseTraceID(tid string) (trace.TraceID, bool) {
 		return trace.TraceID{}, false
 	}
 	b, err := hex.DecodeString(tid)
-	if err != nil {
+	if err != nil || len(b) > 16 {
 		return trace.TraceID{}, false
 	}
 	var traceID trace.TraceID
@@ -90,7 +90,7 @@ func ParseSpanID(sid string) (spanID trace.SpanID, ok bool) {
 		return trace.SpanID{}, false
 	}
 	b, err := hex.DecodeString(sid)
-	if err != nil {
+	if err != nil || len(b) > 8 {
 		return trace.SpanID{}, false
 	}
 	start := 8 - len(b)
diff --git a/vendor/go.opencensus.io/stats/record.go b/vendor/go.opencensus.io/stats/record.go
index ad4691184..2b9728346 100644
--- a/vendor/go.opencensus.io/stats/record.go
+++ b/vendor/go.opencensus.io/stats/record.go
@@ -31,10 +31,19 @@ func init() {
 	}
 }
 
+// Recorder provides an interface for exporting measurement information from
+// the static Record method by using the WithRecorder option.
+type Recorder interface {
+	// Record records a set of measurements associated with the given tags and attachments.
+	// The second argument is a `[]Measurement`.
+	Record(*tag.Map, interface{}, map[string]interface{})
+}
+
 type recordOptions struct {
 	attachments  metricdata.Attachments
 	mutators     []tag.Mutator
 	measurements []Measurement
+	recorder     Recorder
 }
 
 // WithAttachments applies provided exemplar attachments.
@@ -58,6 +67,14 @@ func WithMeasurements(measurements ...Measurement) Options {
 	}
 }
 
+// WithRecorder records the measurements to the specified `Recorder`, rather
+// than to the global metrics recorder.
+func WithRecorder(meter Recorder) Options {
+	return func(ro *recordOptions) {
+		ro.recorder = meter
+	}
+}
+
 // Options apply changes to recordOptions.
 type Options func(*recordOptions)
 
@@ -93,6 +110,9 @@ func RecordWithOptions(ctx context.Context, ros ...Options) error {
 		return nil
 	}
 	recorder := internal.DefaultRecorder
+	if o.recorder != nil {
+		recorder = o.recorder.Record
+	}
 	if recorder == nil {
 		return nil
 	}
diff --git a/vendor/go.opencensus.io/stats/view/export.go b/vendor/go.opencensus.io/stats/view/export.go
index 7cb59718f..73ba11f5b 100644
--- a/vendor/go.opencensus.io/stats/view/export.go
+++ b/vendor/go.opencensus.io/stats/view/export.go
@@ -14,13 +14,6 @@
 
 package view
 
-import "sync"
-
-var (
-	exportersMu sync.RWMutex // guards exporters
-	exporters   = make(map[Exporter]struct{})
-)
-
 // Exporter exports the collected records as view data.
 //
 // The ExportView method should return quickly; if an
@@ -43,16 +36,10 @@ type Exporter interface {
 //
 // Binaries can register exporters, libraries shouldn't register exporters.
 func RegisterExporter(e Exporter) {
-	exportersMu.Lock()
-	defer exportersMu.Unlock()
-
-	exporters[e] = struct{}{}
+	defaultWorker.RegisterExporter(e)
 }
 
 // UnregisterExporter unregisters an exporter.
 func UnregisterExporter(e Exporter) {
-	exportersMu.Lock()
-	defer exportersMu.Unlock()
-
-	delete(exporters, e)
+	defaultWorker.UnregisterExporter(e)
 }
diff --git a/vendor/go.opencensus.io/stats/view/view_to_metric.go b/vendor/go.opencensus.io/stats/view/view_to_metric.go
index 293c1646d..5e1656a1f 100644
--- a/vendor/go.opencensus.io/stats/view/view_to_metric.go
+++ b/vendor/go.opencensus.io/stats/view/view_to_metric.go
@@ -18,6 +18,8 @@ package view
 import (
 	"time"
 
+	"go.opencensus.io/resource"
+
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/stats"
 )
@@ -125,7 +127,7 @@ func rowToTimeseries(v *viewInternal, row *Row, now time.Time, startTime time.Ti
 	}
 }
 
-func viewToMetric(v *viewInternal, now time.Time, startTime time.Time) *metricdata.Metric {
+func viewToMetric(v *viewInternal, r *resource.Resource, now time.Time, startTime time.Time) *metricdata.Metric {
 	if v.metricDescriptor.Type == metricdata.TypeGaugeInt64 ||
 		v.metricDescriptor.Type == metricdata.TypeGaugeFloat64 {
 		startTime = time.Time{}
@@ -144,6 +146,7 @@ func viewToMetric(v *viewInternal, now time.Time, startTime time.Time) *metricda
 	m := &metricdata.Metric{
 		Descriptor: *v.metricDescriptor,
 		TimeSeries: ts,
+		Resource:   r,
 	}
 	return m
 }
diff --git a/vendor/go.opencensus.io/stats/view/worker.go b/vendor/go.opencensus.io/stats/view/worker.go
index 2f3c018af..ab8bfd46d 100644
--- a/vendor/go.opencensus.io/stats/view/worker.go
+++ b/vendor/go.opencensus.io/stats/view/worker.go
@@ -20,6 +20,8 @@ import (
 	"sync"
 	"time"
 
+	"go.opencensus.io/resource"
+
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/metric/metricproducer"
 	"go.opencensus.io/stats"
@@ -28,7 +30,7 @@ import (
 )
 
 func init() {
-	defaultWorker = newWorker()
+	defaultWorker = NewMeter().(*worker)
 	go defaultWorker.start()
 	internal.DefaultRecorder = record
 }
@@ -47,8 +49,69 @@ type worker struct {
 	c          chan command
 	quit, done chan bool
 	mu         sync.RWMutex
+	r          *resource.Resource
+
+	exportersMu sync.RWMutex
+	exporters   map[Exporter]struct{}
+}
+
+// Meter defines an interface which allows a single process to maintain
+// multiple sets of metrics exports (intended for the advanced case where a
+// single process wants to report metrics about multiple objects, such as
+// multiple databases or HTTP services).
+//
+// Note that this is an advanced use case, and the static functions in this
+// module should cover the common use cases.
+type Meter interface {
+	stats.Recorder
+	// Find returns a registered view associated with this name.
+	// If no registered view is found, nil is returned.
+	Find(name string) *View
+	// Register begins collecting data for the given views.
+	// Once a view is registered, it reports data to the registered exporters.
+	Register(views ...*View) error
+	// Unregister the given views. Data will not longer be exported for these views
+	// after Unregister returns.
+	// It is not necessary to unregister from views you expect to collect for the
+	// duration of your program execution.
+	Unregister(views ...*View)
+	// SetReportingPeriod sets the interval between reporting aggregated views in
+	// the program. If duration is less than or equal to zero, it enables the
+	// default behavior.
+	//
+	// Note: each exporter makes different promises about what the lowest supported
+	// duration is. For example, the Stackdriver exporter recommends a value no
+	// lower than 1 minute. Consult each exporter per your needs.
+	SetReportingPeriod(time.Duration)
+
+	// RegisterExporter registers an exporter.
+	// Collected data will be reported via all the
+	// registered exporters. Once you no longer
+	// want data to be exported, invoke UnregisterExporter
+	// with the previously registered exporter.
+	//
+	// Binaries can register exporters, libraries shouldn't register exporters.
+	RegisterExporter(Exporter)
+	// UnregisterExporter unregisters an exporter.
+	UnregisterExporter(Exporter)
+	// SetResource may be used to set the Resource associated with this registry.
+	// This is intended to be used in cases where a single process exports metrics
+	// for multiple Resources, typically in a multi-tenant situation.
+	SetResource(*resource.Resource)
+
+	// Start causes the Meter to start processing Record calls and aggregating
+	// statistics as well as exporting data.
+	Start()
+	// Stop causes the Meter to stop processing calls and terminate data export.
+	Stop()
+
+	// RetrieveData gets a snapshot of the data collected for the the view registered
+	// with the given name. It is intended for testing only.
+	RetrieveData(viewName string) ([]*Row, error)
 }
 
+var _ Meter = (*worker)(nil)
+
 var defaultWorker *worker
 
 var defaultReportingDuration = 10 * time.Second
@@ -56,11 +119,17 @@ var defaultReportingDuration = 10 * time.Second
 // Find returns a registered view associated with this name.
 // If no registered view is found, nil is returned.
 func Find(name string) (v *View) {
+	return defaultWorker.Find(name)
+}
+
+// Find returns a registered view associated with this name.
+// If no registered view is found, nil is returned.
+func (w *worker) Find(name string) (v *View) {
 	req := &getViewByNameReq{
 		name: name,
 		c:    make(chan *getViewByNameResp),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 	resp := <-req.c
 	return resp.v
 }
@@ -68,11 +137,17 @@ func Find(name string) (v *View) {
 // Register begins collecting data for the given views.
 // Once a view is registered, it reports data to the registered exporters.
 func Register(views ...*View) error {
+	return defaultWorker.Register(views...)
+}
+
+// Register begins collecting data for the given views.
+// Once a view is registered, it reports data to the registered exporters.
+func (w *worker) Register(views ...*View) error {
 	req := &registerViewReq{
 		views: views,
 		err:   make(chan error),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 	return <-req.err
 }
 
@@ -81,6 +156,14 @@ func Register(views ...*View) error {
 // It is not necessary to unregister from views you expect to collect for the
 // duration of your program execution.
 func Unregister(views ...*View) {
+	defaultWorker.Unregister(views...)
+}
+
+// Unregister the given views. Data will not longer be exported for these views
+// after Unregister returns.
+// It is not necessary to unregister from views you expect to collect for the
+// duration of your program execution.
+func (w *worker) Unregister(views ...*View) {
 	names := make([]string, len(views))
 	for i := range views {
 		names[i] = views[i].Name
@@ -89,31 +172,42 @@ func Unregister(views ...*View) {
 		views: names,
 		done:  make(chan struct{}),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 	<-req.done
 }
 
 // RetrieveData gets a snapshot of the data collected for the the view registered
 // with the given name. It is intended for testing only.
 func RetrieveData(viewName string) ([]*Row, error) {
+	return defaultWorker.RetrieveData(viewName)
+}
+
+// RetrieveData gets a snapshot of the data collected for the the view registered
+// with the given name. It is intended for testing only.
+func (w *worker) RetrieveData(viewName string) ([]*Row, error) {
 	req := &retrieveDataReq{
 		now: time.Now(),
 		v:   viewName,
 		c:   make(chan *retrieveDataResp),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 	resp := <-req.c
 	return resp.rows, resp.err
 }
 
 func record(tags *tag.Map, ms interface{}, attachments map[string]interface{}) {
+	defaultWorker.Record(tags, ms, attachments)
+}
+
+// Record records a set of measurements ms associated with the given tags and attachments.
+func (w *worker) Record(tags *tag.Map, ms interface{}, attachments map[string]interface{}) {
 	req := &recordReq{
 		tm:          tags,
 		ms:          ms.([]stats.Measurement),
 		attachments: attachments,
 		t:           time.Now(),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 }
 
 // SetReportingPeriod sets the interval between reporting aggregated views in
@@ -124,17 +218,31 @@ func record(tags *tag.Map, ms interface{}, attachments map[string]interface{}) {
 // duration is. For example, the Stackdriver exporter recommends a value no
 // lower than 1 minute. Consult each exporter per your needs.
 func SetReportingPeriod(d time.Duration) {
+	defaultWorker.SetReportingPeriod(d)
+}
+
+// SetReportingPeriod sets the interval between reporting aggregated views in
+// the program. If duration is less than or equal to zero, it enables the
+// default behavior.
+//
+// Note: each exporter makes different promises about what the lowest supported
+// duration is. For example, the Stackdriver exporter recommends a value no
+// lower than 1 minute. Consult each exporter per your needs.
+func (w *worker) SetReportingPeriod(d time.Duration) {
 	// TODO(acetechnologist): ensure that the duration d is more than a certain
 	// value. e.g. 1s
 	req := &setReportingPeriodReq{
 		d: d,
 		c: make(chan bool),
 	}
-	defaultWorker.c <- req
+	w.c <- req
 	<-req.c // don't return until the timer is set to the new duration.
 }
 
-func newWorker() *worker {
+// NewMeter constructs a Meter instance. You should only need to use this if
+// you need to separate out Measurement recordings and View aggregations within
+// a single process.
+func NewMeter() Meter {
 	return &worker{
 		measures:   make(map[string]*measureRef),
 		views:      make(map[string]*viewInternal),
@@ -143,9 +251,23 @@ func newWorker() *worker {
 		c:          make(chan command, 1024),
 		quit:       make(chan bool),
 		done:       make(chan bool),
+
+		exporters: make(map[Exporter]struct{}),
 	}
 }
 
+// SetResource associates all data collected by this Meter with the specified
+// resource. This resource is reported when using metricexport.ReadAndExport;
+// it is not provided when used with ExportView/RegisterExporter, because that
+// interface does not provide a means for reporting the Resource.
+func (w *worker) SetResource(r *resource.Resource) {
+	w.r = r
+}
+
+func (w *worker) Start() {
+	go w.start()
+}
+
 func (w *worker) start() {
 	prodMgr := metricproducer.GlobalManager()
 	prodMgr.AddProducer(w)
@@ -155,7 +277,7 @@ func (w *worker) start() {
 		case cmd := <-w.c:
 			cmd.handleCommand(w)
 		case <-w.timer.C:
-			w.reportUsage(time.Now())
+			w.reportUsage()
 		case <-w.quit:
 			w.timer.Stop()
 			close(w.c)
@@ -165,7 +287,7 @@ func (w *worker) start() {
 	}
 }
 
-func (w *worker) stop() {
+func (w *worker) Stop() {
 	prodMgr := metricproducer.GlobalManager()
 	prodMgr.DeleteProducer(w)
 
@@ -202,44 +324,45 @@ func (w *worker) tryRegisterView(v *View) (*viewInternal, error) {
 		return x, nil
 	}
 	w.views[vi.view.Name] = vi
+	w.startTimes[vi] = time.Now()
 	ref := w.getMeasureRef(vi.view.Measure.Name())
 	ref.views[vi] = struct{}{}
 	return vi, nil
 }
 
-func (w *worker) unregisterView(viewName string) {
+func (w *worker) unregisterView(v *viewInternal) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	delete(w.views, viewName)
+	delete(w.views, v.view.Name)
+	delete(w.startTimes, v)
+	if measure := w.measures[v.view.Measure.Name()]; measure != nil {
+		delete(measure.views, v)
+	}
 }
 
-func (w *worker) reportView(v *viewInternal, now time.Time) {
+func (w *worker) reportView(v *viewInternal) {
 	if !v.isSubscribed() {
 		return
 	}
 	rows := v.collectedRows()
-	_, ok := w.startTimes[v]
-	if !ok {
-		w.startTimes[v] = now
-	}
 	viewData := &Data{
 		View:  v.view,
 		Start: w.startTimes[v],
 		End:   time.Now(),
 		Rows:  rows,
 	}
-	exportersMu.Lock()
-	for e := range exporters {
+	w.exportersMu.Lock()
+	defer w.exportersMu.Unlock()
+	for e := range w.exporters {
 		e.ExportView(viewData)
 	}
-	exportersMu.Unlock()
 }
 
-func (w *worker) reportUsage(now time.Time) {
+func (w *worker) reportUsage() {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	for _, v := range w.views {
-		w.reportView(v, now)
+		w.reportView(v)
 	}
 }
 
@@ -248,11 +371,6 @@ func (w *worker) toMetric(v *viewInternal, now time.Time) *metricdata.Metric {
 		return nil
 	}
 
-	_, ok := w.startTimes[v]
-	if !ok {
-		w.startTimes[v] = now
-	}
-
 	var startTime time.Time
 	if v.metricDescriptor.Type == metricdata.TypeGaugeInt64 ||
 		v.metricDescriptor.Type == metricdata.TypeGaugeFloat64 {
@@ -261,7 +379,7 @@ func (w *worker) toMetric(v *viewInternal, now time.Time) *metricdata.Metric {
 		startTime = w.startTimes[v]
 	}
 
-	return viewToMetric(v, now, startTime)
+	return viewToMetric(v, w.r, now, startTime)
 }
 
 // Read reads all view data and returns them as metrics.
@@ -279,3 +397,17 @@ func (w *worker) Read() []*metricdata.Metric {
 	}
 	return metrics
 }
+
+func (w *worker) RegisterExporter(e Exporter) {
+	w.exportersMu.Lock()
+	defer w.exportersMu.Unlock()
+
+	w.exporters[e] = struct{}{}
+}
+
+func (w *worker) UnregisterExporter(e Exporter) {
+	w.exportersMu.Lock()
+	defer w.exportersMu.Unlock()
+
+	delete(w.exporters, e)
+}
diff --git a/vendor/go.opencensus.io/stats/view/worker_commands.go b/vendor/go.opencensus.io/stats/view/worker_commands.go
index 0267e179a..9ac4cc059 100644
--- a/vendor/go.opencensus.io/stats/view/worker_commands.go
+++ b/vendor/go.opencensus.io/stats/view/worker_commands.go
@@ -95,7 +95,7 @@ func (cmd *unregisterFromViewReq) handleCommand(w *worker) {
 		}
 
 		// Report pending data for this view before removing it.
-		w.reportView(vi, time.Now())
+		w.reportView(vi)
 
 		vi.unsubscribe()
 		if !vi.isSubscribed() {
@@ -103,7 +103,7 @@ func (cmd *unregisterFromViewReq) handleCommand(w *worker) {
 			// The collected data can be cleared.
 			vi.clearRows()
 		}
-		w.unregisterView(name)
+		w.unregisterView(vi)
 	}
 	cmd.done <- struct{}{}
 }
@@ -163,7 +163,7 @@ func (cmd *recordReq) handleCommand(w *worker) {
 		}
 		ref := w.getMeasureRef(m.Measure().Name())
 		for v := range ref.views {
-			v.addSample(cmd.tm, m.Value(), cmd.attachments, time.Now())
+			v.addSample(cmd.tm, m.Value(), cmd.attachments, cmd.t)
 		}
 	}
 }
diff --git a/vendor/go.opencensus.io/trace/lrumap.go b/vendor/go.opencensus.io/trace/lrumap.go
index dc7a295c7..908c2497e 100644
--- a/vendor/go.opencensus.io/trace/lrumap.go
+++ b/vendor/go.opencensus.io/trace/lrumap.go
@@ -44,7 +44,7 @@ func (lm lruMap) len() int {
 }
 
 func (lm lruMap) keys() []interface{} {
-	keys := []interface{}{}
+	keys := make([]interface{}, len(lm.cacheKeys))
 	for k := range lm.cacheKeys {
 		keys = append(keys, k)
 	}
diff --git a/vendor/go.opencensus.io/trace/trace.go b/vendor/go.opencensus.io/trace/trace.go
index 3f8977b41..125e2cd90 100644
--- a/vendor/go.opencensus.io/trace/trace.go
+++ b/vendor/go.opencensus.io/trace/trace.go
@@ -345,7 +345,7 @@ func (s *Span) SetStatus(status Status) {
 }
 
 func (s *Span) interfaceArrayToLinksArray() []Link {
-	linksArr := make([]Link, 0)
+	linksArr := make([]Link, 0, len(s.links.queue))
 	for _, value := range s.links.queue {
 		linksArr = append(linksArr, value.(Link))
 	}
@@ -353,7 +353,7 @@ func (s *Span) interfaceArrayToLinksArray() []Link {
 }
 
 func (s *Span) interfaceArrayToMessageEventArray() []MessageEvent {
-	messageEventArr := make([]MessageEvent, 0)
+	messageEventArr := make([]MessageEvent, 0, len(s.messageEvents.queue))
 	for _, value := range s.messageEvents.queue {
 		messageEventArr = append(messageEventArr, value.(MessageEvent))
 	}
@@ -361,7 +361,7 @@ func (s *Span) interfaceArrayToMessageEventArray() []MessageEvent {
 }
 
 func (s *Span) interfaceArrayToAnnotationArray() []Annotation {
-	annotationArr := make([]Annotation, 0)
+	annotationArr := make([]Annotation, 0, len(s.annotations.queue))
 	for _, value := range s.annotations.queue {
 		annotationArr = append(annotationArr, value.(Annotation))
 	}
@@ -369,7 +369,7 @@ func (s *Span) interfaceArrayToAnnotationArray() []Annotation {
 }
 
 func (s *Span) lruAttributesToAttributeMap() map[string]interface{} {
-	attributes := make(map[string]interface{})
+	attributes := make(map[string]interface{}, s.lruAttributes.len())
 	for _, key := range s.lruAttributes.keys() {
 		value, ok := s.lruAttributes.get(key)
 		if ok {
@@ -420,7 +420,7 @@ func (s *Span) lazyPrintfInternal(attributes []Attribute, format string, a ...in
 	var m map[string]interface{}
 	s.mu.Lock()
 	if len(attributes) != 0 {
-		m = make(map[string]interface{})
+		m = make(map[string]interface{}, len(attributes))
 		copyAttributes(m, attributes)
 	}
 	s.annotations.add(Annotation{
@@ -436,7 +436,7 @@ func (s *Span) printStringInternal(attributes []Attribute, str string) {
 	var a map[string]interface{}
 	s.mu.Lock()
 	if len(attributes) != 0 {
-		a = make(map[string]interface{})
+		a = make(map[string]interface{}, len(attributes))
 		copyAttributes(a, attributes)
 	}
 	s.annotations.add(Annotation{
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5f4764038..6bbe92abd 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -255,7 +255,7 @@ github.com/spf13/pflag
 # github.com/tsenart/vegeta v12.7.1-0.20190725001342-b5f4fca92137+incompatible
 ## explicit
 github.com/tsenart/vegeta/lib
-# go.opencensus.io v0.22.3
+# go.opencensus.io v0.22.4
 ## explicit
 go.opencensus.io
 go.opencensus.io/internal
@@ -922,7 +922,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/test-infra v0.0.0-20200615231324-3a016f44102c
+# knative.dev/test-infra v0.0.0-20200617235125-6382dba95484
 ## explicit
 knative.dev/test-infra/scripts
 # sigs.k8s.io/boskos v0.0.0-20200530174753-71e795271860


Test patch:
diff --git a/configmap/parse_test.go b/configmap/parse_test.go
index 477dd1488..26fb3332d 100644
--- a/configmap/parse_test.go
+++ b/configmap/parse_test.go
@@ -30,6 +30,7 @@ type testConfig struct {
 	boo bool
 	i32 int32
 	i64 int64
+	u32 uint32
 	f64 float64
 	dur time.Duration
 	set sets.String
@@ -51,6 +52,7 @@ func TestParse(t *testing.T) {
 			"test-bool":     "true",
 			"test-int32":    "1",
 			"test-int64":    "2",
+			"test-uint32":   "3",
 			"test-float64":  "1.0",
 			"test-duration": "1m",
 			"test-set":      "a,b,c",
@@ -61,6 +63,7 @@ func TestParse(t *testing.T) {
 			boo: true,
 			i32: 1,
 			i64: 2,
+			u32: 3,
 			f64: 1.0,
 			dur: time.Minute,
 			set: sets.NewString("a", "b", "c"),
@@ -106,6 +109,12 @@ func TestParse(t *testing.T) {
 			"test-int64": "foo",
 		},
 		expectErr: true,
+	}, {
+		name: "uint32 error",
+		data: map[string]string{
+			"test-uint32": "foo",
+		},
+		expectErr: true,
 	}, {
 		name: "float64 error",
 		data: map[string]string{
@@ -133,6 +142,7 @@ func TestParse(t *testing.T) {
 				AsBool("test-bool", &test.conf.boo),
 				AsInt32("test-int32", &test.conf.i32),
 				AsInt64("test-int64", &test.conf.i64),
+				AsUint32("test-uint32", &test.conf.u32),
 				AsFloat64("test-float64", &test.conf.f64),
 				AsDuration("test-duration", &test.conf.dur),
 				AsStringSet("test-set", &test.conf.set),
diff --git a/controller/controller_test.go b/controller/controller_test.go
index 3f857c78d..8d8e14e72 100644
--- a/controller/controller_test.go
+++ b/controller/controller_test.go
@@ -25,15 +25,22 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	coordinationv1 "k8s.io/api/coordination/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	fakekube "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
+	"knative.dev/pkg/ptr"
+	"knative.dev/pkg/system"
+	_ "knative.dev/pkg/system/testing"
 
 	. "knative.dev/pkg/controller/testing"
+	"knative.dev/pkg/leaderelection"
 	. "knative.dev/pkg/logging/testing"
+	"knative.dev/pkg/reconciler"
 	. "knative.dev/pkg/testing"
 )
 
@@ -707,7 +714,7 @@ func TestEnqueues(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			defer ClearAll()
+			t.Cleanup(ClearAll)
 			impl := NewImplWithStats(&NopReconciler{}, TestLogger(t), "Testing", &FakeStatsReporter{})
 			test.work(impl)
 
@@ -724,7 +731,7 @@ func TestEnqueues(t *testing.T) {
 }
 
 func TestEnqeueAfter(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	impl := NewImplWithStats(&NopReconciler{}, TestLogger(t), "Testing", &FakeStatsReporter{})
 	impl.EnqueueAfter(&Resource{
 		ObjectMeta: metav1.ObjectMeta{
@@ -760,7 +767,7 @@ func TestEnqeueAfter(t *testing.T) {
 }
 
 func TestEnqeueKeyAfter(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	impl := NewImplWithStats(&NopReconciler{}, TestLogger(t), "Testing", &FakeStatsReporter{})
 	impl.EnqueueKeyAfter(types.NamespacedName{Namespace: "waiting", Name: "for"}, 5*time.Second)
 	impl.EnqueueKeyAfter(types.NamespacedName{Namespace: "the", Name: "waterfall"}, 500*time.Millisecond)
@@ -782,22 +789,23 @@ func TestEnqeueKeyAfter(t *testing.T) {
 
 type CountingReconciler struct {
 	m     sync.Mutex
-	Count int
+	count int
 }
 
 func (cr *CountingReconciler) Reconcile(context.Context, string) error {
 	cr.m.Lock()
 	defer cr.m.Unlock()
-	cr.Count++
+	cr.count++
 	return nil
 }
 
 func TestStartAndShutdown(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	r := &CountingReconciler{}
 	impl := NewImplWithStats(r, TestLogger(t), "Testing", &FakeStatsReporter{})
 
 	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
 	doneCh := make(chan struct{})
 
 	go func() {
@@ -820,18 +828,160 @@ func TestStartAndShutdown(t *testing.T) {
 		// We expect the work to complete.
 	}
 
-	if got, want := r.Count, 0; got != want {
-		t.Errorf("Count = %v, wanted %v", got, want)
+	if got, want := r.count, 0; got != want {
+		t.Errorf("count = %v, wanted %v", got, want)
+	}
+}
+
+type countingLeaderAwareReconciler struct {
+	reconciler.LeaderAwareFuncs
+
+	m     sync.Mutex
+	count int
+}
+
+var _ reconciler.LeaderAware = (*countingLeaderAwareReconciler)(nil)
+
+func (cr *countingLeaderAwareReconciler) Reconcile(ctx context.Context, key string) error {
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return err
+	}
+
+	if cr.IsLeaderFor(types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	}) {
+		cr.m.Lock()
+		defer cr.m.Unlock()
+		cr.count++
+	}
+	return nil
+}
+
+func TestStartAndShutdownWithLeaderAwareNoElection(t *testing.T) {
+	t.Cleanup(ClearAll)
+	promoted := make(chan struct{})
+	r := &countingLeaderAwareReconciler{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				close(promoted)
+				return nil
+			},
+		},
+	}
+	impl := NewImplWithStats(r, TestLogger(t), "Testing", &FakeStatsReporter{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	doneCh := make(chan struct{})
+
+	go func() {
+		defer close(doneCh)
+		StartAll(ctx, impl)
+	}()
+
+	select {
+	case <-promoted:
+		// We expect to be promoted immediately, since there is no
+		// ElectorBuilder attached to the context.
+	case <-doneCh:
+		t.Fatal("StartAll finished early.")
+	case <-time.After(10 * time.Second):
+		t.Error("Timed out waiting for StartAll.")
+	}
+
+	cancel()
+
+	select {
+	case <-time.After(1 * time.Second):
+		t.Error("Timed out waiting for controller to finish.")
+	case <-doneCh:
+		// We expect the work to complete.
+	}
+
+	if got, want := r.count, 0; got != want {
+		t.Errorf("reconcile count = %v, wanted %v", got, want)
+	}
+}
+
+func TestStartAndShutdownWithLeaderAwareWithLostElection(t *testing.T) {
+	t.Cleanup(ClearAll)
+	promoted := make(chan struct{})
+	r := &countingLeaderAwareReconciler{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				close(promoted)
+				return nil
+			},
+		},
+	}
+	cc := leaderelection.ComponentConfig{
+		Component:     "component",
+		LeaderElect:   true,
+		ResourceLock:  "leases",
+		LeaseDuration: 15 * time.Second,
+		RenewDeadline: 10 * time.Second,
+		RetryPeriod:   2 * time.Second,
+	}
+	kc := fakekube.NewSimpleClientset(
+		&coordinationv1.Lease{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      "component.testing.00-of-01",
+			},
+			Spec: coordinationv1.LeaseSpec{
+				HolderIdentity:       ptr.String("not-us"),
+				LeaseDurationSeconds: ptr.Int32(3000),
+				AcquireTime:          &metav1.MicroTime{time.Now()},
+				RenewTime:            &metav1.MicroTime{time.Now().Add(3000 * time.Second)},
+			},
+		},
+	)
+
+	impl := NewImplWithStats(r, TestLogger(t), "Testing", &FakeStatsReporter{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	ctx = leaderelection.WithStandardLeaderElectorBuilder(ctx, kc, cc)
+	doneCh := make(chan struct{})
+
+	go func() {
+		defer close(doneCh)
+		StartAll(ctx, impl)
+	}()
+
+	select {
+	case <-promoted:
+		t.Fatal("Unexpected promotion.")
+	case <-time.After(3 * time.Second):
+		// Wait for 3 seconds for good measure.
+	case <-doneCh:
+		t.Error("StartAll finished early.")
+	}
+
+	cancel()
+
+	select {
+	case <-time.After(1 * time.Second):
+		t.Error("Timed out waiting for controller to finish.")
+	case <-doneCh:
+		// We expect the work to complete.
+	}
+
+	if got, want := r.count, 0; got != want {
+		t.Errorf("reconcile count = %v, wanted %v", got, want)
 	}
 }
 
 func TestStartAndShutdownWithWork(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	r := &CountingReconciler{}
 	reporter := &FakeStatsReporter{}
 	impl := NewImplWithStats(r, TestLogger(t), "Testing", reporter)
 
 	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
 	doneCh := make(chan struct{})
 
 	impl.EnqueueKey(types.NamespacedName{Namespace: "foo", Name: "bar"})
@@ -856,11 +1006,11 @@ func TestStartAndShutdownWithWork(t *testing.T) {
 		// We expect the work to complete.
 	}
 
-	if got, want := r.Count, 1; got != want {
-		t.Errorf("Count = %v, wanted %v", got, want)
+	if got, want := r.count, 1; got != want {
+		t.Errorf("reconcile count = %v, wanted %v", got, want)
 	}
 	if got, want := impl.WorkQueue.NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), 0; got != want {
-		t.Errorf("Count = %v, wanted %v", got, want)
+		t.Errorf("requeues = %v, wanted %v", got, want)
 	}
 
 	checkStats(t, reporter, 1, 0, 1, trueString)
@@ -902,12 +1052,13 @@ func (er *ErrorReconciler) Reconcile(context.Context, string) error {
 }
 
 func TestStartAndShutdownWithErroringWork(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	r := &ErrorReconciler{}
 	reporter := &FakeStatsReporter{}
 	impl := NewImplWithStats(r, TestLogger(t), "Testing", reporter)
 
 	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
 	doneCh := make(chan struct{})
 
 	impl.EnqueueKey(types.NamespacedName{Namespace: "", Name: "bar"})
@@ -956,12 +1107,13 @@ func (er *PermanentErrorReconciler) Reconcile(context.Context, string) error {
 }
 
 func TestStartAndShutdownWithPermanentErroringWork(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	r := &PermanentErrorReconciler{}
 	reporter := &FakeStatsReporter{}
 	impl := NewImplWithStats(r, TestLogger(t), "Testing", reporter)
 
 	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
 	doneCh := make(chan struct{})
 
 	impl.EnqueueKey(types.NamespacedName{Namespace: "foo", Name: "bar"})
@@ -1048,11 +1200,12 @@ func (*dummyStore) List() []interface{} {
 }
 
 func TestImplGlobalResync(t *testing.T) {
-	defer ClearAll()
+	t.Cleanup(ClearAll)
 	r := &CountingReconciler{}
 	impl := NewImplWithStats(r, TestLogger(t), "Testing", &FakeStatsReporter{})
 
 	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
 	doneCh := make(chan struct{})
 
 	go func() {
@@ -1079,7 +1232,7 @@ func TestImplGlobalResync(t *testing.T) {
 		// We expect the work to complete.
 	}
 
-	if want, got := 3, r.Count; want != got {
+	if want, got := 3, r.count; want != got {
 		t.Errorf("GlobalResync: want = %v, got = %v", want, got)
 	}
 }
@@ -1319,6 +1472,7 @@ func TestRunInformersFinished(t *testing.T) {
 	}()
 
 	ctx, cancel := context.WithCancel(TestContextWithLogger(t))
+	t.Cleanup(cancel)
 
 	waitInformers, err := RunInformers(ctx.Done(), fi)
 	if err != nil {
diff --git a/leaderelection/config_test.go b/leaderelection/config_test.go
index 3d3767788..e2850c8a8 100644
--- a/leaderelection/config_test.go
+++ b/leaderelection/config_test.go
@@ -18,6 +18,8 @@ package leaderelection
 
 import (
 	"errors"
+	"fmt"
+	"strconv"
 	"testing"
 	"time"
 
@@ -30,6 +32,7 @@ import (
 func okConfig() *Config {
 	return &Config{
 		ResourceLock:      "leases",
+		Buckets:           1,
 		LeaseDuration:     15 * time.Second,
 		RenewDeadline:     10 * time.Second,
 		RetryPeriod:       2 * time.Second,
@@ -40,6 +43,7 @@ func okConfig() *Config {
 func okData() map[string]string {
 	return map[string]string{
 		"resourceLock": "leases",
+		"buckets":      "1",
 		// values in this data come from the defaults suggested in the
 		// code:
 		// https://github.com/kubernetes/client-go/blob/kubernetes-1.16.0/tools/leaderelection/leaderelection.go
@@ -72,6 +76,17 @@ func TestNewConfigMapFromData(t *testing.T) {
 			config.EnabledComponents.Insert("controller")
 			return config
 		}(),
+	}, {
+		name: "OK config - controller enabled with multiple buckets",
+		data: kmeta.UnionMaps(okData(), map[string]string{
+			"buckets": "5",
+		}),
+		expected: func() *Config {
+			config := okConfig()
+			config.EnabledComponents.Insert("controller")
+			config.Buckets = 5
+			return config
+		}(),
 	}, {
 		name: "invalid resourceLock",
 		data: kmeta.UnionMaps(okData(), map[string]string{
@@ -96,6 +111,24 @@ func TestNewConfigMapFromData(t *testing.T) {
 			"retryPeriod": "flops",
 		}),
 		err: errors.New(`failed to parse "retryPeriod": time: invalid duration flops`),
+	}, {
+		name: "invalid buckets - not an int",
+		data: kmeta.UnionMaps(okData(), map[string]string{
+			"buckets": "not-an-int",
+		}),
+		err: errors.New(`failed to parse "buckets": strconv.ParseUint: parsing "not-an-int": invalid syntax`),
+	}, {
+		name: "invalid buckets - too small",
+		data: kmeta.UnionMaps(okData(), map[string]string{
+			"buckets": "0",
+		}),
+		err: fmt.Errorf("buckets: value must be between 1 <= 0 <= %d", MaxBuckets),
+	}, {
+		name: "invalid buckets - too large",
+		data: kmeta.UnionMaps(okData(), map[string]string{
+			"buckets": strconv.Itoa(int(MaxBuckets + 1)),
+		}),
+		err: fmt.Errorf(`buckets: value must be between 1 <= %d <= %d`, MaxBuckets+1, MaxBuckets),
 	}}
 
 	for _, tc := range cases {
diff --git a/leaderelection/context_test.go b/leaderelection/context_test.go
new file mode 100644
index 000000000..5e9bc6ce0
--- /dev/null
+++ b/leaderelection/context_test.go
@@ -0,0 +1,146 @@
+// +build !race
+// TODO(https://github.com/kubernetes/kubernetes/issues/90952): Remove the above.
+
+/*
+Copyright 2020 The Knative 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 leaderelection
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	fakekube "k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+	"knative.dev/pkg/reconciler"
+	_ "knative.dev/pkg/system/testing"
+)
+
+func TestWithBuilder(t *testing.T) {
+	cc := ComponentConfig{
+		Component:     "component",
+		LeaderElect:   true,
+		Buckets:       1,
+		ResourceLock:  "leases",
+		LeaseDuration: 15 * time.Second,
+		RenewDeadline: 10 * time.Second,
+		RetryPeriod:   2 * time.Second,
+	}
+	kc := fakekube.NewSimpleClientset()
+	ctx := context.Background()
+
+	promoted := make(chan struct{})
+	demoted := make(chan struct{})
+	laf := &reconciler.LeaderAwareFuncs{
+		PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+			close(promoted)
+			return nil
+		},
+		DemoteFunc: func(bkt reconciler.Bucket) {
+			close(demoted)
+		},
+	}
+	enq := func(reconciler.Bucket, types.NamespacedName) {}
+
+	created := make(chan struct{})
+	kc.PrependReactor("create", "leases",
+		func(action ktesting.Action) (bool, runtime.Object, error) {
+			close(created)
+			return false, nil, nil
+		},
+	)
+
+	updated := make(chan struct{})
+	kc.PrependReactor("update", "leases",
+		func(action ktesting.Action) (bool, runtime.Object, error) {
+			// Only close update once.
+			select {
+			case <-updated:
+			default:
+				close(updated)
+			}
+			return false, nil, nil
+		},
+	)
+
+	if HasLeaderElection(ctx) {
+		t.Error("HasLeaderElection() = true, wanted false")
+	}
+	if le, err := BuildElector(ctx, laf, "name", enq); err != nil {
+		t.Errorf("BuildElector() = %v, wanted an unopposedElector", err)
+	} else if _, ok := le.(*unopposedElector); !ok {
+		t.Errorf("BuildElector() = %T, wanted an unopposedElector", le)
+	}
+
+	ctx = WithStandardLeaderElectorBuilder(ctx, kc, cc)
+	if !HasLeaderElection(ctx) {
+		t.Error("HasLeaderElection() = false, wanted true")
+	}
+
+	le, err := BuildElector(ctx, laf, "name", enq)
+	if err != nil {
+		t.Errorf("BuildElector() = %v", err)
+	}
+
+	// We shouldn't see leases until we Run the elector.
+	select {
+	case <-promoted:
+		t.Error("Got promoted, want no actions.")
+	case <-demoted:
+		t.Error("Got demoted, want no actions.")
+	case <-created:
+		t.Error("Got created, want no actions.")
+	case <-updated:
+		t.Error("Got updated, want no actions.")
+	default:
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go le.Run(ctx)
+
+	select {
+	case <-created:
+		// We expect the lease to be created.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for lease creation.")
+	}
+	select {
+	case <-promoted:
+		// We expect to have been promoted.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for promotion.")
+	}
+
+	// Cancelling the context should case us to give up leadership.
+	cancel()
+
+	select {
+	case <-updated:
+		// We expect the lease to be updated.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for lease update.")
+	}
+	select {
+	case <-demoted:
+		// We expect to have been demoted.
+	case <-time.After(1 * time.Second):
+		t.Fatal("Timed out waiting for demotion.")
+	}
+}
diff --git a/metrics/config_test.go b/metrics/config_test.go
index dede19e4a..8f9857918 100644
--- a/metrics/config_test.go
+++ b/metrics/config_test.go
@@ -1045,11 +1045,11 @@ func TestStackdriverRecord(t *testing.T) {
 				{Measure: servedCount, Aggregation: view.Count()},
 				{Measure: statCount, Aggregation: view.Count()},
 			}
-			err = view.Register(v...)
+			err = RegisterResourceView(v...)
 			if err != nil {
 				t.Errorf("Failed to register %+v in stats backend: %v", v, err)
 			}
-			defer view.Unregister(v...)
+			defer UnregisterResourceView(v...)
 
 			// Try recording each metric and checking the result.
 			Record(ctx, servedCount.M(1))
diff --git a/metrics/metricstest/metricstest.go b/metrics/metricstest/metricstest.go
index 7cf8df8cc..9d74c4019 100644
--- a/metrics/metricstest/metricstest.go
+++ b/metrics/metricstest/metricstest.go
@@ -17,8 +17,10 @@ limitations under the License.
 package metricstest
 
 import (
+	"fmt"
 	"reflect"
 
+	"go.opencensus.io/metric/metricproducer"
 	"go.opencensus.io/stats/view"
 	"knative.dev/pkg/test"
 )
@@ -56,14 +58,17 @@ func CheckStatsNotReported(t test.T, names ...string) {
 // reported are tagged with the tags in wantTags and that wantValue matches reported count.
 func CheckCountData(t test.T, name string, wantTags map[string]string, wantValue int64) {
 	t.Helper()
-	if row := checkExactlyOneRow(t, name); row != nil {
-		checkRowTags(t, row, name, wantTags)
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
 
-		if s, ok := row.Data.(*view.CountData); !ok {
-			t.Error("want CountData", "metric", name, "got", reflect.TypeOf(row.Data))
-		} else if s.Value != wantValue {
-			t.Error("Wrong value", "metric", name, "value", s.Value, "want", wantValue)
-		}
+	if s, ok := row.Data.(*view.CountData); !ok {
+		t.Error("want CountData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else if s.Value != wantValue {
+		t.Error("Wrong value", "metric", name, "value", s.Value, "want", wantValue)
 	}
 }
 
@@ -72,21 +77,24 @@ func CheckCountData(t test.T, name string, wantTags map[string]string, wantValue
 // It also checks that expectedMin and expectedMax match the minimum and maximum reported values, respectively.
 func CheckDistributionData(t test.T, name string, wantTags map[string]string, expectedCount int64, expectedMin float64, expectedMax float64) {
 	t.Helper()
-	if row := checkExactlyOneRow(t, name); row != nil {
-		checkRowTags(t, row, name, wantTags)
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
 
-		if s, ok := row.Data.(*view.DistributionData); !ok {
-			t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
-		} else {
-			if s.Count != expectedCount {
-				t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
-			}
-			if s.Min != expectedMin {
-				t.Error("reporter count wrong", "metric", name, "got", s.Min, "want", expectedMin)
-			}
-			if s.Max != expectedMax {
-				t.Error("reporter count wrong", "metric", name, "got", s.Max, "want", expectedMax)
-			}
+	if s, ok := row.Data.(*view.DistributionData); !ok {
+		t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else {
+		if s.Count != expectedCount {
+			t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
+		}
+		if s.Min != expectedMin {
+			t.Error("reporter count wrong", "metric", name, "got", s.Min, "want", expectedMin)
+		}
+		if s.Max != expectedMax {
+			t.Error("reporter count wrong", "metric", name, "got", s.Max, "want", expectedMax)
 		}
 	}
 }
@@ -95,15 +103,19 @@ func CheckDistributionData(t test.T, name string, wantTags map[string]string, ex
 // are tagged with the tags in wantTags and that expectedCount number of records were reported.
 func CheckDistributionCount(t test.T, name string, wantTags map[string]string, expectedCount int64) {
 	t.Helper()
-	if row := checkExactlyOneRow(t, name); row != nil {
-		checkRowTags(t, row, name, wantTags)
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
 
-		if s, ok := row.Data.(*view.DistributionData); !ok {
-			t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
-		} else if s.Count != expectedCount {
-			t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
-		}
+	if s, ok := row.Data.(*view.DistributionData); !ok {
+		t.Error("want DistributionData", "metric", name, "got", reflect.TypeOf(row.Data))
+	} else if s.Count != expectedCount {
+		t.Error("reporter count wrong", "metric", name, "got", s.Count, "want", expectedCount)
 	}
+
 }
 
 // CheckLastValueData checks the view with a name matching string name to verify that the LastValueData stats
@@ -125,14 +137,17 @@ func CheckLastValueData(t test.T, name string, wantTags map[string]string, wantV
 // reported are tagged with the tags in wantTags and that wantValue matches the reported sum.
 func CheckSumData(t test.T, name string, wantTags map[string]string, wantValue float64) {
 	t.Helper()
-	if row := checkExactlyOneRow(t, name); row != nil {
-		checkRowTags(t, row, name, wantTags)
+	row, err := checkExactlyOneRow(t, name)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	checkRowTags(t, row, name, wantTags)
 
-		if s, ok := row.Data.(*view.SumData); !ok {
-			t.Error("Wrong type", "metric", name, "got", reflect.TypeOf(row.Data), "want", "SumData")
-		} else if s.Value != wantValue {
-			t.Error("Wrong sumdata", "metric", name, "got", s.Value, "want", wantValue)
-		}
+	if s, ok := row.Data.(*view.SumData); !ok {
+		t.Error("Wrong type", "metric", name, "got", reflect.TypeOf(row.Data), "want", "SumData")
+	} else if s.Value != wantValue {
+		t.Error("Wrong sumdata", "metric", name, "got", s.Value, "want", wantValue)
 	}
 }
 
@@ -166,19 +181,29 @@ func lastRow(t test.T, name string) *view.Row {
 	return d[len(d)-1]
 }
 
-func checkExactlyOneRow(t test.T, name string) *view.Row {
-	t.Helper()
-	d, err := view.RetrieveData(name)
-	if err != nil {
-		t.Error("Reporter.Report() error", "metric", name, "error", err)
-		return nil
+func checkExactlyOneRow(t test.T, name string) (*view.Row, error) {
+	// view.Meter implements (and is exposed by) metricproducer.GetAll. Since
+	// this is a test, reach around and cast these to view.Meter.
+	var retval *view.Row
+	for _, producer := range metricproducer.GlobalManager().GetAll() {
+		meter := producer.(view.Meter)
+
+		d, err := meter.RetrieveData(name)
+		if err != nil || len(d) == 0 {
+			continue
+		}
+		if len(d) > 1 {
+			return nil, fmt.Errorf("expected 1 row for metric %q got %d", name, len(d))
+		}
+		if retval != nil {
+			return nil, fmt.Errorf("got 2 rows from different meters: %+v, %+v", *retval, d[0])
+		}
+		retval = d[0]
 	}
-	if len(d) != 1 {
-		t.Error("Reporter.Report() wrong length", "metric", name, "got", len(d), "want", 1)
-		return nil
+	if retval == nil {
+		return nil, fmt.Errorf("could not find row for %q", name)
 	}
-
-	return d[0]
+	return retval, nil
 }
 
 func checkRowTags(t test.T, row *view.Row, name string, wantTags map[string]string) {
diff --git a/reconciler/leader_test.go b/reconciler/leader_test.go
new file mode 100644
index 000000000..cb57b8c7c
--- /dev/null
+++ b/reconciler/leader_test.go
@@ -0,0 +1,82 @@
+/*
+Copyright 2020 The Knative 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 reconciler
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func TestLeaderAwareFuncs(t *testing.T) {
+	laf := LeaderAwareFuncs{}
+	wantBkt := UniversalBucket()
+	wantKey := types.NamespacedName{
+		Namespace: "foo",
+		Name:      "bar",
+	}
+	called := false
+	wantFunc := func(gotBkt Bucket, gotKey types.NamespacedName) {
+		called = true
+		if !cmp.Equal(gotKey, wantKey) {
+			t.Errorf("key (-want, +got) = %s", cmp.Diff(wantKey, gotKey))
+		}
+		if !cmp.Equal(gotBkt, wantBkt) {
+			t.Errorf("bucket (-want, +got) = %s", cmp.Diff(wantBkt, gotBkt))
+		}
+	}
+
+	laf.PromoteFunc = func(bkt Bucket, gotFunc func(Bucket, types.NamespacedName)) error {
+		gotFunc(bkt, wantKey)
+		if !called {
+			t.Error("gotFunc didn't call wantFunc!")
+		}
+
+		// IsLeaderFor takes the bucket's lock, so make sure that the callback
+		// we provide is not called while the lock is still held by calling a
+		// function that we know takes the lock.
+		if !laf.IsLeaderFor(wantKey) {
+			t.Error("IsLeaderFor() = false, wanted true")
+		}
+		return nil
+	}
+	laf.DemoteFunc = func(bkt Bucket) {
+		// Check that we're not called while the lock is held,
+		// and that we are no longer leader.
+		if laf.IsLeaderFor(wantKey) {
+			t.Error("IsLeaderFor() = true, wanted false")
+		}
+	}
+
+	// We don't start as leader.
+	if laf.IsLeaderFor(wantKey) {
+		t.Error("IsLeaderFor() = true, wanted false")
+	}
+
+	// After Promote we are leader.
+	laf.Promote(wantBkt, wantFunc)
+	if !laf.IsLeaderFor(wantKey) {
+		t.Error("IsLeaderFor() = false, wanted true")
+	}
+
+	// After Demote we are not leader.
+	laf.Demote(wantBkt)
+	if laf.IsLeaderFor(wantKey) {
+		t.Error("IsLeaderFor() = true, wanted false")
+	}
+}
diff --git a/reconciler/reconcile_common_test.go b/reconciler/reconcile_common_test.go
index b1956f94e..bda8540ac 100644
--- a/reconciler/reconcile_common_test.go
+++ b/reconciler/reconcile_common_test.go
@@ -19,6 +19,7 @@ package reconciler
 import (
 	"context"
 	"testing"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -99,13 +100,43 @@ func TestPostProcessReconcileBumpsGeneration(t *testing.T) {
 	resource := makeResource()
 
 	krShape := duckv1.KRShaped(resource)
-	PostProcessReconcile(context.Background(), krShape)
+	PostProcessReconcile(context.Background(), krShape, krShape)
 
 	if resource.Status.ObservedGeneration != resource.Generation {
-		t.Errorf("Expected observed generation bump got=%d want=%d", resource.Status.ObservedGeneration, resource.Generation)
+		t.Errorf("Expected observed generation bump got=%d want=%d",
+			resource.Status.ObservedGeneration, resource.Generation)
 	}
 
 	if krShape.GetStatus().ObservedGeneration != krShape.GetGeneration() {
-		t.Errorf("Expected observed generation bump got=%d want=%d", resource.Status.ObservedGeneration, resource.Generation)
+		t.Errorf("Expected observed generation bump got=%d want=%d",
+			resource.Status.ObservedGeneration, resource.Generation)
+	}
+}
+
+func TestPostProcessReconcileUpdatesTransitionTimes(t *testing.T) {
+	oldNow := apis.VolatileTime{Inner: metav1.NewTime(time.Now())}
+	resource := makeResource()
+	oldResource := makeResource()
+	// initialize old conditions with oldNow
+	oldResource.Status.Conditions[0].LastTransitionTime = oldNow
+	oldResource.Status.Conditions[1].LastTransitionTime = oldNow
+	// change the second condition, but keep the old timestamp.
+	resource.Status.Conditions[1].LastTransitionTime = oldNow
+	resource.Status.Conditions[1].Status = corev1.ConditionFalse
+
+	new := duckv1.KRShaped(resource)
+	old := duckv1.KRShaped(oldResource)
+	PostProcessReconcile(context.Background(), new, old)
+
+	unchangedCond := resource.Status.Conditions[0]
+	if unchangedCond.LastTransitionTime != oldNow {
+		t.Errorf("Expected unchanged condition to keep old timestamp. Got=%v Want=%v",
+			unchangedCond.LastTransitionTime, oldNow)
+	}
+
+	changedCond := resource.Status.Conditions[1]
+	if changedCond.LastTransitionTime == oldNow {
+		t.Errorf("Expected changed condition to get a new timestamp. Got=%v Want=%v",
+			changedCond.LastTransitionTime, oldNow)
 	}
 }
diff --git a/test/presubmit-tests.sh b/test/presubmit-tests.sh
index 261599f71..af8199dd0 100755
--- a/test/presubmit-tests.sh
+++ b/test/presubmit-tests.sh
@@ -37,4 +37,16 @@ function pre_build_tests() {
   return 0
 }
 
+# Run the unit tests with an additional flag '-mod=vendor' to avoid
+# downloading the deps in unit tests CI job
+function unit_tests() {
+  # Run the default way.
+  default_unit_test_runner || failed=1
+
+  # Run unit testing select packages without race detection,
+  # so that they may use: // +build !race
+  report_go_test ./leaderelection || failed=1
+
+}
+
 main $@
diff --git a/vendor/knative.dev/test-infra/scripts/e2e-tests.sh b/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
index 01b2d0e40..354819385 100644
--- a/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
+++ b/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
@@ -108,6 +108,8 @@ function dump_cluster_state() {
       echo ">>> Details" >> ${output}
       if [[ "${crd}" == "secrets" ]]; then
         echo "Secrets are ignored for security reasons" >> ${output}
+      elif [[ "${crd}" == "events" ]]; then
+        echo "events are ignored as making a lot of noise" >> ${output}
       else
         kubectl get ${crd} --all-namespaces -o yaml >> ${output}
       fi