Repository: knative/serving
Pull request #10337 (2020-12-10 02:04:06)

Add stepping logic to compute min next time and re-enqueue
Start returning minimal next time from the `Step`.
Then reenque at that time to ensure we reconcile the route
again to step some more.

/assign @tcnghia 

/hold
For #10334 to merge (probably will need merge conflicts :().

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

This pull-request has been approved by: *<a href="https://github.com/knative/serving/pull/10337#" title="Author self-approved">vagababov</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[pkg/reconciler/route/OWNERS](https://github.com/knative/serving/blob/master/pkg/reconciler/route/OWNERS)~~ [vagababov]

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":[]} -->
vagababov: Also added a few more interesting test cases. 😁 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=h1) Report
> Merging [#10337](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=desc) (eff07ed) into [master](https://codecov.io/gh/knative/serving/commit/0243c480fb6778430224ac58b967e21a570d8356?el=desc) (0243c48) will **increase** coverage by `0.00%`.
> The diff coverage is `83.33%`.

[![Impacted file tree graph](https://codecov.io/gh/knative/serving/pull/10337/graphs/tree.svg?width=650&height=150&src=pr&token=MPpDfAj4Ui)](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master   #10337   +/-   ##
=======================================
  Coverage   88.21%   88.22%           
=======================================
  Files         184      184           
  Lines        8674     8685   +11     
=======================================
+ Hits         7652     7662   +10     
  Misses        787      787           
- Partials      235      236    +1     
```


| [Impacted Files](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/reconciler/route/route.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvcm91dGUvcm91dGUuZ28=) | `76.24% <ø> (ø)` | |
| [pkg/reconciler/route/reconcile\_resources.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvcm91dGUvcmVjb25jaWxlX3Jlc291cmNlcy5nbw==) | `84.82% <50.00%> (-2.22%)` | :arrow_down: |
| [pkg/reconciler/route/controller.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvcm91dGUvY29udHJvbGxlci5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/reconciler/route/traffic/rollout.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvcm91dGUvdHJhZmZpYy9yb2xsb3V0Lmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/reconciler/autoscaling/kpa/scaler.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvYXV0b3NjYWxpbmcva3BhL3NjYWxlci5nbw==) | `88.57% <0.00%> (-1.43%)` | :arrow_down: |
| [pkg/activator/net/revision\_backends.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL2FjdGl2YXRvci9uZXQvcmV2aXNpb25fYmFja2VuZHMuZ28=) | `91.40% <0.00%> (+0.90%)` | :arrow_up: |
| [pkg/reconciler/configuration/configuration.go](https://codecov.io/gh/knative/serving/pull/10337/diff?src=pr&el=tree#diff-cGtnL3JlY29uY2lsZXIvY29uZmlndXJhdGlvbi9jb25maWd1cmF0aW9uLmdv) | `88.28% <0.00%> (+1.56%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=footer). Last update [0243c48...eff07ed](https://codecov.io/gh/knative/serving/pull/10337?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

vagababov: /hold cancel

Patch:
diff --git a/cmd/autoscaler/main.go b/cmd/autoscaler/main.go
index 8153d8c5a..4daafda48 100644
--- a/cmd/autoscaler/main.go
+++ b/cmd/autoscaler/main.go
@@ -147,15 +147,30 @@ func main() {
 		logger.Fatalw("Failed to start informers", zap.Error(err))
 	}
 
-	cc, selfIP := componentConfigAndIP(ctx, logger)
-	ctx = leaderelection.WithStandardLeaderElectorBuilder(ctx, kubeClient, cc)
+	cc := componentConfigAndIP(ctx)
 
 	// accept is the func to call when this pod owns the Revision for this StatMessage.
 	accept := func(sm asmetrics.StatMessage) {
 		collector.Record(sm.Key, time.Unix(sm.Stat.Timestamp, 0), sm.Stat)
 		multiScaler.Poke(sm.Key, sm.Stat)
 	}
-	f := statforwarder.New(ctx, logger, kubeClient, selfIP, bucket.AutoscalerBucketSet(cc.Buckets), accept)
+
+	var f *statforwarder.Forwarder
+	if b, bs, err := leaderelection.NewStatefulSetBucketAndSet(int(cc.Buckets)); err == nil {
+		logger.Info("Running with StatefulSet leader election")
+		ctx = leaderelection.WithStatefulSetElectorBuilder(ctx, cc, b)
+		f = statforwarder.New(ctx, bs)
+		if err := statforwarder.StatefulSetBasedProcessor(ctx, f, accept); err != nil {
+			logger.Fatalw("Failed to set up statefulset processors", zap.Error(err))
+		}
+	} else {
+		logger.Info("Running with Standard leader election")
+		ctx = leaderelection.WithStandardLeaderElectorBuilder(ctx, kubeClient, cc)
+		f = statforwarder.New(ctx, bucket.AutoscalerBucketSet(cc.Buckets))
+		if err := statforwarder.LeaseBasedProcessor(ctx, f, accept); err != nil {
+			logger.Fatalw("Failed to set up lease tracking", zap.Error(err))
+		}
+	}
 
 	// Set up a statserver.
 	statsServer := statserver.New(statsServerAddr, statsCh, logger, f.IsBucketOwner)
@@ -235,21 +250,16 @@ func flush(logger *zap.SugaredLogger) {
 	metrics.FlushExporter()
 }
 
-func componentConfigAndIP(ctx context.Context, logger *zap.SugaredLogger) (leaderelection.ComponentConfig, string) {
+func componentConfigAndIP(ctx context.Context) leaderelection.ComponentConfig {
 	id, err := bucket.Identity()
 	if err != nil {
-		logger.Fatalw("Failed to generate Lease holder identify", zap.Error(err))
-	}
-
-	_, ip, err := bucket.ExtractPodNameAndIP(id)
-	if err != nil {
-		logger.Fatalw("Failed to extract IP from identify", zap.Error(err))
+		logging.FromContext(ctx).Fatalw("Failed to generate Lease holder identify", zap.Error(err))
 	}
 
 	// Set up leader election config
 	leaderElectionConfig, err := sharedmain.GetLeaderElectionConfig(ctx)
 	if err != nil {
-		logger.Fatal("Error loading leader election configuration: ", err)
+		logging.FromContext(ctx).Fatalw("Error loading leader election configuration", zap.Error(err))
 	}
 
 	cc := leaderElectionConfig.GetComponentConfig(component)
@@ -258,5 +268,5 @@ func componentConfigAndIP(ctx context.Context, logger *zap.SugaredLogger) (leade
 	}
 	cc.Identity = id
 
-	return cc, ip
+	return cc
 }
diff --git a/go.mod b/go.mod
index 6eac9dd7a..784e2fd62 100644
--- a/go.mod
+++ b/go.mod
@@ -34,8 +34,8 @@ require (
 	k8s.io/client-go v0.18.12
 	k8s.io/code-generator v0.18.12
 	k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29
-	knative.dev/caching v0.0.0-20201208013740-3ca9791186ed
+	knative.dev/caching v0.0.0-20201209181541-0e55b559628c
 	knative.dev/hack v0.0.0-20201201234937-fddbf732e450
-	knative.dev/networking v0.0.0-20201209014241-22a7e22628fc
+	knative.dev/networking v0.0.0-20201209181441-896c0e7c4864
 	knative.dev/pkg v0.0.0-20201209172941-9b9407a76df1
 )
diff --git a/go.sum b/go.sum
index 2f036d511..afa9d7140 100644
--- a/go.sum
+++ b/go.sum
@@ -1216,15 +1216,12 @@ k8s.io/legacy-cloud-providers v0.18.8/go.mod h1:tgp4xYf6lvjrWnjQwTOPvWQE9IVqSBGP
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/caching v0.0.0-20201208013740-3ca9791186ed h1:Hl6jZZFVPRv++MDHtD9hy8lhJ/38HVImg+uJtBPo7BQ=
-knative.dev/caching v0.0.0-20201208013740-3ca9791186ed/go.mod h1:i+e4gaoBB34anvKMMiQDV86N7HbFYN6m+Op1IRhXXY8=
+knative.dev/caching v0.0.0-20201209181541-0e55b559628c h1:8Z08V3wrbHq04dnLdRDedkNN6wZUFTp1h3lXLJYMJ5g=
+knative.dev/caching v0.0.0-20201209181541-0e55b559628c/go.mod h1:RVYUjyxzWfX/l/FEi6YKcYZ3ngjT6qIxpTtiIvFMBg0=
 knative.dev/hack v0.0.0-20201201234937-fddbf732e450 h1:IyitWF7OzfunCgE4b9ZsJAeIRoQ6JOyqDrz/19X4iS4=
 knative.dev/hack v0.0.0-20201201234937-fddbf732e450/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/networking v0.0.0-20201209014241-22a7e22628fc h1:Zav6xAGiSKddU2ra70oVgYkB3KjJUpiTqi7A5iAaUoE=
-knative.dev/networking v0.0.0-20201209014241-22a7e22628fc/go.mod h1:NPlekJSE5w2vw5mcu8zCBdHsiBq9U/waDbFXrEvpw6Q=
-knative.dev/pkg v0.0.0-20201207225040-7e918163fc4f/go.mod h1:Jtbz2/1gB8EZgZA4VvabTxM+AcDt2WL2P0RaFnKcOX8=
-knative.dev/pkg v0.0.0-20201209011241-b1ac587fdd80 h1:wBRwTtE3saOiXsS/eh4FYDAC/Gj+djH8ZNkvd0OgtmE=
-knative.dev/pkg v0.0.0-20201209011241-b1ac587fdd80/go.mod h1:Jtbz2/1gB8EZgZA4VvabTxM+AcDt2WL2P0RaFnKcOX8=
+knative.dev/networking v0.0.0-20201209181441-896c0e7c4864 h1:eSfdkI4Y3UWq49mM9K+dL4J8KvRx0TNQ9Zg+gYhETkI=
+knative.dev/networking v0.0.0-20201209181441-896c0e7c4864/go.mod h1:s+q5zIi9skLPnC7T3R6W4enZRnT8I/eIwWuys8gGJm4=
 knative.dev/pkg v0.0.0-20201209172941-9b9407a76df1 h1:whfizV9at+XvlOK66xm1WfiiPhybrRa9LbZa4uXbWIs=
 knative.dev/pkg v0.0.0-20201209172941-9b9407a76df1/go.mod h1:Jtbz2/1gB8EZgZA4VvabTxM+AcDt2WL2P0RaFnKcOX8=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/pkg/autoscaler/bucket/bucket.go b/pkg/autoscaler/bucket/bucket.go
index 5dd8281ee..34c794482 100644
--- a/pkg/autoscaler/bucket/bucket.go
+++ b/pkg/autoscaler/bucket/bucket.go
@@ -52,10 +52,9 @@ func AutoscalerBucketSet(total uint32) *hash.BucketSet {
 	return hash.NewBucketSet(names)
 }
 
-// Identity returns a identify for this Autoscaler pod used as the Lease holder
-// identity. It's in the format of <POD-NAME>_<POD-IP> whose information is ready from
-// environment variables.
-func Identity() (string, error) {
+// PodIP returns the IP address of the current pod, or an error
+// if it wasn't properly projected.
+func PodIP() (string, error) {
 	selfIP, existing := os.LookupEnv("POD_IP")
 	if !existing {
 		return "", errors.New("POD_IP environment variable not set")
@@ -63,6 +62,17 @@ func Identity() (string, error) {
 	if selfIP == "" {
 		return "", errors.New("POD_IP environment variable is empty")
 	}
+	return selfIP, nil
+}
+
+// Identity returns a identify for this Autoscaler pod used as the Lease holder
+// identity. It's in the format of <POD-NAME>_<POD-IP> whose information is ready from
+// environment variables.
+func Identity() (string, error) {
+	selfIP, err := PodIP()
+	if err != nil {
+		return "", err
+	}
 
 	podName, existing := os.LookupEnv("POD_NAME")
 	if !existing {
diff --git a/pkg/autoscaler/statforwarder/forwarder.go b/pkg/autoscaler/statforwarder/forwarder.go
index 30efe1782..5c310f103 100644
--- a/pkg/autoscaler/statforwarder/forwarder.go
+++ b/pkg/autoscaler/statforwarder/forwarder.go
@@ -23,24 +23,10 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	coordinationv1 "k8s.io/api/coordination/v1"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/equality"
-	apierrs "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	corev1listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
-	leaseinformer "knative.dev/pkg/client/injection/kube/informers/coordination/v1/lease"
-	endpointsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints"
-	"knative.dev/pkg/controller"
 	"knative.dev/pkg/hash"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/logging/logkey"
 	"knative.dev/pkg/network"
-	"knative.dev/pkg/system"
-	"knative.dev/serving/pkg/autoscaler/bucket"
 	asmetrics "knative.dev/serving/pkg/autoscaler/metrics"
 )
 
@@ -92,11 +78,13 @@ type Forwarder struct {
 }
 
 // New creates a new Forwarder.
-func New(ctx context.Context, logger *zap.SugaredLogger, kc kubernetes.Interface, selfIP string, bs *hash.BucketSet, accept statProcessor) *Forwarder {
+// This must be configured with a mechanism for setting up its "processors",
+// such as LeaseBasedProcessor or StatefulSetBasedProcessor, which correlates
+// with the mechanism of leader election being used.
+func New(ctx context.Context, bs *hash.BucketSet) *Forwarder {
 	bkts := bs.Buckets()
-	endpointsInformer := endpointsinformer.Get(ctx)
 	f := &Forwarder{
-		logger:     logger,
+		logger:     logging.FromContext(ctx),
 		bs:         bs,
 		processors: make(map[string]bucketProcessor, len(bkts)),
 		statCh:     make(chan stat, 1000),
@@ -106,254 +94,9 @@ func New(ctx context.Context, logger *zap.SugaredLogger, kc kubernetes.Interface
 	f.processingWg.Add(1)
 	go f.process()
 
-	lt := &leaseTracker{
-		logger:          logger,
-		selfIP:          selfIP,
-		bs:              bs,
-		kc:              kc,
-		endpointsLister: endpointsInformer.Lister(),
-		id2ip:           make(map[string]string),
-		accept:          accept,
-		fwd:             f,
-	}
-
-	leaseInformer := leaseinformer.Get(ctx)
-	leaseInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: lt.filterFunc(system.Namespace()),
-		Handler: cache.ResourceEventHandlerFuncs{
-			AddFunc:    lt.leaseUpdated,
-			UpdateFunc: controller.PassNew(lt.leaseUpdated),
-			// TODO(yanweiguo): Set up DeleteFunc.
-		},
-	})
-
 	return f
 }
 
-// leaseTracker monitors lease resources to update the Forwarder's processor configuration(s)
-// with the appropriate owner's stats endpoint.  When we own the lease, a localProcessor is
-// used, and when someone else owns the lease a remoteProcessor is used with their stats
-// endpoint.
-type leaseTracker struct {
-	logger *zap.SugaredLogger
-	selfIP string
-	// bs is the BucketSet including all Autoscaler buckets.
-	bs *hash.BucketSet
-
-	kc              kubernetes.Interface
-	endpointsLister corev1listers.EndpointsLister
-
-	// id2ip stores the IP extracted from the holder identity to avoid
-	// string split each time.
-	id2ip map[string]string
-
-	// `accept` is the function to process a StatMessage which doesn't need
-	// to be forwarded.
-	accept statProcessor
-
-	fwd *Forwarder
-}
-
-func (f *leaseTracker) extractIP(holder string) (string, error) {
-	if ip, ok := f.id2ip[holder]; ok {
-		return ip, nil
-	}
-
-	_, ip, err := bucket.ExtractPodNameAndIP(holder)
-	if err == nil {
-		f.id2ip[holder] = ip
-	}
-	return ip, err
-}
-
-func (f *leaseTracker) filterFunc(namespace string) func(interface{}) bool {
-	return func(obj interface{}) bool {
-		l := obj.(*coordinationv1.Lease)
-
-		if l.Namespace != namespace {
-			return false
-		}
-
-		if !f.bs.HasBucket(l.Name) {
-			// Not for Autoscaler.
-			return false
-		}
-
-		if l.Spec.HolderIdentity == nil || *l.Spec.HolderIdentity == "" {
-			// No holder yet.
-			return false
-		}
-
-		// The holder identity is in format of <POD-NAME>_<POD-IP>.
-		holder := *l.Spec.HolderIdentity
-		_, err := f.extractIP(holder)
-		if err != nil {
-			f.logger.Warn("Found invalid Lease holder identify ", holder)
-			return false
-		}
-		if p := f.fwd.getProcessor(l.Name); p != nil && p.is(holder) {
-			return false
-		}
-
-		return true
-	}
-}
-
-func (f *leaseTracker) leaseUpdated(obj interface{}) {
-	l := obj.(*coordinationv1.Lease)
-	ns, n := l.Namespace, l.Name
-	ctx := context.Background()
-
-	if l.Spec.HolderIdentity == nil {
-		// This shouldn't happen as we filter it out when queueing.
-		// Add a nil point check for safety.
-		f.logger.Warnf("Lease %s/%s with nil HolderIdentity got enqueued", ns, n)
-		return
-	}
-
-	// Close existing connection if there's one. The target address won't
-	// be the same as the IP has changed.
-	if p := f.fwd.getProcessor(n); p != nil {
-		p.shutdown()
-	}
-
-	holder := *l.Spec.HolderIdentity
-	// The map should have the IP because of the operations in the filter when queueing.
-	ip, ok := f.id2ip[holder]
-	if !ok {
-		// This shouldn't happen because we store the value into the map in the filter
-		// when queueing. Add a log here in case something unexpected happens.
-		f.logger.Warn("IP not found in cached map for ", holder)
-		return
-	}
-
-	if ip != f.selfIP {
-		f.fwd.setProcessor(n, newForwardProcessor(f.logger.With(zap.String("bucket", n)), n, holder,
-			fmt.Sprintf("ws://%s:%d", ip, autoscalerPort),
-			fmt.Sprintf("ws://%s.%s.%s", n, ns, svcURLSuffix)))
-
-		// Skip creating/updating Service and Endpoints if not the leader.
-		return
-	}
-
-	f.fwd.setProcessor(n, &localProcessor{
-		bkt:    n,
-		holder: holder,
-		logger: f.logger.With(zap.String("bucket", n)),
-		accept: f.accept,
-	})
-
-	if err := f.createService(ctx, ns, n); err != nil {
-		f.logger.Fatalf("Failed to create Service for Lease %s/%s: %v", ns, n, err)
-		return
-	}
-	f.logger.Infof("Created Service for Lease %s/%s", ns, n)
-
-	if err := f.createOrUpdateEndpoints(ctx, ns, n); err != nil {
-		f.logger.Fatalf("Failed to create Endpoints for Lease %s/%s: %v", ns, n, err)
-		return
-	}
-	f.logger.Infof("Created Endpoints for Lease %s/%s", ns, n)
-}
-
-func (f *leaseTracker) createService(ctx context.Context, ns, n string) error {
-	var lastErr error
-	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
-		_, lastErr = f.kc.CoreV1().Services(ns).Create(ctx, &corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      n,
-				Namespace: ns,
-			},
-			Spec: corev1.ServiceSpec{
-				Ports: []corev1.ServicePort{{
-					Name:       autoscalerPortName,
-					Port:       autoscalerPort,
-					TargetPort: intstr.FromInt(autoscalerPort),
-				}},
-			},
-		}, metav1.CreateOptions{})
-
-		if apierrs.IsAlreadyExists(lastErr) {
-			return true, nil
-		}
-
-		// Do not return the error to cause a retry.
-		return lastErr == nil, nil
-	}); err != nil {
-		return lastErr
-	}
-
-	return nil
-}
-
-// createOrUpdateEndpoints creates an Endpoints object with the given namespace and
-// name, and the Forwarder.selfIP. If the Endpoints object already
-// exists, it will update the Endpoints with the Forwarder.selfIP.
-func (f *leaseTracker) createOrUpdateEndpoints(ctx context.Context, ns, n string) error {
-	wantSubsets := []corev1.EndpointSubset{{
-		Addresses: []corev1.EndpointAddress{{
-			IP: f.selfIP,
-		}},
-		Ports: []corev1.EndpointPort{{
-			Name:     autoscalerPortName,
-			Port:     autoscalerPort,
-			Protocol: corev1.ProtocolTCP,
-		}}},
-	}
-
-	exists := true
-	var lastErr error
-	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
-		e, err := f.endpointsLister.Endpoints(ns).Get(n)
-		if apierrs.IsNotFound(err) {
-			exists = false
-			return true, nil
-		}
-
-		if err != nil {
-			lastErr = err
-			// Do not return the error to cause a retry.
-			return false, nil
-		}
-
-		if equality.Semantic.DeepEqual(wantSubsets, e.Subsets) {
-			return true, nil
-		}
-
-		want := e.DeepCopy()
-		want.Subsets = wantSubsets
-		if _, lastErr = f.kc.CoreV1().Endpoints(ns).Update(ctx, want, metav1.UpdateOptions{}); lastErr != nil {
-			// Do not return the error to cause a retry.
-			return false, nil
-		}
-
-		f.logger.Infof("Bucket Endpoints %s updated with IP %s", n, f.selfIP)
-		return true, nil
-	}); err != nil {
-		return lastErr
-	}
-
-	if exists {
-		return nil
-	}
-
-	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
-		_, lastErr = f.kc.CoreV1().Endpoints(ns).Create(ctx, &corev1.Endpoints{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      n,
-				Namespace: ns,
-			},
-			Subsets: wantSubsets,
-		}, metav1.CreateOptions{})
-		// Do not return the error to cause a retry.
-		return lastErr == nil, nil
-	}); err != nil {
-		return lastErr
-	}
-
-	return nil
-}
-
 func (f *Forwarder) getProcessor(bkt string) bucketProcessor {
 	f.processorsLock.RLock()
 	defer f.processorsLock.RUnlock()
diff --git a/pkg/autoscaler/statforwarder/leases.go b/pkg/autoscaler/statforwarder/leases.go
new file mode 100644
index 000000000..3a7e7df90
--- /dev/null
+++ b/pkg/autoscaler/statforwarder/leases.go
@@ -0,0 +1,298 @@
+/*
+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 statforwarder
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	coordinationv1 "k8s.io/api/coordination/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	leaseinformer "knative.dev/pkg/client/injection/kube/informers/coordination/v1/lease"
+	endpointsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/hash"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+	"knative.dev/serving/pkg/autoscaler/bucket"
+)
+
+// LeaseBasedProcessor tracks leases and decodes the holder's identity in order to set the
+// appropriate "processor" on the Forwarder.
+func LeaseBasedProcessor(ctx context.Context, f *Forwarder, accept statProcessor) error {
+	selfIP, err := bucket.PodIP()
+	if err != nil {
+		return err
+	}
+	endpointsInformer := endpointsinformer.Get(ctx)
+	lt := &leaseTracker{
+		logger:          logging.FromContext(ctx),
+		selfIP:          selfIP,
+		bs:              f.bs,
+		kc:              kubeclient.Get(ctx),
+		endpointsLister: endpointsInformer.Lister(),
+		id2ip:           make(map[string]string),
+		accept:          accept,
+		fwd:             f,
+	}
+
+	leaseInformer := leaseinformer.Get(ctx)
+	leaseInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: lt.filterFunc(system.Namespace()),
+		Handler: cache.ResourceEventHandlerFuncs{
+			AddFunc:    lt.leaseUpdated,
+			UpdateFunc: controller.PassNew(lt.leaseUpdated),
+			// TODO(yanweiguo): Set up DeleteFunc.
+		},
+	})
+
+	return nil
+}
+
+// leaseTracker monitors lease resources to update the Forwarder's processor configuration(s)
+// with the appropriate owner's stats endpoint.  When we own the lease, a localProcessor is
+// used, and when someone else owns the lease a remoteProcessor is used with their stats
+// endpoint.
+type leaseTracker struct {
+	logger *zap.SugaredLogger
+	selfIP string
+	// bs is the BucketSet including all Autoscaler buckets.
+	bs *hash.BucketSet
+
+	kc              kubernetes.Interface
+	endpointsLister corev1listers.EndpointsLister
+
+	// id2ip stores the IP extracted from the holder identity to avoid
+	// string split each time.
+	id2ip map[string]string
+
+	// `accept` is the function to process a StatMessage which doesn't need
+	// to be forwarded.
+	accept statProcessor
+
+	fwd *Forwarder
+}
+
+func (f *leaseTracker) extractIP(holder string) (string, error) {
+	if ip, ok := f.id2ip[holder]; ok {
+		return ip, nil
+	}
+
+	_, ip, err := bucket.ExtractPodNameAndIP(holder)
+	if err == nil {
+		f.id2ip[holder] = ip
+	}
+	return ip, err
+}
+
+func (f *leaseTracker) filterFunc(namespace string) func(interface{}) bool {
+	return func(obj interface{}) bool {
+		l := obj.(*coordinationv1.Lease)
+
+		if l.Namespace != namespace {
+			return false
+		}
+
+		if !f.bs.HasBucket(l.Name) {
+			// Not for Autoscaler.
+			return false
+		}
+
+		if l.Spec.HolderIdentity == nil || *l.Spec.HolderIdentity == "" {
+			// No holder yet.
+			return false
+		}
+
+		// The holder identity is in format of <POD-NAME>_<POD-IP>.
+		holder := *l.Spec.HolderIdentity
+		_, err := f.extractIP(holder)
+		if err != nil {
+			f.logger.Warn("Found invalid Lease holder identify ", holder)
+			return false
+		}
+		if p := f.fwd.getProcessor(l.Name); p != nil && p.is(holder) {
+			return false
+		}
+
+		return true
+	}
+}
+
+func (f *leaseTracker) leaseUpdated(obj interface{}) {
+	l := obj.(*coordinationv1.Lease)
+	ns, n := l.Namespace, l.Name
+	ctx := context.Background()
+
+	if l.Spec.HolderIdentity == nil {
+		// This shouldn't happen as we filter it out when queueing.
+		// Add a nil point check for safety.
+		f.logger.Warnf("Lease %s/%s with nil HolderIdentity got enqueued", ns, n)
+		return
+	}
+
+	// Close existing connection if there's one. The target address won't
+	// be the same as the IP has changed.
+	if p := f.fwd.getProcessor(n); p != nil {
+		p.shutdown()
+	}
+
+	holder := *l.Spec.HolderIdentity
+	// The map should have the IP because of the operations in the filter when queueing.
+	ip, ok := f.id2ip[holder]
+	if !ok {
+		// This shouldn't happen because we store the value into the map in the filter
+		// when queueing. Add a log here in case something unexpected happens.
+		f.logger.Warn("IP not found in cached map for ", holder)
+		return
+	}
+
+	if ip != f.selfIP {
+		f.fwd.setProcessor(n, newForwardProcessor(f.logger.With(zap.String("bucket", n)), n, holder,
+			fmt.Sprintf("ws://%s:%d", ip, autoscalerPort),
+			fmt.Sprintf("ws://%s.%s.%s", n, ns, svcURLSuffix)))
+
+		// Skip creating/updating Service and Endpoints if not the leader.
+		return
+	}
+
+	f.fwd.setProcessor(n, &localProcessor{
+		bkt:    n,
+		holder: holder,
+		logger: f.logger.With(zap.String("bucket", n)),
+		accept: f.accept,
+	})
+
+	if err := f.createService(ctx, ns, n); err != nil {
+		f.logger.Fatalf("Failed to create Service for Lease %s/%s: %v", ns, n, err)
+		return
+	}
+	f.logger.Infof("Created Service for Lease %s/%s", ns, n)
+
+	if err := f.createOrUpdateEndpoints(ctx, ns, n); err != nil {
+		f.logger.Fatalf("Failed to create Endpoints for Lease %s/%s: %v", ns, n, err)
+		return
+	}
+	f.logger.Infof("Created Endpoints for Lease %s/%s", ns, n)
+}
+
+func (f *leaseTracker) createService(ctx context.Context, ns, n string) error {
+	var lastErr error
+	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
+		_, lastErr = f.kc.CoreV1().Services(ns).Create(ctx, &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      n,
+				Namespace: ns,
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{{
+					Name:       autoscalerPortName,
+					Port:       autoscalerPort,
+					TargetPort: intstr.FromInt(autoscalerPort),
+				}},
+			},
+		}, metav1.CreateOptions{})
+
+		if apierrs.IsAlreadyExists(lastErr) {
+			return true, nil
+		}
+
+		// Do not return the error to cause a retry.
+		return lastErr == nil, nil
+	}); err != nil {
+		return lastErr
+	}
+
+	return nil
+}
+
+// createOrUpdateEndpoints creates an Endpoints object with the given namespace and
+// name, and the Forwarder.selfIP. If the Endpoints object already
+// exists, it will update the Endpoints with the Forwarder.selfIP.
+func (f *leaseTracker) createOrUpdateEndpoints(ctx context.Context, ns, n string) error {
+	wantSubsets := []corev1.EndpointSubset{{
+		Addresses: []corev1.EndpointAddress{{
+			IP: f.selfIP,
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     autoscalerPortName,
+			Port:     autoscalerPort,
+			Protocol: corev1.ProtocolTCP,
+		}}},
+	}
+
+	exists := true
+	var lastErr error
+	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
+		e, err := f.endpointsLister.Endpoints(ns).Get(n)
+		if apierrs.IsNotFound(err) {
+			exists = false
+			return true, nil
+		}
+
+		if err != nil {
+			lastErr = err
+			// Do not return the error to cause a retry.
+			return false, nil
+		}
+
+		if equality.Semantic.DeepEqual(wantSubsets, e.Subsets) {
+			return true, nil
+		}
+
+		want := e.DeepCopy()
+		want.Subsets = wantSubsets
+		if _, lastErr = f.kc.CoreV1().Endpoints(ns).Update(ctx, want, metav1.UpdateOptions{}); lastErr != nil {
+			// Do not return the error to cause a retry.
+			return false, nil
+		}
+
+		f.logger.Infof("Bucket Endpoints %s updated with IP %s", n, f.selfIP)
+		return true, nil
+	}); err != nil {
+		return lastErr
+	}
+
+	if exists {
+		return nil
+	}
+
+	if err := wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
+		_, lastErr = f.kc.CoreV1().Endpoints(ns).Create(ctx, &corev1.Endpoints{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      n,
+				Namespace: ns,
+			},
+			Subsets: wantSubsets,
+		}, metav1.CreateOptions{})
+		// Do not return the error to cause a retry.
+		return lastErr == nil, nil
+	}); err != nil {
+		return lastErr
+	}
+
+	return nil
+}
diff --git a/pkg/autoscaler/statforwarder/statefulset.go b/pkg/autoscaler/statforwarder/statefulset.go
new file mode 100644
index 000000000..e097edd93
--- /dev/null
+++ b/pkg/autoscaler/statforwarder/statefulset.go
@@ -0,0 +1,44 @@
+/*
+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 statforwarder
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"knative.dev/pkg/leaderelection"
+)
+
+// StatefulSetBasedProcessor configured "processors" for each of the statefulset ordinals.
+func StatefulSetBasedProcessor(ctx context.Context, f *Forwarder, accept statProcessor) error {
+	id, bs, err := leaderelection.NewStatefulSetBucketAndSet(len(f.bs.BucketList()))
+	if err != nil {
+		return err
+	}
+	for _, b := range bs.Buckets() {
+		if n := b.Name(); n == id.Name() {
+			f.setProcessor(n, &localProcessor{
+				bkt:    n,
+				logger: f.logger.With(zap.String("bucket", n)),
+				accept: accept,
+			})
+		} else {
+			f.setProcessor(n, newForwardProcessor(f.logger.With(zap.String("bucket", n)), n, "unused", n))
+		}
+	}
+	return nil
+}
diff --git a/pkg/reconciler/route/controller.go b/pkg/reconciler/route/controller.go
index 632e1d57e..a332f806f 100644
--- a/pkg/reconciler/route/controller.go
+++ b/pkg/reconciler/route/controller.go
@@ -93,6 +93,7 @@ func newControllerWithClock(
 		configStore.WatchConfigs(cmw)
 		return controller.Options{ConfigStore: configStore}
 	})
+	c.enqueueAfter = impl.EnqueueAfter
 
 	logger.Info("Setting up event handlers")
 	routeInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
diff --git a/pkg/reconciler/route/reconcile_resources.go b/pkg/reconciler/route/reconcile_resources.go
index e84c373b5..285441da5 100644
--- a/pkg/reconciler/route/reconcile_resources.go
+++ b/pkg/reconciler/route/reconcile_resources.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
@@ -77,22 +78,27 @@ func (c *Reconciler) reconcileIngress(
 			ingress.Annotations[networking.RolloutAnnotationKey])
 
 		// And recompute the rollout state.
-		effectiveRO := curRO.Step(prevRO, int(c.clock.Now().UnixNano()))
-
-		// Comparing and diffing isn't cheap so do it only if we're going
-		// to actually log the message.
-		// Those are well known types, cmp won't panic.
+		now := c.clock.Now().UnixNano()
+		effectiveRO, nextStepTime := curRO.Step(prevRO, now)
 		logger := logging.FromContext(ctx).Desugar()
-		if logger.Core().Enabled(zapcore.DebugLevel) && !cmp.Equal(prevRO, effectiveRO) {
-			logger.Debug("Rollout diff:(-was,+now)",
-				zap.String("diff", cmp.Diff(prevRO, effectiveRO)))
+		if nextStepTime > 0 {
+			nextStepTime -= now
+			c.enqueueAfter(r, time.Duration(nextStepTime))
 		}
 
 		// Now check if the ingress status changed from not ready to ready.
 		rtView := r.Status.GetCondition(v1.RouteConditionIngressReady)
 		if ingress.IsReady() && !rtView.IsTrue() {
 			logger.Debug("Observing Ingress not-ready to ready switch condition for rollout")
-			effectiveRO.ObserveReady(int(c.clock.Now().UnixNano()))
+			effectiveRO.ObserveReady(now)
+		}
+
+		// Comparing and diffing isn't cheap so do it only if we're going
+		// to actually log the message.
+		// Those are well known types, cmp won't panic.
+		if logger.Core().Enabled(zapcore.DebugLevel) && !cmp.Equal(prevRO, effectiveRO) {
+			logger.Debug("Rollout diff:(-was,+now)",
+				zap.String("diff", cmp.Diff(prevRO, effectiveRO)))
 		}
 
 		desired, err := resources.MakeIngressWithRollout(ctx, r, tc, effectiveRO,
diff --git a/pkg/reconciler/route/route.go b/pkg/reconciler/route/route.go
index b25a2aa84..def84686c 100644
--- a/pkg/reconciler/route/route.go
+++ b/pkg/reconciler/route/route.go
@@ -22,6 +22,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	kubelabels "k8s.io/apimachinery/pkg/labels"
@@ -70,7 +71,8 @@ type Reconciler struct {
 	certificateLister   networkinglisters.CertificateLister
 	tracker             tracker.Interface
 
-	clock system.Clock
+	clock        system.Clock
+	enqueueAfter func(interface{}, time.Duration)
 }
 
 // Check that our Reconciler implements routereconciler.Interface
diff --git a/pkg/reconciler/route/traffic/rollout.go b/pkg/reconciler/route/traffic/rollout.go
index 5a1432fd0..1ac95376b 100644
--- a/pkg/reconciler/route/traffic/rollout.go
+++ b/pkg/reconciler/route/traffic/rollout.go
@@ -68,15 +68,15 @@ type RolloutParams struct {
 	// StartTime is the Unix timestamp in ns by when (+/- reconcile precision)
 	// the Rollout has started.
 	// This is required to compute step time and deadline.
-	StartTime int `json:"starttime,omitempty"`
+	StartTime int64 `json:"starttime,omitempty"`
 
 	// NextStepTime is the Unix timestamp in ns when the next
 	// rollout step should performed.
-	NextStepTime int `json:"nextStepTime,omitempty"`
+	NextStepTime int64 `json:"nextStepTime,omitempty"`
 
 	// StepDuration is the number of nanoseconds between two successive steps
 	// of rollout.
-	StepDuration int `json:"stepDuration,omitempty"`
+	StepDuration int64 `json:"stepDuration,omitempty"`
 
 	// How much traffic to move in a single step.
 	StepSize int `json:"stepSize,omitempty"`
@@ -127,7 +127,7 @@ const durationSecs = 120.0
 // ObserveReady traverses the configs and the ones that are in rollout
 // but have not observed step time yet, will have it set, to
 // max(1, nowTS-cfg.StartTime).
-func (cur *Rollout) ObserveReady(nowTS int) {
+func (cur *Rollout) ObserveReady(nowTS int64) {
 	for i := range cur.Configurations {
 		c := &cur.Configurations[i]
 		if c.StepParams.StepDuration == 0 && c.StepParams.StartTime > 0 {
@@ -144,9 +144,11 @@ func (cur *Rollout) ObserveReady(nowTS int) {
 // At the end of the call the returned object will contain the
 // desired traffic shape.
 // Step will return cur if no previous state was available.
-func (cur *Rollout) Step(prev *Rollout, nowTS int) *Rollout {
+// Second return value is the Unix timestamp in ns of the closest
+// rollout action to take or 0, if no rollout is currently scheduled.
+func (cur *Rollout) Step(prev *Rollout, nowTS int64) (*Rollout, int64) {
 	if prev == nil || len(prev.Configurations) == 0 {
-		return cur
+		return cur, 0
 	}
 
 	// The algorithm below is simplest, but probably not the most performant.
@@ -162,12 +164,14 @@ func (cur *Rollout) Step(prev *Rollout, nowTS int) *Rollout {
 	}
 
 	var ret []ConfigurationRollout
+	returnTS := int64(math.MaxInt64)
 	for t, ccfgs := range currConfigs {
 		pcfgs, ok := prevConfigs[t]
 		// A new tag was added, so we have no previous state to roll from,
 		// thus just add it to the return, we'll rollout to 100% from the get go (and it is
 		// always 100%, since default tag is _always_ there).
 		// So just append to the return list.
+		// TODO(vagababov): perhaps just ignore and remove from the return?
 		if !ok {
 			ret = append(ret, *ccfgs[0])
 			continue
@@ -182,12 +186,17 @@ func (cur *Rollout) Step(prev *Rollout, nowTS int) *Rollout {
 				ret = append(ret, *ccfgs[i])
 				i++
 			case ccfgs[i].ConfigurationName == pcfgs[j].ConfigurationName:
-				// Config might have 0% traffic assigned, if it is a tag only route (i.e.
+				// Config might have >0% traffic assigned, unless it is a tag only route (i.e.
 				// receives no traffic via default tag). So just skip it from the rollout
 				// altogether.
 				switch p := ccfgs[i].Percent; {
 				case p > 1:
-					ret = append(ret, *stepConfig(ccfgs[i], pcfgs[j], nowTS))
+					sc := *stepConfig(ccfgs[i], pcfgs[j], nowTS)
+					ret = append(ret, sc)
+					// Keep the minimum value if it is not 0.
+					if nst := sc.StepParams.NextStepTime; nst > 0 && nst < returnTS {
+						returnTS = nst
+					}
 				case p == 1:
 					// Skip all the work if it's a common A/B scenario where the test config
 					// receives just 1% of traffic.
@@ -215,7 +224,11 @@ func (cur *Rollout) Step(prev *Rollout, nowTS int) *Rollout {
 	// We need to sort the rollout, since we have map iterations in between,
 	// which are random.
 	sortRollout(ro)
-	return ro
+	// If no active rollouts return 0.
+	if returnTS == math.MaxInt64 {
+		returnTS = 0
+	}
+	return ro, returnTS
 }
 
 // adjustPercentage updates the rollout with the new percentage values.
@@ -247,7 +260,7 @@ func adjustPercentage(goal int, cr *ConfigurationRollout) {
 
 // stepRevisions performs re-adjustment of percentages on the revisions
 // to rollout more traffic to the last one.
-func stepRevisions(goal *ConfigurationRollout, nowTS int) {
+func stepRevisions(goal *ConfigurationRollout, nowTS int64) {
 	// Not yet ready to adjust the steps or we're done
 	// (shouldn't really be here, but better be defensive).
 	if nowTS < goal.StepParams.NextStepTime || len(goal.Revisions) < 2 {
@@ -305,7 +318,7 @@ func stepRevisions(goal *ConfigurationRollout, nowTS int) {
 
 // stepConfig takes previous and goal configuration shapes and returns a new
 // config rollout, after computing the percetage allocations.
-func stepConfig(goal, prev *ConfigurationRollout, nowTS int) *ConfigurationRollout {
+func stepConfig(goal, prev *ConfigurationRollout, nowTS int64) *ConfigurationRollout {
 	pc := len(prev.Revisions)
 	ret := &ConfigurationRollout{
 		ConfigurationName: goal.ConfigurationName,
@@ -412,9 +425,9 @@ func (cur *ConfigurationRollout) computeProperties(nowTS, minStepSec, durationSe
 	// The time we sleep between the steps.
 	stepDuration := durationSecs / numSteps * float64(time.Second)
 
-	cur.StepParams.StepDuration = int(stepDuration)
+	cur.StepParams.StepDuration = int64(stepDuration)
 	cur.StepParams.StepSize = int(stepSize)
-	cur.StepParams.NextStepTime = int(nowTS + stepDuration)
+	cur.StepParams.NextStepTime = int64(nowTS + stepDuration)
 }
 
 // sortRollout sorts the rollout based on tag so it's consistent
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 12fff9c67..42c96a884 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -954,7 +954,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/caching v0.0.0-20201208013740-3ca9791186ed
+# knative.dev/caching v0.0.0-20201209181541-0e55b559628c
 ## explicit
 knative.dev/caching/config
 knative.dev/caching/pkg/apis/caching
@@ -978,7 +978,7 @@ knative.dev/caching/pkg/client/listers/caching/v1alpha1
 # knative.dev/hack v0.0.0-20201201234937-fddbf732e450
 ## explicit
 knative.dev/hack
-# knative.dev/networking v0.0.0-20201209014241-22a7e22628fc
+# knative.dev/networking v0.0.0-20201209181441-896c0e7c4864
 ## explicit
 knative.dev/networking/config
 knative.dev/networking/pkg


Test patch:
diff --git a/pkg/autoscaler/statforwarder/forwarder_test.go b/pkg/autoscaler/statforwarder/forwarder_test.go
index 41cb44daf..a27710eae 100644
--- a/pkg/autoscaler/statforwarder/forwarder_test.go
+++ b/pkg/autoscaler/statforwarder/forwarder_test.go
@@ -19,6 +19,7 @@ package statforwarder
 import (
 	"errors"
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -37,7 +38,6 @@ import (
 	fakeserviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/hash"
-	"knative.dev/pkg/logging"
 	rtesting "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/system"
 	asmetrics "knative.dev/serving/pkg/autoscaler/metrics"
@@ -80,9 +80,15 @@ var (
 	noOp = func(sm asmetrics.StatMessage) {}
 )
 
+func must(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+}
+
 func TestForwarderReconcile(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	endpoints := fakeendpointsinformer.Get(ctx)
 	service := fakeserviceinformer.Get(ctx)
@@ -93,8 +99,12 @@ func TestForwarderReconcile(t *testing.T) {
 		t.Fatal("Failed to start informers:", err)
 	}
 
-	f1 := New(ctx, logger, kubeClient, testIP1, testBs, noOp)
-	f2 := New(ctx, logger, kubeClient, testIP2, testBs, noOp)
+	os.Setenv("POD_IP", testIP1)
+	f1 := New(ctx, testBs)
+	must(t, LeaseBasedProcessor(ctx, f1, noOp))
+	os.Setenv("POD_IP", testIP2)
+	f2 := New(ctx, testBs)
+	must(t, LeaseBasedProcessor(ctx, f2, noOp))
 
 	defer func() {
 		f1.Cancel()
@@ -172,7 +182,6 @@ func TestForwarderReconcile(t *testing.T) {
 
 func TestForwarderRetryOnSvcCreationFailure(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	lease := fakeleaseinformer.Get(ctx)
 
@@ -186,7 +195,8 @@ func TestForwarderRetryOnSvcCreationFailure(t *testing.T) {
 		waitInformers()
 	}()
 
-	New(ctx, logger, kubeClient, testIP1, testBs, noOp)
+	os.Setenv("POD_IP", testIP1)
+	must(t, LeaseBasedProcessor(ctx, New(ctx, testBs), noOp))
 
 	svcCreation := 0
 	retried := make(chan struct{})
@@ -213,7 +223,6 @@ func TestForwarderRetryOnSvcCreationFailure(t *testing.T) {
 
 func TestForwarderRetryOnEndpointsCreationFailure(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	lease := fakeleaseinformer.Get(ctx)
 
@@ -227,7 +236,8 @@ func TestForwarderRetryOnEndpointsCreationFailure(t *testing.T) {
 		waitInformers()
 	}()
 
-	New(ctx, logger, kubeClient, testIP1, testBs, noOp)
+	os.Setenv("POD_IP", testIP1)
+	must(t, LeaseBasedProcessor(ctx, New(ctx, testBs), noOp))
 
 	endpointsCreation := 0
 	retried := make(chan struct{})
@@ -254,7 +264,6 @@ func TestForwarderRetryOnEndpointsCreationFailure(t *testing.T) {
 
 func TestForwarderRetryOnEndpointsUpdateFailure(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	endpoints := fakeendpointsinformer.Get(ctx)
 	lease := fakeleaseinformer.Get(ctx)
@@ -269,7 +278,8 @@ func TestForwarderRetryOnEndpointsUpdateFailure(t *testing.T) {
 		waitInformers()
 	}()
 
-	New(ctx, logger, kubeClient, testIP1, testBs, noOp)
+	os.Setenv("POD_IP", testIP1)
+	must(t, LeaseBasedProcessor(ctx, New(ctx, testBs), noOp))
 
 	endpointsUpdate := 0
 	retried := make(chan struct{})
@@ -304,7 +314,6 @@ func TestForwarderRetryOnEndpointsUpdateFailure(t *testing.T) {
 
 func TestForwarderSkipReconciling(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	lease := fakeleaseinformer.Get(ctx)
 
@@ -318,7 +327,8 @@ func TestForwarderSkipReconciling(t *testing.T) {
 		waitInformers()
 	}()
 
-	New(ctx, logger, kubeClient, testIP1, testBs, noOp)
+	os.Setenv("POD_IP", testIP1)
+	must(t, LeaseBasedProcessor(ctx, New(ctx, testBs), noOp))
 
 	svcCreated := make(chan struct{})
 	kubeClient.PrependReactor("create", "services",
@@ -392,7 +402,6 @@ func TestForwarderSkipReconciling(t *testing.T) {
 
 func TestProcess(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
-	logger := logging.FromContext(ctx)
 	kubeClient := fakekubeclient.Get(ctx)
 	lease := fakeleaseinformer.Get(ctx)
 
@@ -413,7 +422,9 @@ func TestProcess(t *testing.T) {
 		acceptCount++
 		acceptCh <- acceptCount
 	}
-	f := New(ctx, logger, kubeClient, testIP1, hash.NewBucketSet(sets.NewString(bucket1, bucket2)), accept)
+	os.Setenv("POD_IP", testIP1)
+	f := New(ctx, hash.NewBucketSet(sets.NewString(bucket1, bucket2)))
+	must(t, LeaseBasedProcessor(ctx, f, accept))
 
 	// A Forward without any leadership information should process with retry.
 	// Stat1 should be accepted and stat2 should be forwarded.
diff --git a/pkg/reconciler/route/table_test.go b/pkg/reconciler/route/table_test.go
index 2a935087f..4e2cdfecd 100644
--- a/pkg/reconciler/route/table_test.go
+++ b/pkg/reconciler/route/table_test.go
@@ -435,12 +435,12 @@ func TestReconcile(t *testing.T) {
 					RevisionName: "config-00001", Percent: 1,
 				}}, fakeCurTime.Add(-3*time.Second),
 					func(r *traffic.Rollout) {
-						numSteps := 120 / 3 // 40
+						const numSteps = 120 / 3 // 40
 						// Step duration is 3s (duration / numSteps = 120/40)
-						r.Configurations[0].StepParams.StepDuration = int(time.Second) * 120 / numSteps // 3s in ns.
-						r.Configurations[0].StepParams.StepSize = (100 - 1) / numSteps                  // 2
+						r.Configurations[0].StepParams.StepDuration = int64(time.Second) * 120 / numSteps // 3s in ns.
+						r.Configurations[0].StepParams.StepSize = (100 - 1) / numSteps                    // 2
 						// StepDuration is 3, and so next step is `now` + 3.
-						r.Configurations[0].StepParams.NextStepTime = int(fakeCurTime.Add(3 * time.Second).UnixNano())
+						r.Configurations[0].StepParams.NextStepTime = fakeCurTime.Add(3 * time.Second).UnixNano()
 					},
 				)),
 		}, {
@@ -2695,7 +2695,7 @@ func simpleRollout(cfg string, revs []traffic.RevisionRollout,
 			Configurations: []traffic.ConfigurationRollout{{
 				ConfigurationName: cfg,
 				StepParams: traffic.RolloutParams{
-					StartTime: int(now.UnixNano()),
+					StartTime: now.UnixNano(),
 				},
 				Percent:   100,
 				Revisions: revs,
diff --git a/pkg/reconciler/route/traffic/rollout_test.go b/pkg/reconciler/route/traffic/rollout_test.go
index f8cdc88e1..91d186924 100644
--- a/pkg/reconciler/route/traffic/rollout_test.go
+++ b/pkg/reconciler/route/traffic/rollout_test.go
@@ -35,6 +35,7 @@ func TestStep(t *testing.T) {
 	tests := []struct {
 		name            string
 		prev, cur, want *Rollout
+		wantNextStep    int64
 	}{{
 		name: "no prev",
 		cur:  &Rollout{},
@@ -116,12 +117,6 @@ func TestStep(t *testing.T) {
 					RevisionName: "let-it-bleed",
 					Percent:      100,
 				}},
-				StepParams: RolloutParams{
-					NextStepTime: 2009,
-					StepDuration: 2020,
-					StartTime:    2004,
-					StepSize:     14,
-				},
 			}},
 		},
 		prev: &Rollout{
@@ -155,13 +150,198 @@ func TestStep(t *testing.T) {
 					Percent:      8, // +3
 				}},
 				StepParams: RolloutParams{
-					NextStepTime: 2020 + 5, // now + duration.
+					NextStepTime: now + 5, // now + duration.
 					StepDuration: 5,
 					StartTime:    2004,
 					StepSize:     3,
 				},
 			}},
 		},
+		wantNextStep: now + 5,
+	}, {
+		name: "multiple configs step",
+		cur: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           42,
+				Revisions: []RevisionRollout{{
+					RevisionName: "let-it-bleed",
+					Percent:      42,
+				}},
+			}, {
+				ConfigurationName: "keith",
+				Percent:           52,
+				Revisions: []RevisionRollout{{
+					RevisionName: "beggars-banquet",
+					Percent:      52,
+				}},
+			}},
+			// 6% allocated to the revision direct, say.
+		},
+		prev: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           42,
+				StepParams: RolloutParams{
+					NextStepTime: 2019,
+					StepDuration: 43,
+					StartTime:    1961,
+					StepSize:     3,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "their-satanic-majesties-request",
+					Percent:      37,
+				}, {
+					RevisionName: "let-it-bleed",
+					Percent:      5,
+				}},
+			}, {
+				ConfigurationName: "keith",
+				Percent:           52,
+				StepParams: RolloutParams{
+					NextStepTime: 2018,
+					StepDuration: 55,
+					StartTime:    1971,
+					StepSize:     4,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "sticky-fingers",
+					Percent:      36,
+				}, {
+					RevisionName: "beggars-banquet",
+					Percent:      16,
+				}},
+			}},
+		},
+		want: &Rollout{
+			// Note: order will change, since we sort.
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "keith",
+				Percent:           52,
+				StepParams: RolloutParams{
+					NextStepTime: now + 55,
+					StepDuration: 55,
+					StartTime:    1971,
+					StepSize:     4,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "sticky-fingers",
+					Percent:      32,
+				}, {
+					RevisionName: "beggars-banquet",
+					Percent:      20,
+				}},
+			}, {
+				ConfigurationName: "mick",
+				Percent:           42,
+				Revisions: []RevisionRollout{{
+					RevisionName: "their-satanic-majesties-request",
+					Percent:      34, // -3
+				}, {
+					RevisionName: "let-it-bleed",
+					Percent:      8, // +3
+				}},
+				StepParams: RolloutParams{
+					NextStepTime: now + 43, // now + duration.
+					StepDuration: 43,
+					StartTime:    1961,
+					StepSize:     3,
+				},
+			}},
+		},
+		wantNextStep: now + 43, // the min of the two.
+	}, {
+		name: "multiple configs step and roll",
+		cur: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           42,
+				Revisions: []RevisionRollout{{
+					RevisionName: "let-it-bleed",
+					Percent:      42,
+				}},
+			}, {
+				ConfigurationName: "keith",
+				Percent:           52,
+				Revisions: []RevisionRollout{{
+					RevisionName: "tattoo-you",
+					Percent:      52,
+				}},
+			}},
+			// 6% allocated to the revision direct, say.
+		},
+		prev: &Rollout{
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "mick",
+				Percent:           42,
+				StepParams: RolloutParams{
+					NextStepTime: 2019,
+					StepDuration: 43,
+					StartTime:    1961,
+					StepSize:     3,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "their-satanic-majesties-request",
+					Percent:      37,
+				}, {
+					RevisionName: "let-it-bleed",
+					Percent:      5,
+				}},
+			}, {
+				ConfigurationName: "keith",
+				Percent:           52,
+				StepParams: RolloutParams{
+					NextStepTime: 2018,
+					StepDuration: 55,
+					StartTime:    1971,
+					StepSize:     4,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "sticky-fingers",
+					Percent:      36,
+				}, {
+					RevisionName: "beggars-banquet",
+					Percent:      16,
+				}},
+			}},
+		},
+		want: &Rollout{
+			// Note: order will change, since we sort.
+			Configurations: []ConfigurationRollout{{
+				ConfigurationName: "keith",
+				Percent:           52,
+				StepParams: RolloutParams{
+					StartTime: 2020,
+				},
+				Revisions: []RevisionRollout{{
+					RevisionName: "sticky-fingers",
+					Percent:      36,
+				}, {
+					RevisionName: "beggars-banquet",
+					Percent:      15,
+				}, {
+					RevisionName: "tattoo-you",
+					Percent:      1,
+				}},
+			}, {
+				ConfigurationName: "mick",
+				Percent:           42,
+				Revisions: []RevisionRollout{{
+					RevisionName: "their-satanic-majesties-request",
+					Percent:      34, // -3
+				}, {
+					RevisionName: "let-it-bleed",
+					Percent:      8, // +3
+				}},
+				StepParams: RolloutParams{
+					NextStepTime: now + 43, // now + duration.
+					StepDuration: 43,
+					StartTime:    1961,
+					StepSize:     3,
+				},
+			}},
+		},
+		wantNextStep: now + 43, // the min of the two.
 	}, {
 		name: "simplest, step, when stepsize=0",
 		cur: &Rollout{
@@ -185,10 +365,8 @@ func TestStep(t *testing.T) {
 				ConfigurationName: "mick",
 				Percent:           100,
 				StepParams: RolloutParams{
-					NextStepTime: 2019, // <- Those should be reset and/or updated.
-					StartTime:    2004,
-					StepDuration: 5,
-					StepSize:     0,
+					StartTime: 2004,
+					// The other fields should not be set yet.
 				},
 				Revisions: []RevisionRollout{{
 					RevisionName: "their-satanic-majesties-request",
@@ -211,10 +389,7 @@ func TestStep(t *testing.T) {
 					Percent:      5,
 				}},
 				StepParams: RolloutParams{
-					NextStepTime: 2019,
-					StartTime:    2004,
-					StepDuration: 5,
-					StepSize:     0,
+					StartTime: 2004,
 				},
 			}},
 		},
@@ -302,6 +477,7 @@ func TestStep(t *testing.T) {
 				}},
 				StepParams: RolloutParams{ // <- Those should be thrown out.
 					NextStepTime: 1984,
+					StartTime:    1981,
 					StepDuration: 1988,
 					StepSize:     11,
 				},
@@ -814,13 +990,16 @@ func TestStep(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			got := tc.cur.Step(tc.prev, now)
+			got, gotNS := tc.cur.Step(tc.prev, now)
 			if want := tc.want; !cmp.Equal(got, want, cmpopts.EquateEmpty()) {
 				t.Errorf("Wrong rolled rollout, diff(-want,+got):\n%s", cmp.Diff(want, got))
 			}
 			if !got.Validate() {
 				t.Errorf("Step returned an invalid config:\n%#v", got)
 			}
+			if got, want := gotNS, tc.wantNextStep; got != want {
+				t.Errorf("Incorrect NextStepTime = %d, want: %d", got, want)
+			}
 		})
 	}
 }
@@ -885,18 +1064,18 @@ func TestObserveReady(t *testing.T) {
 			Percent:           100,
 			StepParams: RolloutParams{
 				StartTime:    oldenDays,
-				StepDuration: 3 * int(time.Second),
+				StepDuration: int64(3 * time.Second),
 				StepSize:     100 / 40,
-				NextStepTime: now + 3*int(time.Second),
+				NextStepTime: now + 3*int64(time.Second),
 			},
 		}, {
 			ConfigurationName: "Percent not 100%",
 			Percent:           50,
 			StepParams: RolloutParams{
 				StartTime:    oldenDays,
-				StepDuration: 3 * int(time.Second),
+				StepDuration: int64(3 * time.Second),
 				StepSize:     50 / 40,
-				NextStepTime: now + 3*int(time.Second),
+				NextStepTime: now + 3*int64(time.Second),
 			},
 		}},
 	}
@@ -1197,7 +1376,7 @@ func TestJSONRoundtrip(t *testing.T) {
 func TestStepRevisions(t *testing.T) {
 	tests := []struct {
 		name string
-		now  int
+		now  int64
 		cfg  *ConfigurationRollout
 		want *ConfigurationRollout
 	}{{
diff --git a/test/test_images/flaky/README.md b/test/test_images/flaky/README.md
deleted file mode 100644
index b43b7dc9e..000000000
--- a/test/test_images/flaky/README.md
+++ /dev/null
@@ -1,16 +0,0 @@
-# Flaky test image
-
-The image contains a simple Go webserver, `main.go`, that will only succeed
-every Nth request. The value of N is specified in the PERIOD environment
-variable.
-
-## Trying out
-
-To run the image as a Service outside of the test suite:
-
-`ko apply -f service.yaml`
-
-## Building
-
-For details about building and adding new images, see the
-[section about test images](/test/README.md#test-images).
diff --git a/test/test_images/flaky/main.go b/test/test_images/flaky/main.go
deleted file mode 100644
index 263904438..000000000
--- a/test/test_images/flaky/main.go
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
-Copyright 2019 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 main
-
-import (
-	"fmt"
-	"log"
-	"net/http"
-	"os"
-	"strconv"
-
-	"go.uber.org/atomic"
-
-	network "knative.dev/networking/pkg"
-	"knative.dev/serving/test"
-)
-
-var (
-	period uint64
-	count  = atomic.NewUint64(0)
-)
-
-func handler(w http.ResponseWriter, r *http.Request) {
-	// Always succeed probes.
-	if network.IsKubeletProbe(r) {
-		w.WriteHeader(http.StatusOK)
-		return
-	}
-
-	// Increment the request count per non-probe request.
-	val := count.Inc()
-
-	if val%period > 0 {
-		w.WriteHeader(http.StatusInternalServerError)
-	}
-	w.Write([]byte(fmt.Sprint("count = ", val)))
-}
-
-func main() {
-	pstr, ok := os.LookupEnv("PERIOD")
-	if !ok {
-		log.Fatal("Must specify PERIOD environment variable.")
-	}
-	var err error
-	period, err = strconv.ParseUint(pstr, 10 /*base*/, 64 /*size*/)
-	if err != nil {
-		log.Fatal(`Error parsing "PERIOD" envvar as uint: `, err)
-	} else if period < 1 {
-		log.Fatal("Period must be positive, got: 0")
-	}
-	h := network.NewProbeHandler(http.HandlerFunc(handler))
-	test.ListenAndServeGracefully(":"+os.Getenv("PORT"), h.ServeHTTP)
-}
diff --git a/test/test_images/flaky/service.yaml b/test/test_images/flaky/service.yaml
deleted file mode 100644
index fcacf7cc1..000000000
--- a/test/test_images/flaky/service.yaml
+++ /dev/null
@@ -1,14 +0,0 @@
-apiVersion: serving.knative.dev/v1
-kind: Service
-metadata:
-  name: flaky-test-image
-  namespace: default
-spec:
-  template:
-    spec:
-      containers:
-      - image: ko://knative.dev/serving/test/test_images/flaky
-        env:
-        # Succeed every Nth request.
-        - name: PERIOD
-          value: "5"
diff --git a/third_party/kourier-latest/kourier.yaml b/third_party/kourier-latest/kourier.yaml
index 4401aa244..dad69317c 100644
--- a/third_party/kourier-latest/kourier.yaml
+++ b/third_party/kourier-latest/kourier.yaml
@@ -18,7 +18,7 @@ metadata:
   name: kourier-system
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 
 ---
 # Copyright 2020 The Knative Authors
