[helm] Add support advertised listeners for out of cluster clients

helm/templates/_listeners.tpl

@@ -0,0 +1,241 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+{{/*
+Returns "true" when the optional EXTERNAL listener is enabled.
+Usage:
+  include "fluss.listeners.external.enabled" .
+*/}}
+{{- define "fluss.listeners.external.enabled" -}}
+{{- $ext := .Values.listeners.external | default dict -}}
+{{- if $ext.enabled -}}true{{- end -}}
+{{- end -}}
+
+{{/*
+Fluss listener name for the optional EXTERNAL listener (bind/advertise/protocol map).
+Usage:
+  include "fluss.listeners.external.name" .
+*/}}
+{{- define "fluss.listeners.external.name" -}}
+{{- $ext := .Values.listeners.external | default dict -}}
+{{- $ext.name | default "EXTERNAL" | toString | trim -}}
+{{- end -}}
+
+{{/*
+Bind port for the optional EXTERNAL listener.
+Usage:
+  include "fluss.listeners.external.port" .
+*/}}
+{{- define "fluss.listeners.external.port" -}}
+{{- $ext := .Values.listeners.external | default dict -}}
+{{- $ext.port | default 9125 -}}
+{{- end -}}
+
+{{/*
+Security protocol for the EXTERNAL listener. It always mirrors the client
+listener because the EXTERNAL JAAS entry reuses the client SASL users, so an
+independent protocol would have no credentials of its own.
+Usage:
+  include "fluss.listeners.external.protocol" .
+*/}}
+{{- define "fluss.listeners.external.protocol" -}}
+{{- include "fluss.security.listener.protocol" (dict "context" .Values "listener" "client") | trim -}}
+{{- end -}}
+
+{{/*
+Resolved advertised host or port for a component/kind after applying
+per-component overrides and helm tpl (Release.Namespace and similar).
+Empty means "use the chart default" (in-cluster FQDN for CLIENT host, bind
+port for advertised ports).
+Usage:
+  include "fluss.listeners.rawAdvertisedValue" (dict "ctx" . "component" "tablet" "kind" "external" "field" "advertisedHost")
+*/}}
+{{- define "fluss.listeners.rawAdvertisedValue" -}}
+{{- $ctx := .ctx -}}
+{{- $comp := index $ctx.Values .component | default dict -}}
+{{- $compListeners := index $comp "listeners" | default dict -}}
+{{- $compKind := index $compListeners .kind | default dict -}}
+{{- $globalKind := index $ctx.Values.listeners .kind | default dict -}}
+{{- $compVal := index $compKind .field | default "" | toString | trim -}}
+{{- $globalVal := index $globalKind .field | default "" | toString | trim -}}
+{{- $raw := $compVal | default $globalVal -}}
+{{- if $raw -}}
+{{- tpl $raw $ctx | trim -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Advertised port for a component listener. Falls back to the bind port.
+Usage:
+  include "fluss.listeners.advertisedPort" (dict "ctx" . "component" "tablet" "kind" "client")
+*/}}
+{{- define "fluss.listeners.advertisedPort" -}}
+{{- $raw := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" .ctx "component" .component "kind" .kind "field" "advertisedPort") -}}
+{{- if $raw -}}
+{{- $raw -}}
+{{- else if eq .kind "client" -}}
+{{- .ctx.Values.listeners.client.port -}}
+{{- else -}}
+{{- include "fluss.listeners.external.port" .ctx -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+bind.listeners value written into server.yaml at pod start.
+Usage:
+  include "fluss.listeners.bind" .
+*/}}
+{{- define "fluss.listeners.bind" -}}
+{{- $parts := list (printf "INTERNAL://${POD_IP}:%v" .Values.listeners.internal.port) (printf "CLIENT://${POD_IP}:%v" .Values.listeners.client.port) -}}
+{{- if (include "fluss.listeners.external.enabled" .) -}}
+{{- $parts = append $parts (printf "%s://${POD_IP}:%v" (include "fluss.listeners.external.name" .) (include "fluss.listeners.external.port" .)) -}}
+{{- end -}}
+{{- join ", " $parts -}}
+{{- end -}}
+
+{{/*
+advertised.listeners value written into server.yaml at pod start.
+Pass the chart root as ctx and the component name so each StatefulSet can use
+its own headless Service DNS and optional overrides.
+Usage:
+  include "fluss.listeners.advertised" (dict "ctx" . "component" "tablet")
+*/}}
+{{- define "fluss.listeners.advertised" -}}
+{{- $ctx := .ctx -}}
+{{- $component := .component -}}
+{{- $headless := ternary "tablet-server-hs" "coordinator-server-hs" (eq $component "tablet") -}}
+{{- $clientHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" $ctx "component" $component "kind" "client" "field" "advertisedHost") -}}
+{{- if not $clientHost -}}
+{{- $clientHost = printf "${POD_NAME}.%s.${POD_NAMESPACE}.svc.cluster.local" $headless -}}
+{{- end -}}
+{{- $clientPort := include "fluss.listeners.advertisedPort" (dict "ctx" $ctx "component" $component "kind" "client") -}}
+{{- $parts := list (printf "CLIENT://%s:%v" $clientHost $clientPort) -}}
+{{- if (include "fluss.listeners.external.enabled" $ctx) -}}
+{{- $extName := include "fluss.listeners.external.name" $ctx -}}
+{{- $extHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" $ctx "component" $component "kind" "external" "field" "advertisedHost") -}}
+{{- $extPort := include "fluss.listeners.advertisedPort" (dict "ctx" $ctx "component" $component "kind" "external") -}}
+{{- $parts = append $parts (printf "%s://%s:%v" $extName $extHost $extPort) -}}
+{{- end -}}
+{{- join ", " $parts -}}
+{{- end -}}
+
+{{/*
+True when an expression references a specific shell env token, including
+expansion forms like ${TOKEN##*-}.
+Usage:
+  include "fluss.listeners.usesEnvToken" (dict "blob" $expr "token" "POD_NAME")
+*/}}
+{{- define "fluss.listeners.usesEnvToken" -}}
+{{- $blob := .blob | toString -}}
+{{- $token := .token | toString -}}
+{{- if or (contains (printf "$%s" $token) $blob) (contains (printf "${%s" $token) $blob) -}}true{{- end -}}
+{{- end -}}
+
+{{/*
+True when a component's advertised listeners reference the Kubernetes node IP,
+so the StatefulSet needs the NODE_IP env from the Downward API. Only added when
+referenced so default deployments keep an unchanged pod spec.
+Usage:
+  include "fluss.listeners.usesNodeIp" (dict "ctx" . "component" "tablet")
+*/}}
+{{- define "fluss.listeners.usesNodeIp" -}}
+{{- $adv := include "fluss.listeners.advertised" (dict "ctx" .ctx "component" .component) -}}
+{{- if (include "fluss.listeners.usesEnvToken" (dict "blob" $adv "token" "NODE_IP")) -}}true{{- end -}}
+{{- end -}}
+
+{{/*
+True when a host:port pair embeds a per-pod identity token so replicas do not
+all advertise the same endpoint.
+Usage:
+  include "fluss.listeners.hasPerPodIdentity" (dict "host" $host "port" $port)
+*/}}
+{{- define "fluss.listeners.hasPerPodIdentity" -}}
+{{- $blob := printf "%s:%s" (.host | toString) (.port | toString) -}}
+{{- if or
+    (include "fluss.listeners.usesEnvToken" (dict "blob" $blob "token" "POD_NAME"))
+    (include "fluss.listeners.usesEnvToken" (dict "blob" $blob "token" "POD_IP"))
+    (include "fluss.listeners.usesEnvToken" (dict "blob" $blob "token" "NODE_IP"))
+-}}
+true
+{{- end -}}
+{{- end -}}
+
+{{/*
+Warning when tablet replicas > 1 would advertise a shared host:port.
+Usage:
+  include "fluss.listeners.validateWarning" .
+*/}}
+{{- define "fluss.listeners.validateWarning" -}}
+{{/* Only tablet is checked because the coordinator is single-replica today (no HA, FIP-9). */}}
+{{- $replicas := .Values.tablet.numberOfReplicas | int -}}
+{{- if le $replicas 1 -}}
+{{- else -}}
+{{- $msgs := list -}}
+{{- $clientHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" . "component" "tablet" "kind" "client" "field" "advertisedHost") -}}
+{{- if $clientHost -}}
+{{- $clientPort := include "fluss.listeners.advertisedPort" (dict "ctx" . "component" "tablet" "kind" "client") | toString -}}
+{{- if not (include "fluss.listeners.hasPerPodIdentity" (dict "host" $clientHost "port" $clientPort)) -}}
+{{- $msgs = append $msgs (printf "listeners.client advertised host/port is the same for every tablet pod (tablet.numberOfReplicas is %d). Include $POD_NAME, $POD_IP, $NODE_IP, or ${POD_NAME##*-} in the host or port so each pod advertises a unique endpoint." $replicas) -}}
+{{- end -}}
+{{- end -}}
+{{- if (include "fluss.listeners.external.enabled" .) -}}
+{{- $extHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" . "component" "tablet" "kind" "external" "field" "advertisedHost") -}}
+{{- $extPort := include "fluss.listeners.advertisedPort" (dict "ctx" . "component" "tablet" "kind" "external") | toString -}}
+{{- if and $extHost (not (include "fluss.listeners.hasPerPodIdentity" (dict "host" $extHost "port" $extPort))) -}}
+{{- $msgs = append $msgs (printf "listeners.external advertised host/port is the same for every tablet pod (tablet.numberOfReplicas is %d). Include $POD_NAME, $POD_IP, $NODE_IP, or ${POD_NAME##*-} in the host or port so each pod advertises a unique endpoint." $replicas) -}}
+{{- end -}}
+{{- end -}}
+{{- join "\n" $msgs -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Listener configuration errors. Empty string when valid.
+Usage:
+  include "fluss.listeners.validateError" .
+*/}}
+{{- define "fluss.listeners.validateError" -}}
+{{- $msgs := list -}}
+{{- if hasKey (.Values.configurationOverrides | default dict) "bind.listeners" -}}
+{{- $msgs = append $msgs "configurationOverrides cannot set bind.listeners; the chart always writes it from listeners.* values" -}}
+{{- end -}}
+{{- if hasKey (.Values.configurationOverrides | default dict) "advertised.listeners" -}}
+{{- $msgs = append $msgs "configurationOverrides cannot set advertised.listeners; the chart always writes it from listeners.* values" -}}
+{{- end -}}
+{{- if (include "fluss.listeners.external.enabled" .) -}}
+{{- $name := include "fluss.listeners.external.name" . -}}
+{{- if eq $name "" -}}
+{{- $msgs = append $msgs "listeners.external.name must not be empty when listeners.external.enabled is true" -}}
+{{- end -}}
+{{- if has (upper $name) (list "INTERNAL" "CLIENT") -}}
+{{- $msgs = append $msgs "listeners.external.name must not be INTERNAL or CLIENT" -}}
+{{- end -}}
+{{- $extHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" . "component" "tablet" "kind" "external" "field" "advertisedHost") -}}
+{{- $coordHost := include "fluss.listeners.rawAdvertisedValue" (dict "ctx" . "component" "coordinator" "kind" "external" "field" "advertisedHost") -}}
+{{- if or (not $extHost) (not $coordHost) -}}
+{{- $msgs = append $msgs "listeners.external.advertisedHost must be set when listeners.external.enabled is true" -}}
+{{- end -}}
+{{- $internalPort := .Values.listeners.internal.port | int -}}
+{{- $clientPort := .Values.listeners.client.port | int -}}
+{{- $externalPort := include "fluss.listeners.external.port" . | int -}}
+{{- if or (eq $externalPort $internalPort) (eq $externalPort $clientPort) -}}
+{{- $msgs = append $msgs "listeners.external.port must differ from listeners.internal.port and listeners.client.port" -}}
+{{- end -}}
+{{- end -}}
+{{- join "\n" $msgs -}}
+{{- end -}}

helm/templates/_validate.tpl

@@ -25,6 +25,7 @@ Usage:
 {{- $messages := list -}}
 
 {{- $messages = append $messages (include "fluss.security.validateWarning" .) -}}
+{{- $messages = append $messages (include "fluss.listeners.validateWarning" .) -}}
 
 {{- $messages = without $messages "" -}}
 {{- join "\n" $messages -}}
@@ -41,6 +42,7 @@ Usage:
 {{- $messages = append $messages (include "fluss.security.validateError" .) -}}
 {{- $messages = append $messages (include "fluss.metrics.validateError" .) -}}
 {{- $messages = append $messages (include "fluss.secrets.validateError" .) -}}
+{{- $messages = append $messages (include "fluss.listeners.validateError" .) -}}
 
 {{- $messages = without $messages "" -}}
 {{- join "\n" $messages -}}

helm/templates/configmap-jaas-template.yaml

@@ -46,6 +46,14 @@ data:
        user_${FLUSS_JAAS_CLIENT_USERNAME_{{ $idx }}}="${FLUSS_JAAS_CLIENT_PASSWORD_{{ $idx }}}"
 {{- end }};
     };
+{{- if (include "fluss.listeners.external.enabled" .) }}
+    {{ include "fluss.listeners.external.name" . | lower | trim }}.FlussServer {
+       org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required
+{{- range $idx, $user := .Values.security.client.sasl.plain.users | default (list) }}
+       user_${FLUSS_JAAS_CLIENT_USERNAME_{{ $idx }}}="${FLUSS_JAAS_CLIENT_PASSWORD_{{ $idx }}}"
+{{- end }};
+    };
+{{- end }}
 {{- end }}
 {{- end }}
 {{- if (include "fluss.security.zookeeper.sasl.enabled" .) }}

helm/templates/configmap.yaml

@@ -35,7 +35,11 @@ data:
     {{- $enabledMechanisms := include "fluss.security.sasl.enabledMechanisms" . | trim -}}
     {{- $internalMechanism := include "fluss.security.listener.mechanism" (dict "context" .Values "listener" "internal") -}}
     {{- if (include "fluss.security.sasl.enabled" .) }}
-    security.protocol.map: INTERNAL:{{ $internalProtocol }},CLIENT:{{ $clientProtocol }}
+    {{- $protocolEntries := list (printf "INTERNAL:%s" $internalProtocol) (printf "CLIENT:%s" $clientProtocol) -}}
+    {{- if (include "fluss.listeners.external.enabled" .) -}}
+    {{- $protocolEntries = append $protocolEntries (printf "%s:%s" (include "fluss.listeners.external.name" .) (include "fluss.listeners.external.protocol" . | trim)) -}}
+    {{- end }}
+    security.protocol.map: {{ join "," $protocolEntries }}
     security.sasl.enabled.mechanisms: {{ $enabledMechanisms }}
     {{- if ne $internalMechanism "" }}
     client.security.protocol: SASL

helm/templates/sts-coordinator.yaml

@@ -117,15 +117,19 @@ spec:
               containerPort: {{ .Values.listeners.internal.port }}
             - name: client
               containerPort: {{ .Values.listeners.client.port }}
+            {{- if (include "fluss.listeners.external.enabled" .) }}
+            - name: external
+              containerPort: {{ include "fluss.listeners.external.port" . }}
+            {{- end }}
           command:
             - "/bin/sh"
             - "-c"
             - |
               export FLUSS_SERVER_ID=${POD_NAME##*-} && \
               cp /opt/conf/server.yaml $FLUSS_HOME/conf && \
 
-              BIND_LISTENERS="INTERNAL://${POD_IP}:{{ .Values.listeners.internal.port }}, CLIENT://${POD_IP}:{{ .Values.listeners.client.port }}" && \
-              ADVERTISED_LISTENERS="CLIENT://${POD_NAME}.coordinator-server-hs.${POD_NAMESPACE}.svc.cluster.local:{{ .Values.listeners.client.port }}" && \
+              BIND_LISTENERS="{{ include "fluss.listeners.bind" . }}" && \
+              ADVERTISED_LISTENERS="{{ include "fluss.listeners.advertised" (dict "ctx" . "component" "coordinator") }}" && \
 
               echo "" >> $FLUSS_HOME/conf/server.yaml && \
               echo "bind.listeners: ${BIND_LISTENERS}" >> $FLUSS_HOME/conf/server.yaml && \

helm/templates/sts-tablet.yaml

@@ -94,6 +94,12 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            {{- if (include "fluss.listeners.usesNodeIp" (dict "ctx" . "component" "tablet")) }}
+            - name: NODE_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.hostIP
+            {{- end }}
             {{- if (include "fluss.security.jaas.required" .) }}
             - name: FLUSS_ENV_JAVA_OPTS
               value: "-Djava.security.auth.login.config=/etc/fluss/conf/jaas.conf"
@@ -113,15 +119,19 @@ spec:
               containerPort: {{ .Values.listeners.internal.port }}
             - name: client
               containerPort: {{ .Values.listeners.client.port }}
+            {{- if (include "fluss.listeners.external.enabled" .) }}
+            - name: external
+              containerPort: {{ include "fluss.listeners.external.port" . }}
+            {{- end }}
           command:
             - "/bin/sh"
             - "-c"
             - |
               export FLUSS_SERVER_ID=${POD_NAME##*-} && \
               cp /opt/conf/server.yaml $FLUSS_HOME/conf && \
 
-              BIND_LISTENERS="INTERNAL://${POD_IP}:{{ .Values.listeners.internal.port }}, CLIENT://${POD_IP}:{{ .Values.listeners.client.port }}" && \
-              ADVERTISED_LISTENERS="CLIENT://${POD_NAME}.tablet-server-hs.${POD_NAMESPACE}.svc.cluster.local:{{ .Values.listeners.client.port }}" && \
+              BIND_LISTENERS="{{ include "fluss.listeners.bind" . }}" && \
+              ADVERTISED_LISTENERS="{{ include "fluss.listeners.advertised" (dict "ctx" . "component" "tablet") }}" && \
 
               echo "" >> $FLUSS_HOME/conf/server.yaml && \
               echo "tablet-server.id: ${FLUSS_SERVER_ID}" >> $FLUSS_HOME/conf/server.yaml && \

helm/templates/svc-coordinator.yaml

@@ -39,6 +39,12 @@ spec:
       protocol: TCP
       port: {{ .Values.listeners.client.port }}
       targetPort: client
+    {{- if (include "fluss.listeners.external.enabled" .) }}
+    - name: external
+      protocol: TCP
+      port: {{ include "fluss.listeners.external.port" . }}
+      targetPort: external
+    {{- end }}
   selector:
     {{- include "fluss.selectorLabels" . | nindent 4 }}
     app.kubernetes.io/component: coordinator

helm/templates/svc-tablet.yaml

@@ -39,6 +39,12 @@ spec:
       protocol: TCP
       port: {{ .Values.listeners.client.port }}
       targetPort: client
+    {{- if (include "fluss.listeners.external.enabled" .) }}
+    - name: external
+      protocol: TCP
+      port: {{ include "fluss.listeners.external.port" . }}
+      targetPort: external
+    {{- end }}
   selector:
     {{- include "fluss.selectorLabels" . | nindent 4 }}
     app.kubernetes.io/component: tablet