charts/redpanda: Enable using localhost for SR/PP internal clients

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`.
Default behavior unchanged (continues using headless service FQDN for
backward compatibility).

Author: sbocinec

charts/redpanda/chart/templates/_configmap.go.tpl

@@ -120,10 +120,10 @@
 {{- $redpandaYaml := (dict "redpanda" $redpanda "schema_registry" (get (fromJson (include "redpanda.schemaRegistry" (dict "a" (list $state)))) "r") "pandaproxy" (get (fromJson (include "redpanda.pandaProxyListener" (dict "a" (list $state)))) "r") "config_file" "/etc/redpanda/redpanda.yaml") -}}
 {{- if $includeNonHashableItems -}}
 {{- $_ := (set $redpandaYaml "rpk" (get (fromJson (include "redpanda.rpkNodeConfig" (dict "a" (list $state $pool)))) "r")) -}}
-{{- $_ := (set $redpandaYaml "pandaproxy_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state)))) "r")) -}}
-{{- $_ := (set $redpandaYaml "schema_registry_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state)))) "r")) -}}
+{{- $_ := (set $redpandaYaml "pandaproxy_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state "pandaproxy")))) "r")) -}}
+{{- $_ := (set $redpandaYaml "schema_registry_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state "schema_registry")))) "r")) -}}
 {{- if (and (and (get (fromJson (include "redpanda.RedpandaAtLeast_23_3_0" (dict "a" (list $state)))) "r") $state.Values.auditLogging.enabled) (get (fromJson (include "redpanda.Auth.IsSASLEnabled" (dict "a" (list $state.Values.auth)))) "r")) -}}
-{{- $_ := (set $redpandaYaml "audit_log_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state)))) "r")) -}}
+{{- $_ := (set $redpandaYaml "audit_log_client" (get (fromJson (include "redpanda.kafkaClient" (dict "a" (list $state "audit_log")))) "r")) -}}
 {{- end -}}
 {{- end -}}
 {{- $_is_returning = true -}}
@@ -465,15 +465,34 @@
 
 {{- define "redpanda.kafkaClient" -}}
 {{- $state := (index .a 0) -}}
+{{- $clientType := (index .a 1) -}}
 {{- range $_ := (list 1) -}}
 {{- $_is_returning := false -}}
 {{- $brokerList := (list) -}}
+{{- $useLocalhostKey := (printf "%s_client.use_localhost" $clientType) -}}
+{{- $useLocalhost := false -}}
+{{- $_506_val_ok := (get (fromJson (include "_shims.dicttest" (dict "a" (list $state.Values.config.node $useLocalhostKey (coalesce nil))))) "r") -}}
+{{- $val := (index $_506_val_ok 0) -}}
+{{- $ok := (index $_506_val_ok 1) -}}
+{{- if $ok -}}
+{{- if (kindIs "bool" $val) -}}
+{{- $useLocalhost = (eq $val true) -}}
+{{- else -}}{{- if (kindIs "string" $val) -}}
+{{- $strVal := (get (fromJson (include "_shims.typeassertion" (dict "a" (list "string" $val)))) "r") -}}
+{{- $useLocalhost = (or (or (or (eq $strVal "true") (eq $strVal "True")) (eq $strVal "TRUE")) (eq $strVal "1")) -}}
+{{- end -}}
+{{- end -}}
+{{- end -}}
+{{- if $useLocalhost -}}
+{{- $brokerList = (concat (default (list) $brokerList) (list (dict "address" "localhost" "port" ($state.Values.listeners.kafka.port | int)))) -}}
+{{- else -}}
 {{- range $_, $broker := (get (fromJson (include "redpanda.BrokerList" (dict "a" (list $state -1)))) "r") -}}
 {{- $brokerList = (concat (default (list) $brokerList) (list (dict "address" $broker "port" ($state.Values.listeners.kafka.port | int)))) -}}
 {{- end -}}
 {{- if $_is_returning -}}
 {{- break -}}
 {{- end -}}
+{{- end -}}
 {{- $kafkaTLS := $state.Values.listeners.kafka.tls -}}
 {{- $brokerTLS := (coalesce nil) -}}
 {{- if (get (fromJson (include "redpanda.InternalTLS.IsEnabled" (dict "a" (list $state.Values.listeners.kafka.tls $state.Values.tls)))) "r") -}}
@@ -623,17 +642,17 @@
 {{- end -}}
 {{- $enabledOptions := (dict "true" true "1" true "" true) -}}
 {{- $lockMemory := false -}}
-{{- $_671_value_14_ok_15 := (get (fromJson (include "_shims.dicttest" (dict "a" (list $flags "--lock-memory" "")))) "r") -}}
-{{- $value_14 := (index $_671_value_14_ok_15 0) -}}
-{{- $ok_15 := (index $_671_value_14_ok_15 1) -}}
+{{- $_692_value_14_ok_15 := (get (fromJson (include "_shims.dicttest" (dict "a" (list $flags "--lock-memory" "")))) "r") -}}
+{{- $value_14 := (index $_692_value_14_ok_15 0) -}}
+{{- $ok_15 := (index $_692_value_14_ok_15 1) -}}
 {{- if $ok_15 -}}
 {{- $lockMemory = (ternary (index $enabledOptions $value_14) false (hasKey $enabledOptions $value_14)) -}}
 {{- $_ := (unset $flags "--lock-memory") -}}
 {{- end -}}
 {{- $overprovisioned := false -}}
-{{- $_678_value_16_ok_17 := (get (fromJson (include "_shims.dicttest" (dict "a" (list $flags "--overprovisioned" "")))) "r") -}}
-{{- $value_16 := (index $_678_value_16_ok_17 0) -}}
-{{- $ok_17 := (index $_678_value_16_ok_17 1) -}}
+{{- $_699_value_16_ok_17 := (get (fromJson (include "_shims.dicttest" (dict "a" (list $flags "--overprovisioned" "")))) "r") -}}
+{{- $value_16 := (index $_699_value_16_ok_17 0) -}}
+{{- $ok_17 := (index $_699_value_16_ok_17 1) -}}
 {{- if $ok_17 -}}
 {{- $overprovisioned = (ternary (index $enabledOptions $value_16) false (hasKey $enabledOptions $value_16)) -}}
 {{- $_ := (unset $flags "--overprovisioned") -}}

charts/redpanda/chart/values.yaml

@@ -958,6 +958,10 @@ config:
 
   # Reference schema registry client https://docs.redpanda.com/current/reference/node-configuration-sample/
   schema_registry_client: {}
+    #  # Override to use localhost instead of headless service FQDN.
+    #  # Useful when DNS issues occur. Default: false (uses headless service)
+    # use_localhost: false
+    #
     #  # Number of times to retry a request to a broker
     #  # Default: 5
     # retries: 5
@@ -1000,6 +1004,10 @@ config:
 
   # Reference panda proxy client https://docs.redpanda.com/current/reference/node-configuration-sample/
   pandaproxy_client: {}
+    #  # Override to use localhost instead of headless service FQDN.
+    #  # Useful when DNS issues occur. Default: false (uses headless service)
+    # use_localhost: false
+    #
     #  # Number of times to retry a request to a broker
     #  # Default: 5
     # retries: 5

charts/redpanda/configmap.tpl.go

@@ -147,10 +147,10 @@ func RedpandaConfigFile(state *RenderState, includeNonHashableItems bool, pool P
 
 	if includeNonHashableItems {
 		redpandaYaml["rpk"] = rpkNodeConfig(state, pool)
-		redpandaYaml["pandaproxy_client"] = kafkaClient(state)
-		redpandaYaml["schema_registry_client"] = kafkaClient(state)
+		redpandaYaml["pandaproxy_client"] = kafkaClient(state, "pandaproxy")
+		redpandaYaml["schema_registry_client"] = kafkaClient(state, "schema_registry")
 		if RedpandaAtLeast_23_3_0(state) && state.Values.AuditLogging.Enabled && state.Values.Auth.IsSASLEnabled() {
-			redpandaYaml["audit_log_client"] = kafkaClient(state)
+			redpandaYaml["audit_log_client"] = kafkaClient(state, "audit_log")
 		}
 	}
 
@@ -489,13 +489,34 @@ func rpkSchemaRegistryClientTLSConfiguration(state *RenderState) map[string]any
 // kafkaClient returns the configuration for internal components of redpanda to
 // connect to its own Kafka API. This is distinct from RPK's configuration for
 // Kafka API interactions.
-func kafkaClient(state *RenderState) map[string]any {
+func kafkaClient(state *RenderState, clientType string) map[string]any {
 	brokerList := []map[string]any{}
-	for _, broker := range BrokerList(state, -1) {
+
+	useLocalhostKey := fmt.Sprintf("%s_client.use_localhost", clientType)
+	useLocalhost := false
+
+	val, ok := state.Values.Config.Node[useLocalhostKey]
+	if ok {
+		if helmette.KindIs("bool", val) {
+			useLocalhost = val == true
+		} else if helmette.KindIs("string", val) {
+			strVal := val.(string)
+			useLocalhost = strVal == "true" || strVal == "True" || strVal == "TRUE" || strVal == "1"
+		}
+	}
+
+	if useLocalhost {
 		brokerList = append(brokerList, map[string]any{
-			"address": broker,
+			"address": "localhost",
 			"port":    state.Values.Listeners.Kafka.Port,
 		})
+	} else {
+		for _, broker := range BrokerList(state, -1) {
+			brokerList = append(brokerList, map[string]any{
+				"address": broker,
+				"port":    state.Values.Listeners.Kafka.Port,
+			})
+		}
 	}
 
 	kafkaTLS := state.Values.Listeners.Kafka.TLS