@@ -42,7 +42,7 @@ metadata:
   namespace: knative-serving
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -51,7 +51,7 @@ metadata:
   namespace: knative-serving
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 rules:
   - apiGroups: [""]
     resources: ["events"]
@@ -78,7 +78,7 @@ metadata:
   name: 3scale-kourier
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
@@ -110,7 +110,7 @@ metadata:
   namespace: knative-serving
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 spec:
   replicas: 1
   selector:
@@ -122,7 +122,7 @@ spec:
         app: 3scale-kourier-control
     spec:
       containers:
-        - image: gcr.io/knative-nightly/knative.dev/net-kourier/cmd/kourier@sha256:b98121b90c318402093c36506d28063077f6e7d42e3e39b32ad15b93d40b30c6
+        - image: gcr.io/knative-nightly/knative.dev/net-kourier/cmd/kourier@sha256:b6c53a8ee52685888f797946ec48f30583b3244d77b9de0eba4fa3b3ae8518e2
           name: kourier-control
           env:
             - name: CERTS_SECRET_NAMESPACE
@@ -158,7 +158,7 @@ metadata:
   namespace: knative-serving
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 spec:
   ports:
     - name: grpc-xds
@@ -191,7 +191,7 @@ metadata:
   namespace: kourier-system
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 spec:
   selector:
     matchLabels:
@@ -257,7 +257,7 @@ metadata:
   namespace: kourier-system
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 spec:
   ports:
     - name: http2
@@ -279,7 +279,7 @@ metadata:
   namespace: kourier-system
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 spec:
   ports:
     - name: http2
@@ -297,7 +297,7 @@ metadata:
   namespace: kourier-system
   labels:
     networking.knative.dev/ingress-provider: kourier
-    serving.knative.dev/release: "v20201209-a876478c"
+    serving.knative.dev/release: "v20201210-7f90fe70"
 data:
   envoy-bootstrap.yaml: |
     dynamic_resources: