feat: add option to disable ClusterIP routing via revision annotation

Adds the ability to optionally disable ClusterIP routing on a
per-revision basis
through the 'serving.knative.dev/disable-clusterip-routing' annotation.

When this annotation is set to 'true' on a revision, the activator will
bypass
ClusterIP routing and use direct pod IP routing exclusively, even when a
healthy
ClusterIP is available.

Key changes:
- Add DisableClusterIPRoutingAnnotationKey constant for the new
annotation
- Extend revisionThrottler to track and respect the disableClusterIP
setting
- Modify routing logic to skip ClusterIP when disabled
- Read annotation value during throttler creation from revision metadata
- Maintain full backward compatibility (ClusterIP routing enabled by
default)

This feature is useful for scenarios where direct pod routing is
preferred over
ClusterIP load balancing, such as for debugging, performance testing, or
when
specific routing behaviors are required.

Includes comprehensive tests to verify both disabled and default
behaviors.

Author: elijah-rou

example-disable-clusterip-routing.yaml

@@ -0,0 +1,20 @@
+apiVersion: serving.knative.dev/v1
+kind: Service
+metadata:
+  name: example-service
+spec:
+  template:
+    metadata:
+      annotations:
+        # Disable ClusterIP routing for this revision
+        # When set to "true", the activator will only use pod IP routing
+        serving.knative.dev/disable-clusterip-routing: "true"
+    spec:
+      containers:
+        - image: gcr.io/knative-samples/helloworld-go
+          ports:
+            - containerPort: 8080
+          env:
+            - name: TARGET
+              value: "Go Sample v1"
+

pkg/activator/net/throttler.go

