How Render Scaled Knative to 100k+ Web Apps

diff --git a/pkg/activator/net/helpers.go b/pkg/activator/net/helpers.go
index 3a8e5ef49..0e7d5c3d0 100644
--- a/pkg/activator/net/helpers.go
+++ b/pkg/activator/net/helpers.go
@@ -92,6 +92,47 @@ func endpointsToDests(endpoints *corev1.Endpoints, portName string) (ready, notR
 	return ready, notReady
 }

+// podsToDests returns the pod's IP as the ready set if the pod is Ready,
+// and as the notReady set if not.
+func podsToDests(pods []*corev1.Pod) (_ready, _notReady sets.String) {
+	ready := sets.String{}
+	notReady := sets.String{}
+	// 8012 is the port of the queue-proxy container that runs alongside
+	// the user-server container.
+	port := "8012"
+	for _, pod := range pods {
+		if podIsReady(pod) {
+			for _, ip := range pod.Status.PodIPs {
+				ready.Insert(net.JoinHostPort(ip.IP, port))
+			}
+		} else {
+			for _, ip := range pod.Status.PodIPs {
+				notReady.Insert(net.JoinHostPort(ip.IP, port))
+			}
+		}
+
+	}
+	return ready, notReady
+}
+
+func podIsReady(pod *corev1.Pod) bool {
+	// This is the same logic that the K8s endpoints controller uses to list
+	// the Pod's IP under the Endpoint's `Addresses` (as opposed to `NotReadyAddresses`).
+	//
+	// https://github.com/kubernetes/kubernetes/blob/8a3f72074e6390f8f14a89a7b78cef4379737216/pkg/controller/endpoint/endpoints_controller.go#L624
+	//
+	// If we "mistime" it by interpreting the pod as ready before it is indeed
+	// ready, then the request that triggers scale-from-zero will always return
+	// some connection error, as the activator will try to reach the user-server
+	// container before it can serve requests.
+	for _, cond := range pod.Status.Conditions {
+		if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 // getServicePort takes a service and a protocol and returns the port number of
 // the port named for that protocol. If the port is not found then ok is false.
 func getServicePort(protocol networking.ProtocolType, svc *corev1.Service) (int, bool) {

diff --git a/pkg/activator/net/revision_backends.go b/pkg/activator/net/revision_backends.go
index 3d1524925..94758eae1 100644
--- a/pkg/activator/net/revision_backends.go
+++ b/pkg/activator/net/revision_backends.go
@@ -31,6 +31,8 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
+	"k8s.io/apimachinery/pkg/labels"
+	podsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"

 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -47,6 +49,7 @@ import (
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/logging/logkey"
 	"knative.dev/pkg/reconciler"
+
 	"knative.dev/serving/pkg/apis/serving"
 	revisioninformer "knative.dev/serving/pkg/client/injection/informers/serving/v1/revision"
 	servinglisters "knative.dev/serving/pkg/client/listers/serving/v1"
@@ -441,6 +444,7 @@ type revisionBackendsManager struct {
 	ctx            context.Context
 	revisionLister servinglisters.RevisionLister
 	serviceLister  corev1listers.ServiceLister
+	podLister      corev1listers.PodLister

 	revisionWatchers    map[types.NamespacedName]*revisionWatcher
 	revisionWatchersMux sync.RWMutex
@@ -466,6 +470,7 @@ func newRevisionBackendsManagerWithProbeFrequency(ctx context.Context, tr http.R
 		ctx:              ctx,
 		revisionLister:   revisioninformer.Get(ctx).Lister(),
 		serviceLister:    serviceinformer.Get(ctx).Lister(),
+		podLister:        podsinformer.Get(ctx).Lister(),
 		revisionWatchers: make(map[types.NamespacedName]*revisionWatcher),
 		updateCh:         make(chan revisionDestsUpdate),
 		transport:        tr,
@@ -478,6 +483,7 @@ func newRevisionBackendsManagerWithProbeFrequency(ctx context.Context, tr http.R
 	endpointsInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: reconciler.ChainFilterFuncs(
 			reconciler.LabelExistsFilterFunc(serving.RevisionUID),
+			reconciler.Not(reconciler.LabelExistsFilterFunc(serving.RenderServicelessKey)),
 			// We are only interested in the private services, since that is
 			// what is populated by the actual revision backends.
 			reconciler.LabelFilterFunc(networking.ServiceTypeKey, string(networking.ServiceTypePrivate), false),
@@ -488,6 +494,18 @@ func newRevisionBackendsManagerWithProbeFrequency(ctx context.Context, tr http.R
 			DeleteFunc: rbm.endpointsDeleted,
 		},
 	})
+	podsInformer := podsinformer.Get(ctx)
+	podsInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: reconciler.ChainFilterFuncs(
+			reconciler.LabelExistsFilterFunc(serving.RevisionUID),
+			reconciler.LabelExistsFilterFunc(serving.RenderServicelessKey),
+		),
+		Handler: cache.ResourceEventHandlerFuncs{
+			AddFunc:    rbm.podUpdated,
+			UpdateFunc: controller.PassNew(rbm.podUpdated),
+			DeleteFunc: rbm.podDeleted,
+		},
+	})

 	go func() {
 		// updateCh can only be closed after revisionWatchers are done running
@@ -566,6 +584,40 @@ func (rbm *revisionBackendsManager) endpointsUpdated(newObj interface{}) {
 	}
 }

+// podUpdated is a handler function to be used by the Endpoints informer.
+// It updates the endpoints in the RevisionBackendsManager if the hosts changed
+func (rbm *revisionBackendsManager) podUpdated(newObj interface{}) {
+	// Ignore the updates when we've terminated.
+	select {
+	case <-rbm.ctx.Done():
+		return
+	default:
+	}
+	pod := newObj.(*corev1.Pod)
+	revID := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Labels[serving.RevisionLabelKey]}
+
+	rw, err := rbm.getOrCreateRevisionWatcher(revID)
+	if err != nil {
+		rbm.logger.Errorw("Failed to get revision watcher", zap.Error(err), zap.String(logkey.Key, revID.String()))
+		return
+	}
+	selector, _ := labels.Parse(fmt.Sprintf("%s=%s", serving.RevisionLabelKey, pod.Labels[serving.RevisionLabelKey]))
+	allPods, err := rbm.podLister.Pods(pod.Namespace).List(selector)
+	if err != nil {
+		rbm.logger.Errorw("Failed to list pods belonging to revision", zap.Error(err), zap.String(logkey.Key, revID.String()))
+		return
+	}
+	ready, notReady := podsToDests(allPods)
+	select {
+	case <-rbm.ctx.Done():
+		return
+	case rw.destsCh <- dests{ready: ready, notReady: notReady}:
+	}
+}
+
 // deleteRevisionWatcher deletes the revision watcher for rev if it exists. It expects
 // a write lock is held on revisionWatchersMux when calling.
 func (rbm *revisionBackendsManager) deleteRevisionWatcher(rev types.NamespacedName) {
@@ -590,3 +642,7 @@ func (rbm *revisionBackendsManager) endpointsDeleted(obj interface{}) {
 	defer rbm.revisionWatchersMux.Unlock()
 	rbm.deleteRevisionWatcher(revID)
 }
+
+func (rbm *revisionBackendsManager) podDeleted(obj interface{}) {
+	rbm.podUpdated(obj)
+}

diff --git a/pkg/apis/serving/register.go b/pkg/apis/serving/register.go
index 61a2e8b4a..c067f962b 100644
--- a/pkg/apis/serving/register.go
+++ b/pkg/apis/serving/register.go
@@ -25,6 +25,8 @@ const (
 	// GroupNamePrefix is the prefix for label key and annotation key
 	GroupNamePrefix = GroupName + "/"

+	RenderServicelessKey = "render.com/knative-serviceless"
+
 	// ConfigurationLabelKey is the label key attached to a Revision indicating by
 	// which Configuration it is created.
 	ConfigurationLabelKey = GroupName + "/configuration"