Implement endpoint steering for Kafka seed server traffic

.changes/unreleased/operator-Added-20260915-120000.yaml

@@ -1,4 +1,4 @@
 project: operator
 kind: Added
-body: 'The operator can now manage a Service''s EndpointSlices itself, publishing each port of the Service for a different subset of brokers — per-listener readiness within a single Kubernetes Service, which pod readiness cannot express. It gates the Schema Registry ports today: a broker whose registry store is still replaying `_schemas` leaves the Schema Registry ports, internal and external, while it keeps serving Kafka. Opt in per cluster by annotating a Cluster (V1) or Redpanda (V2) with `operator.redpanda.com/enable-endpoint-steering=true`, which renders the Service carrying that cluster''s Schema Registry listener for steering — the `<cluster>-cluster` Service on V1, the internal Service on V2. Any other Service opts in by carrying `cluster.redpanda.com/endpoints-for=<cluster name>` and no selector, which is how to steer a Service the operator does not render, such as a load balancer Service. Broker pods are matched through the `app.kubernetes.io/instance` label they already carry, so enabling it restarts nothing. StretchClusters are not covered: a multicluster deployment leaves its Services to the native EndpointSlice controller.'
+body: 'The operator can now manage a Service''s EndpointSlices itself, publishing each port of the Service for a different subset of brokers — per-listener readiness within a single Kubernetes Service, which pod readiness cannot express. It gates the Kafka and Schema Registry ports: a broker leaves a Kafka port while that listener refuses TCP connections, and leaves the Schema Registry ports, internal and external, while its registry store is still replaying `_schemas`, in each case while it keeps serving every other port. Opt in per cluster by annotating a Cluster (V1) or Redpanda (V2) with `operator.redpanda.com/enable-endpoint-steering=true`, which renders the Service carrying that cluster''s Schema Registry listener for steering — the `<cluster>-cluster` Service on V1, the internal Service on V2. Any other Service opts in by carrying `cluster.redpanda.com/endpoints-for=<cluster name>` and no selector, which is how to steer a Service the operator does not render, such as a load balancer Service. Broker pods are matched through the `app.kubernetes.io/instance` label they already carry, so enabling it restarts nothing. StretchClusters are not covered: a multicluster deployment leaves its Services to the native EndpointSlice controller.'
 time: 2026-09-15T12:00:00.000000000-04:00

acceptance/features/endpoint-steering.feature

@@ -1,15 +1,17 @@
 @cluster:steering @variant:vectorized
 Feature: Endpoint steering
   A cluster annotated for endpoint steering has the operator publish its
-  Service endpoints itself, port by port: every broker on the Kafka port, but
-  on the Schema Registry port only the brokers whose registry answers
-  GET /status/ready. Pod readiness never enters into it, so a broker keeps
-  serving Kafka while its registry is out of rotation.
+  Service endpoints itself, port by port: on the Kafka port the brokers whose
+  Kafka listener accepts a connection, and on the Schema Registry port the
+  brokers whose registry answers GET /status/ready. Pod readiness never
+  enters into it, so a broker keeps serving Kafka while its registry is out
+  of rotation.
 
   A pod that both Services take as a broker -- its labels say so -- and that
-  serves no Schema Registry stands in for a registry that is still replaying
-  _schemas, which nothing can trigger on demand. Taking a live broker's
-  registry away instead would mean shutting one of its ports, and the only
+  serves nothing stands in for a broker whose listeners are not up yet, and
+  so for a registry still replaying _schemas, which nothing can trigger on
+  demand. Taking a live broker's registry away instead would mean shutting
+  one of its ports, and the only
   lever for that in a cluster is a NetworkPolicy, which on k3d is enforced
   for traffic between nodes and not for traffic within one: whether it
   reached the operator's probes at all came down to where the operator
@@ -18,20 +20,21 @@ Feature: Endpoint steering
   So this scenario asserts the steady state, in a real cluster, on both the
   Service the operator renders and a Service managed alongside it -- for a V2
   Redpanda, and for a V1 Cluster as the vectorized variant.
