Enable using localhost for SR/PP internal clients in v1 (#1233)

Add opt-in configuration to use localhost instead of headless service
FQDN for Schema Registry and Pandaproxy internal client broker addresses
via `pandaproxy_client.use_localhost` and `schema_registry_client.use_localhost`
in additionalConfiguration. Default behavior unchanged (continues using
headless service FQDN for backward compatibility).

Author: sbocinec

operator/pkg/resources/configmap_test.go

@@ -518,6 +518,51 @@ func TestConfigMapResource_replicas(t *testing.T) { //nolint:funlen // test tabl
 	}
 }
 
+func TestConfigmap_InternalClientsUseLocalhost(t *testing.T) {
+	panda := pandaCluster().DeepCopy()
+	panda.Spec.Configuration.SchemaRegistry = &vectorizedv1alpha1.SchemaRegistryAPI{
+		Port: 8081,
+	}
+	panda.Spec.Configuration.PandaproxyAPI = []vectorizedv1alpha1.PandaproxyAPI{
+		{Port: 8082},
+	}
+	panda.Spec.AdditionalConfiguration = map[string]string{
+		"pandaproxy_client.use_localhost":      "true",
+		"schema_registry_client.use_localhost": "true",
+	}
+
+	c := fake.NewClientBuilder().Build()
+	secret := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "archival",
+			Namespace: "default",
+		},
+		Data: map[string][]byte{
+			"archival": []byte("XXX"),
+		},
+	}
+	require.NoError(t, c.Create(t.Context(), &secret))
+
+	cfg, err := resources.CreateConfiguration(
+		t.Context(), c, nil, panda,
+		"cluster.default.svc.cluster.local",
+		types.NamespacedName{Namespace: "namespace", Name: "pandaproxy"},
+		types.NamespacedName{Namespace: "namespace", Name: "schemaregistry"},
+		types.NamespacedName{Namespace: "namespace", Name: "rpk"},
+		TestBrokerTLSConfigProvider{},
+	)
+	require.NoError(t, err)
+
+	redpandaYaml, err := cfg.ReifyNodeConfiguration(t.Context())
+	require.NoError(t, err)
+
+	// Should use localhost when override is set
+	require.Equal(t, []config.SocketAddress{{Address: "localhost", Port: 123}},
+		redpandaYaml.PandaproxyClient.Brokers)
+	require.Equal(t, []config.SocketAddress{{Address: "localhost", Port: 123}},
+		redpandaYaml.SchemaRegistryClient.Brokers)
+}
+
 func TestConfigmap_BrokerTLSClients(t *testing.T) {
 	panda := pandaCluster().DeepCopy()
 	panda.Spec.Configuration.KafkaAPI[0].TLS = vectorizedv1alpha1.KafkaAPITLS{

operator/pkg/resources/configuration.go

@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"maps"
 	"slices"
+	"strconv"
 
 	cmmetav1 "github.com/cert-manager/cert-manager/pkg/apis/meta/v1"
 	"github.com/cockroachdb/errors"
@@ -429,9 +430,18 @@ func preparePandaproxyClient(
 	// Alternatively, localhost could be used to the same end but using DNS
 	// might save us in some cases where the local broker is having some
 	// trouble and it makes the config more generally reusable.
+	// Can be overridden to localhost via additionalConfiguration:
+	//   pandaproxy_client.use_localhost: "true"
+	brokerAddress := serviceFQDN
+	if useLocalhost, exists := pandaCluster.Spec.AdditionalConfiguration["pandaproxy_client.use_localhost"]; exists {
+		if parsedBool, err := strconv.ParseBool(useLocalhost); err == nil && parsedBool {
+			brokerAddress = "localhost"
+		}
+	}
+
 	cfg.Node.PandaproxyClient = &config.KafkaClient{
 		Brokers: []config.SocketAddress{
-			{Address: serviceFQDN, Port: pandaCluster.InternalListener().Port},
+			{Address: brokerAddress, Port: pandaCluster.InternalListener().Port},
 		},
 	}
 
@@ -497,9 +507,18 @@ func prepareSchemaRegistryClient(
 	// Alternatively, localhost could be used to the same end but using DNS
 	// might save us in some cases where the local broker is having some
 	// trouble and it makes the config more generally reusable.
+	// Can be overridden to localhost via additionalConfiguration:
+	//   schema_registry_client.use_localhost: "true"
+	brokerAddress := serviceFQDN
+	if useLocalhost, exists := pandaCluster.Spec.AdditionalConfiguration["schema_registry_client.use_localhost"]; exists {
+		if parsedBool, err := strconv.ParseBool(useLocalhost); err == nil && parsedBool {
+			brokerAddress = "localhost"
+		}
+	}
+
 	cfg.Node.SchemaRegistryClient = &config.KafkaClient{
 		Brokers: []config.SocketAddress{
-			{Address: serviceFQDN, Port: pandaCluster.InternalListener().Port},
+			{Address: brokerAddress, Port: pandaCluster.InternalListener().Port},
 		},
 	}