@@ -132,6 +132,7 @@ type revisionThrottler struct {
 	revID                types.NamespacedName
 	containerConcurrency int
 	lbPolicy             lbPolicy
+	disableClusterIP     bool
 
 	// These are used in slicing to infer which pods to assign
 	// to this activator.
@@ -171,6 +172,7 @@ type revisionThrottler struct {
 func newRevisionThrottler(revID types.NamespacedName,
 	containerConcurrency int, proto string,
 	breakerParams queue.BreakerParams,
+	disableClusterIP bool,
 	logger *zap.SugaredLogger,
 ) *revisionThrottler {
 	logger = logger.With(zap.String(logkey.Key, revID.String()))
@@ -198,6 +200,7 @@ func newRevisionThrottler(revID types.NamespacedName,
 		logger:               logger,
 		protocol:             proto,
 		lbPolicy:             lbp,
+		disableClusterIP:     disableClusterIP,
 	}
 
 	// Start with unknown
@@ -414,10 +417,11 @@ func (rt *revisionThrottler) handleUpdate(update revisionDestsUpdate) {
 	rt.logger.Debugw("Handling update",
 		zap.String("ClusterIP", update.ClusterIPDest), zap.Object("dests", logging.StringSet(update.Dests)))
 
-	// ClusterIP is not yet ready, so we want to send requests directly to the pods.
+	// If ClusterIP routing is disabled OR ClusterIP is not yet ready,
+	// we want to send requests directly to the pods.
 	// NB: this will not be called in parallel, thus we can build a new podTrackers
 	// array before taking out a lock.
-	if update.ClusterIPDest == "" {
+	if rt.disableClusterIP || update.ClusterIPDest == "" {
 		// Create a map for fast lookup of existing trackers.
 		trackersMap := make(map[string]*podTracker, len(rt.podTrackers))
 		for _, tracker := range rt.podTrackers {
@@ -546,11 +550,16 @@ func (t *Throttler) getOrCreateRevisionThrottler(revID types.NamespacedName) (*r
 		if err != nil {
 			return nil, err
 		}
+		// Check if ClusterIP routing is disabled via annotation
+		disableClusterIP := rev.Annotations != nil &&
+			rev.Annotations[serving.DisableClusterIPRoutingAnnotationKey] == "true"
+
 		revThrottler = newRevisionThrottler(
 			revID,
 			int(rev.Spec.GetContainerConcurrency()),
 			pkgnet.ServicePortName(rev.GetProtocol()),
 			queue.BreakerParams{QueueDepth: breakerQueueDepth, MaxConcurrency: revisionMaxConcurrency},
+			disableClusterIP,
 			t.logger,
 		)
 		t.revisionThrottlers[revID] = revThrottler

pkg/activator/net/throttler_test.go

@@ -618,7 +618,7 @@ func TestPodAssignmentFinite(t *testing.T) {
 	defer cancel()
 
 	throttler := newTestThrottler(ctx)
-	rt := newRevisionThrottler(revName, 42 /*cc*/, pkgnet.ServicePortNameHTTP1, testBreakerParams, logger)
+	rt := newRevisionThrottler(revName, 42 /*cc*/, pkgnet.ServicePortNameHTTP1, testBreakerParams, false /*disableClusterIP*/, logger)
 	rt.numActivators.Store(4)
 	rt.activatorIndex.Store(0)
 	throttler.revisionThrottlers[revName] = rt
@@ -670,7 +670,7 @@ func TestPodAssignmentInfinite(t *testing.T) {
 	defer cancel()
 
 	throttler := newTestThrottler(ctx)
-	rt := newRevisionThrottler(revName, 0 /*cc*/, pkgnet.ServicePortNameHTTP1, testBreakerParams, logger)
+	rt := newRevisionThrottler(revName, 0 /*cc*/, pkgnet.ServicePortNameHTTP1, testBreakerParams, false /*disableClusterIP*/, logger)
 	throttler.revisionThrottlers[revName] = rt
 
 	update := revisionDestsUpdate{
@@ -902,12 +902,87 @@ func TestMultipleActivators(t *testing.T) {
 func TestInfiniteBreakerCreation(t *testing.T) {
 	// This test verifies that we use infiniteBreaker when CC==0.
 	tttl := newRevisionThrottler(types.NamespacedName{Namespace: "a", Name: "b"}, 0, /*cc*/
-		pkgnet.ServicePortNameHTTP1, queue.BreakerParams{}, TestLogger(t))
+		pkgnet.ServicePortNameHTTP1, queue.BreakerParams{}, false /*disableClusterIP*/, TestLogger(t))
 	if _, ok := tttl.breaker.(*infiniteBreaker); !ok {
 		t.Errorf("The type of revisionBreaker = %T, want %T", tttl, (*infiniteBreaker)(nil))
 	}
 }
 
+func TestDisableClusterIPRouting(t *testing.T) {
+	// Test that when disableClusterIP is true, we always use pod trackers even when ClusterIP is available
+	logger := TestLogger(t)
+	revName := types.NamespacedName{Namespace: testNamespace, Name: testRevision}
+
+	// Create a revision throttler with ClusterIP disabled
+	rt := newRevisionThrottler(revName, 42 /*cc*/, pkgnet.ServicePortNameHTTP1,
+		testBreakerParams, true /*disableClusterIP*/, logger)
+
+	// Update with both ClusterIP and pod destinations
+	update := revisionDestsUpdate{
+		Rev:           revName,
+		ClusterIPDest: "10.0.0.1:80", // ClusterIP is available
+		Dests:         sets.New("ip1", "ip2", "ip3"),
+	}
+
+	rt.handleUpdate(update)
+
+	// Verify that we're using pod trackers, not ClusterIP
+	rt.mux.RLock()
+	defer rt.mux.RUnlock()
+
+	if rt.clusterIPTracker != nil {
+		t.Error("Expected clusterIPTracker to be nil when ClusterIP routing is disabled")
+	}
+
+	if len(rt.podTrackers) != 3 {
+		t.Errorf("Expected 3 pod trackers, got %d", len(rt.podTrackers))
+	}
+
+	// Verify pod trackers have correct destinations
+	expectedDests := sets.New("ip1", "ip2", "ip3")
+	for _, tracker := range rt.podTrackers {
+		if !expectedDests.Has(tracker.dest) {
+			t.Errorf("Unexpected pod tracker destination: %s", tracker.dest)
+		}
+		expectedDests.Delete(tracker.dest)
+	}
+}
+
+func TestEnableClusterIPRoutingByDefault(t *testing.T) {
+	// Test that when disableClusterIP is false (default), we use ClusterIP when available
+	logger := TestLogger(t)
+	revName := types.NamespacedName{Namespace: testNamespace, Name: testRevision}
+
+	// Create a revision throttler with ClusterIP enabled (default)
+	rt := newRevisionThrottler(revName, 42 /*cc*/, pkgnet.ServicePortNameHTTP1,
+		testBreakerParams, false /*disableClusterIP*/, logger)
+
+	// Update with both ClusterIP and pod destinations
+	update := revisionDestsUpdate{
+		Rev:           revName,
+		ClusterIPDest: "10.0.0.1:80", // ClusterIP is available
+		Dests:         sets.New("ip1", "ip2", "ip3"),
+	}
+
+	rt.handleUpdate(update)
+
+	// Verify that we're using ClusterIP, not pod trackers
+	rt.mux.RLock()
+	defer rt.mux.RUnlock()
+
+	if rt.clusterIPTracker == nil {
+		t.Error("Expected clusterIPTracker to be set when ClusterIP routing is enabled")
+	}
+
+	if rt.clusterIPTracker.dest != "10.0.0.1:80" {
+		t.Errorf("Expected clusterIPTracker dest to be '10.0.0.1:80', got %s", rt.clusterIPTracker.dest)
+	}
+
+	if rt.podTrackers != nil {
+		t.Error("Expected podTrackers to be nil when using ClusterIP")
+	}
+}
+
 func (t *Throttler) try(ctx context.Context, requests int, try func(string) error) chan tryResult {
 	resultChan := make(chan tryResult)
 

pkg/apis/serving/register.go

@@ -144,6 +144,11 @@ const (
 
 	// ProgressDeadlineAnnotationKey is the label key for the per revision progress deadline to set for the deployment
 	ProgressDeadlineAnnotationKey = GroupName + "/progress-deadline"
+
+	// DisableClusterIPRoutingAnnotationKey is the annotation key to disable ClusterIP routing for a revision
+	// When set to "true", the activator will only use pod IP routing and not use ClusterIP for load balancing.
+	// By default, ClusterIP routing is enabled.
+	DisableClusterIPRoutingAnnotationKey = GroupName + "/disable-clusterip-routing"
 )
 
 var (