-  TestSteersSchemaRegistryPort covers the transitions -- a registry that stops
-  answering losing its endpoint, and regaining it when it recovers -- against
-  a real API server, where a registry can be switched off on demand.
+  TestSteersProbedPorts covers the transitions -- a registry or Kafka listener
+  that stops answering losing its endpoint, and regaining it when it
+  recovers, independently of the other -- against a real API server, where
+  either can be switched off on demand.
 
   @skip:gke @skip:aks @skip:eks
-  Scenario: Schema Registry endpoints carry only brokers whose registry answers
+  Scenario: Kafka and Schema Registry endpoints carry only brokers whose listener answers
     Given cluster "steering" is available
     # A Service managed outside the operator opts in with the annotation and
     # no selector. Its port names differ from the cluster's own on purpose:
-    # steering keys on the cluster's Schema Registry port numbers.
+    # steering keys on the cluster's listener port numbers.
     #
     # It publishes not-ready addresses, as the cluster's own Services and the
-    # cloud seed load balancers do, which is what leaves the Schema Registry
-    # probe as the only thing that can move an endpoint here.
+    # cloud seed load balancers do, which is what leaves the probes as the
+    # only thing that can move an endpoint here.
     When I apply Kubernetes manifest:
     """
     ---
@@ -84,10 +87,10 @@ Feature: Endpoint steering
               memory: 64Mi
     """
     And Pod "steering-mute" will eventually be Running
-    Then the "kafka" listener of cluster "steering" should publish pods "steering-0, steering-1, steering-2, steering-mute"
-    And the "brokers" port of service "steering-clients" should publish pods "steering-0, steering-1, steering-2, steering-mute"
+    Then the "kafka" listener of cluster "steering" should publish pods "steering-0, steering-1, steering-2"
+    And the "brokers" port of service "steering-clients" should publish pods "steering-0, steering-1, steering-2"
     And the "schema registry" listener of cluster "steering" should publish pods "steering-0, steering-1, steering-2"
     And the "registry" port of service "steering-clients" should publish pods "steering-0, steering-1, steering-2"
-    # Readiness is untouched by any of this: the pod the Schema Registry port
-    # left out is a pod Kubernetes considers ready.
+    # Readiness is untouched by any of this: the pod both ports left out is a
+    # pod Kubernetes considers ready.
     And pod "steering-mute" should be ready

operator/internal/controller/endpointsteering/checker.go

