diff --git a/.changes/unreleased/operator-Added-20260915-120000.yaml b/.changes/unreleased/operator-Added-20260915-120000.yaml index 40cf891bd..6a214eec2 100644 --- a/.changes/unreleased/operator-Added-20260915-120000.yaml +++ b/.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` Service on V1, the internal Service on V2. Any other Service opts in by carrying `cluster.redpanda.com/endpoints-for=` 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` Service on V1, the internal Service on V2. Any other Service opts in by carrying `cluster.redpanda.com/endpoints-for=` 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 diff --git a/acceptance/features/endpoint-steering.feature b/acceptance/features/endpoint-steering.feature index 3309011bc..4423bfd6a 100644 --- a/acceptance/features/endpoint-steering.feature +++ b/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 diff --git a/operator/internal/controller/endpointsteering/checker.go b/operator/internal/controller/endpointsteering/checker.go index b5b9e4776..c585ea8c7 100644 --- a/operator/internal/controller/endpointsteering/checker.go +++ b/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 diff --git a/operator/internal/controller/endpointsteering/checker_test.go b/operator/internal/controller/endpointsteering/checker_test.go index dc4bb50c9..748fd70a9 100644 --- a/operator/internal/controller/endpointsteering/checker_test.go +++ b/operator/internal/controller/endpointsteering/checker_test.go @@ -41,7 +41,8 @@ import ( const ( testCluster = "rp" testNamespace = "ns" - kafkaPort = 9093 + // adminPort is a port no probe has a say over. + adminPort = 9644 ) // fakeCluster is the [Cluster] the in-package tests resolve to; the real @@ -54,11 +55,13 @@ type fakeCluster struct { // externalPorts are the registry's external listener ports, which the // probed internal one gates alongside its own. externalPorts []int32 + kafkaPorts []int32 } func (f *fakeCluster) PodSelector() map[string]string { return f.podSelector } func (f *fakeCluster) ClusterDomain() string { return f.clusterDomain } func (f *fakeCluster) SchemaRegistry() *schemaregistry.Listener { return f.schemaRegistry } +func (f *fakeCluster) KafkaPorts() []int32 { return f.kafkaPorts } // SchemaRegistryPorts mirrors the real implementation's invariant: the // probed listener's port is always in the set. @@ -99,6 +102,13 @@ func externalRegistry(schemaRegistry *schemaregistry.Listener, external ...int32 return brokers } +// kafkaBrokers is testBrokers serving Kafka on ports. +func kafkaBrokers(ports ...int32) Cluster { + brokers := testBrokers(nil).(*fakeCluster) + brokers.kafkaPorts = ports + return brokers +} + func brokerPod(ready bool) *corev1.Pod { status := corev1.ConditionFalse if ready { @@ -204,6 +214,32 @@ func closedPort(t *testing.T) int32 { return int32(port) } +// toggledListener listens on a loopback port, returning it and a switch that +// takes the listener down, so dials are refused, and back up on the same +// port. Nothing accepts: the kernel completes the handshake regardless. +func toggledListener(t *testing.T) (int32, func(up bool)) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + address := listener.Addr().String() + t.Cleanup(func() { + if listener != nil { + _ = listener.Close() + } + }) + return int32(listener.Addr().(*net.TCPAddr).Port), func(up bool) { + t.Helper() + switch { + case up && listener == nil: + listener, err = net.Listen("tcp", address) + require.NoError(t, err) + case !up && listener != nil: + require.NoError(t, listener.Close()) + listener = nil + } + } +} + func TestCheckerDecide(t *testing.T) { var hits atomic.Int32 readyPort := schemaRegistryServer(t, statusHandler(http.StatusOK, &hits)) @@ -215,6 +251,7 @@ func TestCheckerDecide(t *testing.T) { }) unhealthyPort := schemaRegistryServer(t, statusHandler(http.StatusServiceUnavailable, &hits)) downPort := closedPort(t) + kafkaUpPort, _ := toggledListener(t) podDNSName := testCluster + "-0." + testCluster + "." + testNamespace + ".svc.cluster.local" tlsPort, tlsPool := tlsSchemaRegistryServer(t, podDNSName, statusHandler(http.StatusOK, &hits)) @@ -243,7 +280,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{err: ErrUnknownCluster}, svc: testService(true), pod: brokerPod(true), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Exclude, }, { @@ -251,7 +288,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{err: errors.New("secret unavailable")}, svc: testService(true), pod: brokerPod(true), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Abstain, }, { @@ -261,7 +298,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{}, svc: testService(true), pod: brokerPod(true), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Abstain, }, { @@ -269,7 +306,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{brokers: testBrokers(nil)}, svc: testService(true), pod: consolePod, - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Exclude, }, { @@ -277,7 +314,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{brokers: testBrokers(nil)}, svc: testService(false), pod: brokerPod(false), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Exclude, }, { @@ -285,7 +322,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{brokers: testBrokers(nil)}, svc: testService(true), pod: brokerPod(false), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Include, }, { @@ -293,7 +330,7 @@ func TestCheckerDecide(t *testing.T) { resolver: &fakeResolver{brokers: testBrokers(plaintext(readyPort))}, svc: testService(false), pod: brokerPod(true), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Include, }, { @@ -414,9 +451,41 @@ func TestCheckerDecide(t *testing.T) { }})}, svc: testService(true), pod: brokerPod(true), - port: tcpPort("kafka", kafkaPort), + port: tcpPort("admin", adminPort), want: portmapper.Include, }, + { + name: "kafka listener accepting includes", + resolver: &fakeResolver{brokers: kafkaBrokers(kafkaUpPort)}, + svc: testService(true), + pod: brokerPod(true), + port: tcpPort("kafka", kafkaUpPort), + want: portmapper.Include, + }, + { + name: "kafka listener refusing excludes", + resolver: &fakeResolver{brokers: kafkaBrokers(downPort)}, + svc: testService(true), + pod: brokerPod(true), + port: tcpPort("kafka", downPort), + want: portmapper.Exclude, + }, + { + name: "each kafka listener is dialed on its own", + resolver: &fakeResolver{brokers: kafkaBrokers(kafkaUpPort, downPort)}, + svc: testService(true), + pod: brokerPod(true), + port: tcpPort("kafka-default", downPort), + want: portmapper.Exclude, + }, + { + name: "a refusing kafka listener does not affect other ports", + resolver: &fakeResolver{brokers: kafkaBrokers(downPort)}, + svc: testService(true), + pod: brokerPod(true), + port: tcpPort("admin", adminPort), + want: portmapper.Include, + }, } { t.Run(tc.name, func(t *testing.T) { hits.Store(0) @@ -427,46 +496,74 @@ func TestCheckerDecide(t *testing.T) { } } -// TestCheckerDampsSchemaRegistryFailures pins the damping in both -// directions: a published Schema Registry survives fewer than -// probeFailureThreshold consecutive failures, and an unpublished one needs -// probeSuccessThreshold consecutive successes to come back -- so a listener -// that blinks on and off neither drops out nor writes itself back in on -// every other probe. -func TestCheckerDampsSchemaRegistryFailures(t *testing.T) { - var code atomic.Int32 - code.Store(http.StatusOK) - port := schemaRegistryServer(t, func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(int(code.Load())) - }) - - checker := newTestChecker(&fakeResolver{brokers: testBrokers(&schemaregistry.Listener{Port: port})}) - svc, pod, registry := testService(true), brokerPod(true), tcpPort("registry", port) - decide := func() portmapper.Decision { - return checker.Decide(t.Context(), svc, pod, registry) - } - - require.Equal(t, portmapper.Include, decide(), "the first probe is taken as-is") - - code.Store(http.StatusServiceUnavailable) - for i := 1; i < probeFailureThreshold; i++ { - require.Equal(t, portmapper.Include, decide(), "failure %d of %d must not unpublish", i, probeFailureThreshold) - } - require.Equal(t, portmapper.Exclude, decide(), "failure %d unpublishes", probeFailureThreshold) - - code.Store(http.StatusOK) - for i := 1; i < probeSuccessThreshold; i++ { - require.Equal(t, portmapper.Exclude, decide(), "success %d of %d must not republish", i, probeSuccessThreshold) - } - require.Equal(t, portmapper.Include, decide(), "success %d republishes", probeSuccessThreshold) - - // A listener alternating between answering and not stays where it is: - // neither run of agreement is long enough to flip it. - for range 3 { - code.Store(http.StatusServiceUnavailable) - require.Equal(t, portmapper.Include, decide()) - code.Store(http.StatusOK) - require.Equal(t, portmapper.Include, decide()) +// TestCheckerDampsProbeFailures pins the damping in both directions: a +// published listener survives fewer than probeFailureThreshold consecutive +// failures, and an unpublished one needs probeSuccessThreshold consecutive +// successes to come back -- so a listener that blinks on and off neither +// drops out nor writes itself back in on every other probe. +func TestCheckerDampsProbeFailures(t *testing.T) { + for _, tc := range []struct { + name string + // serve starts a listener of the kind under test and returns the + // cluster serving it, its port, and a switch for its health. + serve func(t *testing.T) (Cluster, portmapper.Port, func(healthy bool)) + }{ + { + name: "schema registry", + serve: func(t *testing.T) (Cluster, portmapper.Port, func(bool)) { + var code atomic.Int32 + code.Store(http.StatusOK) + port := schemaRegistryServer(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(int(code.Load())) + }) + return testBrokers(&schemaregistry.Listener{Port: port}), tcpPort("registry", port), func(healthy bool) { + if healthy { + code.Store(http.StatusOK) + } else { + code.Store(http.StatusServiceUnavailable) + } + } + }, + }, + { + name: "kafka", + serve: func(t *testing.T) (Cluster, portmapper.Port, func(bool)) { + port, toggle := toggledListener(t) + return kafkaBrokers(port), tcpPort("kafka", port), toggle + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + brokers, port, setHealthy := tc.serve(t) + checker := newTestChecker(&fakeResolver{brokers: brokers}) + svc, pod := testService(true), brokerPod(true) + decide := func() portmapper.Decision { + return checker.Decide(t.Context(), svc, pod, port) + } + + require.Equal(t, portmapper.Include, decide(), "the first probe is taken as-is") + + setHealthy(false) + for i := 1; i < probeFailureThreshold; i++ { + require.Equal(t, portmapper.Include, decide(), "failure %d of %d must not unpublish", i, probeFailureThreshold) + } + require.Equal(t, portmapper.Exclude, decide(), "failure %d unpublishes", probeFailureThreshold) + + setHealthy(true) + for i := 1; i < probeSuccessThreshold; i++ { + require.Equal(t, portmapper.Exclude, decide(), "success %d of %d must not republish", i, probeSuccessThreshold) + } + require.Equal(t, portmapper.Include, decide(), "success %d republishes", probeSuccessThreshold) + + // A listener alternating between answering and not stays where + // it is: neither run of agreement is long enough to flip it. + for range 3 { + setHealthy(false) + require.Equal(t, portmapper.Include, decide()) + setHealthy(true) + require.Equal(t, portmapper.Include, decide()) + } + }) } } @@ -480,6 +577,7 @@ func TestClassifyProbeError(t *testing.T) { {name: "wrapped deadline exceeded", err: &url.Error{Op: "Get", Err: context.DeadlineExceeded}, want: portmapper.Exclude}, {name: "dial timeout", err: &net.OpError{Op: "dial", Err: &os.SyscallError{Syscall: "connect", Err: syscall.ETIMEDOUT}}, want: portmapper.Exclude}, {name: "connection refused is the listener being down", err: &url.Error{Op: "Get", Err: &net.OpError{Op: "dial", Err: &os.SyscallError{Syscall: "connect", Err: syscall.ECONNREFUSED}}}, want: portmapper.Exclude}, + {name: "refused dial", err: &net.OpError{Op: "dial", Err: &os.SyscallError{Syscall: "connect", Err: syscall.ECONNREFUSED}}, want: portmapper.Exclude}, {name: "connection reset", err: &net.OpError{Op: "read", Err: &os.SyscallError{Syscall: "read", Err: syscall.ECONNRESET}}, want: portmapper.Exclude}, {name: "server hung up", err: io.EOF, want: portmapper.Exclude}, {name: "truncated response", err: io.ErrUnexpectedEOF, want: portmapper.Exclude}, diff --git a/operator/internal/controller/endpointsteering/endpointsteering.go b/operator/internal/controller/endpointsteering/endpointsteering.go index 492ff9f19..a8093ac4e 100644 --- a/operator/internal/controller/endpointsteering/endpointsteering.go +++ b/operator/internal/controller/endpointsteering/endpointsteering.go @@ -22,6 +22,10 @@ // ports -- internal and external, one registry backs them all -- only for // the brokers that answer. // +// Kafka ports get the same treatment with a plain TCP dial, per listener: a +// broker whose Kafka listener refuses -- not yet bound after a restart, or +// already closed on the way down -- leaves that port while keeping the rest. +// // A Service opts in by naming its cluster in [ServiceAnnotation] and // declaring no selector. Brokers are matched by the // app.kubernetes.io/instance label they already carry, so turning steering diff --git a/operator/internal/controller/endpointsteering/integration_test.go b/operator/internal/controller/endpointsteering/integration_test.go index 1b682971d..95329457f 100644 --- a/operator/internal/controller/endpointsteering/integration_test.go +++ b/operator/internal/controller/endpointsteering/integration_test.go @@ -44,30 +44,26 @@ import ( "github.com/redpanda-data/redpanda-operator/pkg/multicluster" ) -// kafkaTestPort is the Kafka port the steered Services under test carry; it -// exists only to be a port the Schema Registry probe has no say over. -const ( - kafkaTestPort = 9092 - // externalRegistryTestPort stands in for the registry's external - // listener. Nothing listens there: the point is that the internal - // listener's probe decides it. - externalRegistryTestPort = 18081 -) - -// TestSteersSchemaRegistryPort drives the controller against a real API -// server, for both cluster kinds at once as the operator runs them: broker -// pods, a steered Service each, and one stand-in Schema Registry the test can -// stop answering. It covers what no unit test can -- that the decisions reach -// the published EndpointSlices, per port, through the same factory and -// resolvers the operator resolves clusters with -- and it covers the -// transitions in both directions, which the acceptance scenarios cannot, +// externalRegistryTestPort stands in for the registry's external listener. +// Nothing listens there: the point is that the internal listener's probe +// decides it. +const externalRegistryTestPort = 18081 + +// TestSteersProbedPorts drives the controller against a real API server, for +// both cluster kinds at once as the operator runs them: broker pods, a +// steered Service each, and one stand-in Schema Registry and Kafka listener +// the test can take down. It covers what no unit test can -- that the +// decisions reach the published EndpointSlices, per port, through the same +// factory and resolvers the operator resolves clusters with -- and it covers +// the transitions in both directions, which the acceptance scenarios cannot, // since nothing can put a real broker's registry back into replay on demand. -func TestSteersSchemaRegistryPort(t *testing.T) { +func TestSteersProbedPorts(t *testing.T) { // The published addresses have to be real pod addresses as far as the API // server is concerned -- it rejects loopback in an EndpointSlice -- and - // the probes then dial exactly those, so the stand-in registry has to - // listen on an address this host actually answers on. - listener, err := net.Listen("tcp", net.JoinHostPort(routableHostAddress(t), "0")) + // the probes then dial exactly those, so the stand-ins have to listen on + // an address this host actually answers on. + host := routableHostAddress(t) + listener, err := net.Listen("tcp", net.JoinHostPort(host, "0")) require.NoError(t, err) var serving atomic.Bool @@ -84,6 +80,10 @@ func TestSteersSchemaRegistryPort(t *testing.T) { t.Cleanup(registry.Close) address := registry.Listener.Addr().(*net.TCPAddr) + kafka := serveKafka(t, net.JoinHostPort(host, "0")) + kafkaAddress := kafka.Addr().String() + kafkaPort := int32(kafka.Addr().(*net.TCPAddr).Port) + ctl := kubetest.NewEnv(t, kube.Options{Options: client.Options{Scheme: controller.UnifiedScheme}}) require.NoError(t, kube.ApplyAllAndWait(t.Context(), ctl, func(crd *apiextensionsv1.CustomResourceDefinition, err error) (bool, error) { if err != nil { @@ -103,8 +103,8 @@ func TestSteersSchemaRegistryPort(t *testing.T) { require.NoError(t, err) clusters := []steeredCluster{ - applyV1Cluster(t, ctl, namespace.Name, "rp-v1", address), - applyV2Cluster(t, ctl, namespace.Name, "rp-v2", address), + applyV1Cluster(t, ctl, namespace.Name, "rp-v1", address, kafkaPort), + applyV2Cluster(t, ctl, namespace.Name, "rp-v2", address, kafkaPort), } // Both managers bind ephemeral ports. kubetest.RunManager would do for @@ -193,6 +193,36 @@ func TestSteersSchemaRegistryPort(t *testing.T) { serving.Store(true) requirePublished("schema-registry", allBrokers) requirePublished("schema-registry-external", allBrokers) + + // A Kafka listener that refuses leaves the Kafka port, and only that. + require.NoError(t, kafka.Close()) + requirePublished("kafka", none) + requirePublished("schema-registry", allBrokers) + + serveKafka(t, kafkaAddress) + requirePublished("kafka", allBrokers) +} + +// serveKafka stands in for a broker's Kafka listener at address, which only +// has to accept connections. Each is closed straight away: left in the +// accept queue, the probes would fill the listen backlog -- 128 on macOS -- +// and every dial after that would time out. +func serveKafka(t *testing.T, address string) net.Listener { + t.Helper() + + listener, err := net.Listen("tcp", address) + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + _ = conn.Close() + } + }() + return listener } // steeredCluster is one cluster under test: the Service the operator steers @@ -204,8 +234,9 @@ type steeredCluster struct { } // applyV1Cluster creates a v1 Cluster, its broker pods, and a Service steered -// for it, with its Schema Registry listener pointed at address. -func applyV1Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address *net.TCPAddr) steeredCluster { +// for it, with its Schema Registry listener pointed at address and its Kafka +// listener at kafkaPort on the same host. +func applyV1Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address *net.TCPAddr, kafkaPort int32) steeredCluster { t.Helper() _, err := kube.Create(t.Context(), ctl, vectorizedv1alpha1.Cluster{ @@ -216,7 +247,7 @@ func applyV1Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address Replicas: ptr.To(int32(3)), Configuration: vectorizedv1alpha1.RedpandaConfig{ RPCServer: vectorizedv1alpha1.SocketAddress{Port: 33145}, - KafkaAPI: []vectorizedv1alpha1.KafkaAPI{{Port: kafkaTestPort}}, + KafkaAPI: []vectorizedv1alpha1.KafkaAPI{{Port: int(kafkaPort)}}, AdminAPI: []vectorizedv1alpha1.AdminAPI{{Port: 9644}}, SchemaRegistry: &vectorizedv1alpha1.SchemaRegistryAPI{Port: address.Port}, SchemaRegistryAPI: []vectorizedv1alpha1.SchemaRegistryAPI{{ @@ -235,13 +266,13 @@ func applyV1Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address return steeredCluster{ name: "v1 Cluster", - service: applySteeredService(t, ctl, namespace, name, address), + service: applySteeredService(t, ctl, namespace, name, address, kafkaPort), pods: applyBrokerPods(t, ctl, namespace, name, podLabels, address), } } // applyV2Cluster is applyV1Cluster for a v2 Redpanda. -func applyV2Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address *net.TCPAddr) steeredCluster { +func applyV2Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address *net.TCPAddr, kafkaPort int32) steeredCluster { t.Helper() _, err := kube.Create(t.Context(), ctl, redpandav1alpha2.Redpanda{ @@ -252,7 +283,7 @@ func applyV2Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address // stand-in. TLS: &redpandav1alpha2.TLS{Enabled: ptr.To(false)}, Listeners: &redpandav1alpha2.Listeners{ - Kafka: &redpandav1alpha2.Kafka{Listener: redpandav1alpha2.Listener{Port: ptr.To(int32(kafkaTestPort))}}, + Kafka: &redpandav1alpha2.Kafka{Listener: redpandav1alpha2.Listener{Port: ptr.To(kafkaPort)}}, SchemaRegistry: &redpandav1alpha2.SchemaRegistry{ Listener: redpandav1alpha2.Listener{Port: ptr.To(int32(address.Port))}, External: map[string]*redpandav1alpha2.ExternalListener{ @@ -267,7 +298,7 @@ func applyV2Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address return steeredCluster{ name: "v2 Redpanda", - service: applySteeredService(t, ctl, namespace, name, address), + service: applySteeredService(t, ctl, namespace, name, address, kafkaPort), pods: applyBrokerPods(t, ctl, namespace, name, map[string]string{ "app.kubernetes.io/instance": name, "app.kubernetes.io/name": "redpanda", @@ -278,7 +309,7 @@ func applyV2Cluster(t *testing.T, ctl *kube.Ctl, namespace, name string, address // applySteeredService creates a Service shaped as the operator renders the // clusters' own: annotated for the cluster, no selector, and publishing // not-ready addresses because Raft needs them. -func applySteeredService(t *testing.T, ctl *kube.Ctl, namespace, cluster string, address *net.TCPAddr) string { +func applySteeredService(t *testing.T, ctl *kube.Ctl, namespace, cluster string, address *net.TCPAddr, kafkaPort int32) string { t.Helper() name := cluster + "-steered" @@ -291,7 +322,7 @@ func applySteeredService(t *testing.T, ctl *kube.Ctl, namespace, cluster string, Spec: corev1.ServiceSpec{ PublishNotReadyAddresses: true, Ports: []corev1.ServicePort{ - {Name: "kafka", Port: kafkaTestPort, TargetPort: intstr.FromInt32(kafkaTestPort), Protocol: corev1.ProtocolTCP}, + {Name: "kafka", Port: kafkaPort, TargetPort: intstr.FromInt32(kafkaPort), Protocol: corev1.ProtocolTCP}, {Name: "schema-registry", Port: int32(address.Port), TargetPort: intstr.FromInt32(int32(address.Port)), Protocol: corev1.ProtocolTCP}, // The external listener, exposed the way a node-ported or // load-balanced Service exposes it: a different Service port, @@ -306,7 +337,7 @@ func applySteeredService(t *testing.T, ctl *kube.Ctl, namespace, cluster string, } // applyBrokerPods creates three pods that a cluster's Services will take as -// brokers, every one of them answering at address. +// brokers, every one of them answering on address's host. func applyBrokerPods(t *testing.T, ctl *kube.Ctl, namespace, cluster string, podLabels map[string]string, address *net.TCPAddr) []string { t.Helper() @@ -328,8 +359,8 @@ func applyBrokerPods(t *testing.T, ctl *kube.Ctl, namespace, cluster string, pod }) require.NoError(t, err) - // The registry stand-in listens on one address, so that is where - // every pod's probes have to be pointed. + // The stand-ins listen on one host, so that is where every pod's + // probes have to be pointed. pod.Status = corev1.PodStatus{ Phase: corev1.PodRunning, PodIP: address.IP.String(), diff --git a/operator/internal/controller/endpointsteering/resolver.go b/operator/internal/controller/endpointsteering/resolver.go index c8e21717f..def81dca3 100644 --- a/operator/internal/controller/endpointsteering/resolver.go +++ b/operator/internal/controller/endpointsteering/resolver.go @@ -30,7 +30,7 @@ var ErrUnknownCluster = errors.New("no cluster found for pod group") // Cluster is what a membership decision needs to know about a Redpanda // cluster: how to tell its broker pods from everything else in the -// namespace, and the Schema Registry listener they serve. +// namespace, and the Kafka and Schema Registry listeners they serve. // // NB: client.Factory.ClusterBrokers produces it, but is named here as an // interface rather than imported: pkg/client sits above the v1 and v2 @@ -49,6 +49,10 @@ type Cluster interface { // internal and external alike. One registry backs them all, so the one // probe decides the lot. SchemaRegistryPorts() []int32 + // KafkaPorts is every port a broker serves Kafka on, internal and + // external alike. Each is a listener of its own, so each is dialed on + // its own. + KafkaPorts() []int32 } // Resolver maps a pod group -- the cluster named by [ServiceAnnotation] on a diff --git a/operator/pkg/client/cluster_brokers.go b/operator/pkg/client/cluster_brokers.go index ccbfa6ebc..62f39a597 100644 --- a/operator/pkg/client/cluster_brokers.go +++ b/operator/pkg/client/cluster_brokers.go @@ -26,8 +26,8 @@ import ( ) // ClusterBrokers describes a cluster's broker pods from the outside: how to -// tell them from everything else in the namespace, and the Schema Registry -// listener they serve. +// tell them from everything else in the namespace, and the Kafka and Schema +// Registry listeners they serve. // // NB: accessors rather than fields, so the endpoint steering controller can // name the shape it needs as an interface instead of importing this package. @@ -38,6 +38,7 @@ type ClusterBrokers struct { clusterDomain string schemaRegistry *schemaregistry.Listener schemaRegistryPorts []int32 + kafkaPorts []int32 } // PodSelector is the cluster's own Service selector, carried by every one of @@ -69,6 +70,12 @@ func (c *ClusterBrokers) SchemaRegistryPorts() []int32 { return c.schemaRegistryPorts } +// KafkaPorts is every port a broker serves Kafka on, internal and external +// alike. +func (c *ClusterBrokers) KafkaPorts() []int32 { + return c.kafkaPorts +} + // ClusterBrokers returns the ClusterBrokers of a v1 Cluster or v2 Redpanda. // A StretchCluster has none: endpoint steering, its only caller, leaves // those Services to Kubernetes. @@ -91,21 +98,14 @@ func (c *Factory) redpandaClusterBrokers(ctx context.Context, cluster *redpandav brokers := &ClusterBrokers{ podSelector: redpandachart.ClusterPodLabelsSelector(state), clusterDomain: strings.TrimSuffix(state.Values.ClusterDomain, "."), + kafkaPorts: listenerPorts(state.Values.Listeners.Kafka), } listener := state.Values.Listeners.SchemaRegistry if !listener.Enabled { return brokers, nil } - brokers.schemaRegistryPorts = []int32{listener.Port} - for name := range listener.External { - // IsEnabled is the predicate the broker's own schema_registry_api - // list is built from, so this is exactly what a broker binds. - if external := listener.External[name]; external.IsEnabled() { - brokers.schemaRegistryPorts = append(brokers.schemaRegistryPorts, external.Port) - } - } - slices.Sort(brokers.schemaRegistryPorts) + brokers.schemaRegistryPorts = listenerPorts(listener) brokers.schemaRegistry = &schemaregistry.Listener{Port: listener.Port} if listener.TLS.IsEnabled(&state.Values.TLS) { brokers.schemaRegistry.TLSConfig = memoizeTLS(func(context.Context) (*tls.Config, error) { @@ -115,6 +115,20 @@ func (c *Factory) redpandaClusterBrokers(ctx context.Context, cluster *redpandav return brokers, nil } +// listenerPorts is every port a broker binds for listener. IsEnabled is the +// predicate the broker's own listener list is built from, so this is exactly +// what a broker binds. +func listenerPorts[T ~string](listener redpandachart.ListenerConfig[T]) []int32 { + ports := []int32{listener.Port} + for name := range listener.External { + if external := listener.External[name]; external.IsEnabled() { + ports = append(ports, external.Port) + } + } + slices.Sort(ports) + return ports +} + // memoizeTLS caches a successful TLS build, so a listener's certificates are // read once however many brokers and ports are probed against them. A // failure is not cached, so a certificate that shows up late is picked up on @@ -158,6 +172,7 @@ func (c *Factory) redpandaRenderState(ctx context.Context, cluster *redpandav1al func (c *Factory) v1ClusterBrokers(_ context.Context, cluster *vectorizedv1alpha1.Cluster, clusterName string) (*ClusterBrokers, error) { brokers := &ClusterBrokers{ podSelector: labels.ForCluster(cluster).AsAPISelector().MatchLabels, + kafkaPorts: v1KafkaPorts(cluster), } brokers.schemaRegistryPorts = v1SchemaRegistryPorts(cluster) @@ -185,6 +200,25 @@ func (c *Factory) v1ClusterBrokers(_ context.Context, cluster *vectorizedv1alpha return brokers, nil } +// v1KafkaPorts is the container port of every Kafka listener a v1 cluster +// declares. An external listener naming no port binds the internal one's +// plus one, as the broker's configuration renders it. +func v1KafkaPorts(cluster *vectorizedv1alpha1.Cluster) []int32 { + internal := cluster.InternalListener() + if internal == nil { + return nil + } + ports := []int32{int32(internal.Port)} + for _, external := range cluster.KafkaAPIExternalListeners() { + port := external.Port + if port == 0 { + port = internal.Port + 1 + } + ports = append(ports, int32(port)) + } + return ports +} + // v1SchemaRegistryPorts is the container port of every Schema Registry // listener a v1 cluster declares, over both the current field and the one it // replaced. diff --git a/operator/pkg/client/cluster_brokers_test.go b/operator/pkg/client/cluster_brokers_test.go new file mode 100644 index 000000000..cf67108f7 --- /dev/null +++ b/operator/pkg/client/cluster_brokers_test.go @@ -0,0 +1,82 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.md +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0 + +package client + +import ( + "testing" + + "github.com/stretchr/testify/require" + "k8s.io/utils/ptr" + + redpandachart "github.com/redpanda-data/redpanda-operator/charts/redpanda/v25/chart" + vectorizedv1alpha1 "github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1" +) + +func TestListenerPorts(t *testing.T) { + type external = redpandachart.ExternalListener[redpandachart.KafkaAuthenticationMethod] + + for _, tc := range []struct { + name string + external map[string]external + want []int32 + }{ + {name: "internal only", want: []int32{9093}}, + { + name: "enabled externals, sorted", + external: map[string]external{ + "b": {Port: 9095}, + "a": {Port: 9094, Enabled: ptr.To(true)}, + }, + want: []int32{9093, 9094, 9095}, + }, + { + name: "disabled and portless externals bind nothing", + external: map[string]external{ + "off": {Port: 9094, Enabled: ptr.To(false)}, + "unset": {}, + }, + want: []int32{9093}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + listener := redpandachart.ListenerConfig[redpandachart.KafkaAuthenticationMethod]{Port: 9093, External: tc.external} + require.Equal(t, tc.want, listenerPorts(listener)) + }) + } +} + +func TestV1KafkaPorts(t *testing.T) { + external := vectorizedv1alpha1.ExternalConnectivityConfig{Enabled: true} + + for _, tc := range []struct { + name string + listeners []vectorizedv1alpha1.KafkaAPI + want []int32 + }{ + {name: "no internal listener", listeners: []vectorizedv1alpha1.KafkaAPI{{Port: 30092, External: external}}}, + {name: "internal only", listeners: []vectorizedv1alpha1.KafkaAPI{{Port: 9092}}, want: []int32{9092}}, + { + name: "external naming its port", + listeners: []vectorizedv1alpha1.KafkaAPI{{Port: 9092}, {Port: 30092, External: external}}, + want: []int32{9092, 30092}, + }, + { + name: "external naming no port binds the internal one's plus one", + listeners: []vectorizedv1alpha1.KafkaAPI{{Port: 9092}, {External: external}}, + want: []int32{9092, 9093}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + cluster := &vectorizedv1alpha1.Cluster{} + cluster.Spec.Configuration.KafkaAPI = tc.listeners + require.Equal(t, tc.want, v1KafkaPorts(cluster)) + }) + } +}