@@ -15,6 +15,7 @@ import (
 	"io"
 	"net"
 	"slices"
+	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -56,22 +57,30 @@ const (
 	// probe. Two costs a recovered broker one resync, against a replay
 	// measured in minutes.
 	probeSuccessThreshold = 2
+
+	// kafkaDialTimeout bounds a Kafka port's TCP dial, at kubelet's
+	// tcpSocket default. The kernel completes the handshake whatever the
+	// broker is doing, so only a dropped SYN waits this out.
+	kafkaDialTimeout = time.Second
 )
 
 // checker decides whether to publish one pod on one Service port. It says
 // yes when the pod is one of the named cluster's brokers, Kubernetes would
-// publish it too, and -- on the Schema Registry port only -- its Schema
-// Registry answers GET /status/ready.
+// publish it too, and -- on a Schema Registry port -- its Schema Registry
+// answers GET /status/ready, or -- on a Kafka port -- that listener accepts
+// a TCP connection.
 type checker struct {
 	resolver     Resolver
 	probeTimeout time.Duration
+	dialTimeout  time.Duration
 	// podReady is Kubernetes' own rule, which every port is still held to.
 	// It reads the pod's conditions alone, so publishNotReadyAddresses is
 	// applied around it.
 	podReady portmapper.Checker
-	// schemaRegistry damps probeSchemaRegistry so one slow or dropped probe
-	// doesn't flap a healthy registry.
+	// schemaRegistry and kafka damp their probes so one slow or dropped
+	// probe doesn't flap a healthy listener.
 	schemaRegistry portmapper.Decider
+	kafka          portmapper.Decider
 
 	mu       sync.Mutex
 	clusters map[string]clusterEntry
@@ -91,17 +100,25 @@ func newChecker(resolver Resolver) *checker {
 	c := &checker{
 		resolver:     resolver,
 		probeTimeout: schemaregistry.ProbeTimeout,
+		dialTimeout:  kafkaDialTimeout,
 		podReady:     portmapper.PodReady(),
 		clusters:     map[string]clusterEntry{},
 		now:          time.Now,
 	}
-	// NB: Stable always returns a Decider -- that is how it damps an Abstain
-	// without breaking a streak -- and asserting it lets Decide call it
-	// directly.
-	c.schemaRegistry = portmapper.Stable(portmapper.DeciderFunc(c.probeSchemaRegistry), probeSuccessThreshold, probeFailureThreshold).(portmapper.Decider)
+	c.schemaRegistry = stable(c.probeSchemaRegistry)
+	c.kafka = stable(c.probeKafka)
 	return c
 }
 
+// stable damps probe with the checker's thresholds.
+//
+// NB: Stable always returns a Decider -- that is how it damps an Abstain
+// without breaking a streak -- and asserting it lets Decide call it
+// directly.
+func stable(probe portmapper.DeciderFunc) portmapper.Decider {
+	return portmapper.Stable(probe, probeSuccessThreshold, probeFailureThreshold).(portmapper.Decider)
+}
+
 // Decide implements [portmapper.Decider].
 func (c *checker) Decide(ctx context.Context, svc *corev1.Service, pod *corev1.Pod, port portmapper.Port) portmapper.Decision {
 	logger := log.FromContext(ctx).WithValues("pod", client.ObjectKeyFromObject(pod), "port", port.Name, "targetPort", port.Port)
@@ -129,12 +146,15 @@ func (c *checker) Decide(ctx context.Context, svc *corev1.Service, pod *corev1.P
 	if !svc.Spec.PublishNotReadyAddresses && !c.podReady.Check(ctx, svc, pod, port) {
 		return portmapper.Exclude
 	}
+	switch {
+	case isSchemaRegistryPort(port, brokers):
+		return c.schemaRegistry.Decide(ctx, svc, pod, port)
+	case isKafkaPort(port, brokers):
+		return c.kafka.Decide(ctx, svc, pod, port)
+	}
 	// Any other port is published just as Kubernetes would, so steering
 	// changes nothing for it.
-	if !isSchemaRegistryPort(port, brokers) {
-		return portmapper.Include
-	}
-	return c.schemaRegistry.Decide(ctx, svc, pod, port)
+	return portmapper.Include
 }
 
 // groupOf is the cluster a check is about: the Service's annotation, which
@@ -164,6 +184,12 @@ func isSchemaRegistryPort(port portmapper.Port, brokers Cluster) bool {
 		slices.Contains(brokers.SchemaRegistryPorts(), port.Port)
 }
 
+// isKafkaPort reports whether port is one a broker serves Kafka on, internal
+// or external.
+func isKafkaPort(port portmapper.Port, brokers Cluster) bool {
+	return port.Protocol == corev1.ProtocolTCP && slices.Contains(brokers.KafkaPorts(), port.Port)
+}
+
 // probeSchemaRegistry asks one broker's Schema Registry whether its store
 // has caught up, over a client scoped to this pod's address.
 //
@@ -199,6 +225,29 @@ func (c *checker) probeSchemaRegistry(ctx context.Context, svc *corev1.Service,
 	return portmapper.Include
 }
 
+// probeKafka dials one broker's Kafka listener on the port being decided.
+// Accepting a connection is all it asks; TLS, SASL and the protocol are left
+// to the client.
+func (c *checker) probeKafka(ctx context.Context, _ *corev1.Service, pod *corev1.Pod, port portmapper.Port) portmapper.Decision {
+	logger := log.FromContext(ctx).WithValues("checker", "KafkaAccepting", "pod", client.ObjectKeyFromObject(pod), "port", port.Name, "targetPort", port.Port)
+
+	host := port.Address
+	if host == "" {
+		host = pod.Status.PodIP
+	}
+	address := net.JoinHostPort(host, strconv.Itoa(int(port.Port)))
+
+	dialer := net.Dialer{Timeout: c.dialTimeout}
+	conn, err := dialer.DialContext(ctx, "tcp", address)
+	if err != nil {
+		decision := classifyProbeError(err)
+		logger.V(1).Info("kafka listener not accepting", "address", address, "error", err.Error(), "decision", decision.String())
+		return decision
+	}
+	_ = conn.Close()
+	return portmapper.Include
+}
+
 // classifyProbeError separates failures that describe the broker from
 // failures that describe the prober. A timeout (still replaying, or
 // unreachable), a refused or dropped connection, and a non-2xx answer all