[helm] Enable SASL authenticated connection to Zookeeper nodes

.github/workflows/helm-chart.yaml

@@ -0,0 +1,54 @@
+################################################################################
+# 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.
+################################################################################
+
+name: "Helm Chart"
+
+permissions:
+  contents: read
+
+on:
+  pull_request:
+    paths:
+      - 'helm/**'
+      - '.github/workflows/helm-chart.yaml'
+  push:
+    branches: [main, release-*, ci-*]
+    paths:
+      - 'helm/**'
+      - '.github/workflows/helm-chart.yaml'
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  test-helm-chart:
+    name: "Helm Lint and Unittest"
+    runs-on: ubuntu-latest
+    steps:
+      - name: "Checkout code"
+        uses: actions/checkout@v6
+
+      - name: "Run helm-unittest"
+        uses: d3adb5/helm-unittest-action@v2
+        with:
+          helm-version: latest
+          charts: helm
+
+      - name: "Lint Helm chart"
+        run: helm lint ./helm

helm/README.md

@@ -22,7 +22,7 @@ This chart deploys an Apache Fluss cluster on Kubernetes, following Helm best pr
 It requires a Zookeeper ensemble to be running in the same Kubernetes cluster. In future releases, we may add support for an embedded Zookeeper cluster.
 
 
-## Development environment 
+## Development environment
 
 | component                                                                      | version |
 | ------------------------------------------------------------------------------ | ------- |
@@ -33,7 +33,7 @@ It requires a Zookeeper ensemble to be running in the same Kubernetes cluster. I
 | [Apache Fluss](https://fluss.apache.org/docs/)                                 | v0.10.0-incubating  |
 
 
-## Image requirements 
+## Image requirements
 
 A container image for Fluss is available on DockerHub as `fluss/fluss`. You can use it directly or build your own from this repo. To use your own image you need to build the project with [Maven](https://fluss.apache.org/community/dev/building/) and build it with Docker.
 
@@ -80,7 +80,7 @@ This assumes, that Zookeeper is reachable at `zk-zookeeper.<your-namespace>.svc.
 
 ```bash
 helm install fluss ./fluss-helm \
-  --set zookeeper.address=<your-zk-address>
+  --set configurationOverrides.zookeeper.address=<your-zk-address>
 ```
 
 ## Configuration reference

helm/templates/_helpers.tpl

@@ -63,4 +63,4 @@ Selector labels
 {{- define "fluss.selectorLabels" -}}
 app.kubernetes.io/name: {{ include "fluss.name" . }}
 app.kubernetes.io/instance: {{ .Release.Name }}
-{{- end }}
+{{- end }}

helm/templates/_sasl.tpl

@@ -0,0 +1,175 @@
+#
+# 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.
+#
+
+{{/*
+Return true if SASL is configured in any of the listener protocols
+*/}}
+{{- define "fluss.sasl.enabled" -}}
+{{- $enabled := false -}}
+{{- range $id, $l := .Values.listeners -}}
+  {{- if and (not $enabled) (regexFind "SASL" (upper $l.protocol)) -}}
+    {{- $enabled = true -}}
+  {{- end -}}
+{{- end -}}
+{{- if $enabled -}}
+{{- true -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return true if ZooKeeper SASL is enabled
+*/}}
+{{- define "fluss.zookeeper.sasl.enabled" -}}
+{{- $zkSaslUsername := default "" .Values.security.zookeeperSasl.username | trim -}}
+{{- $zkSaslPassword := default "" .Values.security.zookeeperSasl.password | trim -}}
+{{- if or (ne $zkSaslUsername "") (ne $zkSaslPassword "") -}}
+{{- true -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return true if any JAAS configuration is required
+*/}}
+{{- define "fluss.jaas.required" -}}
+{{- if or (include "fluss.sasl.enabled" .) (include "fluss.zookeeper.sasl.enabled" .) -}}
+{{- true -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return true if listener protocol is SASL
+Usage: include "fluss.listener.sasl.enabled" (dict "root" . "listener" "internal")
+*/}}
+{{- define "fluss.listener.sasl.enabled" -}}
+{{- $listener := index .root.Values.listeners .listener -}}
+{{- if and $listener $listener.protocol (regexFind "SASL" (upper $listener.protocol)) -}}
+{{- true -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return upper-cased SASL mechanism for a listener (defaults to PLAIN)
+Usage: include "fluss.listener.sasl.mechanism" (dict "root" . "listener" "internal")
+*/}}
+{{- define "fluss.listener.sasl.mechanism" -}}
+{{- $listener := index .root.Values.listeners .listener -}}
+{{- $security := default (dict) $listener.security -}}
+{{- $mechanism := default "PLAIN" $security.mechanism -}}
+{{- upper $mechanism -}}
+{{- end -}}
+
+{{/*
+Validate SASL mechanism and users for a listener.
+Fails if mechanism is not PLAIN or if any user has empty username/password.
+Usage: include "fluss.sasl.validateListener" (dict "listener" "internal" "mechanism" $mechanism "users" $users "saslEnabled" true)
+*/}}
+{{- define "fluss.sasl.validateListener" -}}
+{{- if .saslEnabled -}}
+{{- if ne .mechanism "PLAIN" -}}
+{{- fail (printf "listeners.%s.security.mechanism must be PLAIN when listeners.%s.protocol is SASL, got %s" .listener .listener .mechanism) -}}
+{{- end -}}
+{{- range $idx, $user := .users -}}
+{{- if or (empty $user.username) (empty $user.password) -}}
+{{- fail (printf "listeners.%s.security.users[%d] must set both username and password" $.listener $idx) -}}
+{{- end -}}
+{{- end -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Auto-generate internal SASL users when none are provided.
+Returns a JSON-encoded list of user dicts. Preserves previously generated credentials from existing secret.
+Usage: $internalUsers := include "fluss.sasl.autoGenerateInternalUsers" (dict "root" . "existingUsers" $users) | fromJsonArray
+*/}}
+{{- define "fluss.sasl.autoGenerateInternalUsers" -}}
+{{- $users := .existingUsers -}}
+{{- if gt (len $users) 0 -}}
+{{- $users | toJson -}}
+{{- else -}}
+{{- $secretName := printf "%s-sasl-jaas-config" (include "fluss.fullname" .root) -}}
+{{- $existingSecret := (lookup "v1" "Secret" .root.Release.Namespace $secretName) -}}
+{{- $existingSecretData := default (dict) $existingSecret.data -}}
+{{- $generatedUser := "" -}}
+{{- $generatedPassword := "" -}}
+{{- if hasKey $existingSecretData "internal-generated-username" -}}
+{{- $generatedUser = (get $existingSecretData "internal-generated-username") | b64dec -}}
+{{- end -}}
+{{- if hasKey $existingSecretData "internal-generated-password" -}}
+{{- $generatedPassword = (get $existingSecretData "internal-generated-password") | b64dec -}}
+{{- end -}}
+{{- if or (empty $generatedUser) (empty $generatedPassword) -}}
+{{- $generatedUser = printf "internal-%s" ((randAlphaNum 12) | lower) -}}
+{{- $generatedPassword = randAlphaNum 32 -}}
+{{- end -}}
+{{- list (dict "username" $generatedUser "password" $generatedPassword) | toJson -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Render a JAAS server block for a named context with the given users.
+Usage: include "fluss.sasl.jaasServerBlock" (dict "name" "internal" "users" $users)
+*/}}
+{{- define "fluss.sasl.jaasServerBlock" }}
+    {{ .name }}.FlussServer {
+       org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required
+       {{- range .users }}
+       user_{{ .username }}="{{ .password }}"
+       {{- end }};
+    };
+{{- end -}}
+
+{{/*
+Render a JAAS client block for inter-node authentication.
+Usage: include "fluss.sasl.jaasClientBlock" (dict "user" $user)
+*/}}
+{{- define "fluss.sasl.jaasClientBlock" }}
+    FlussClient {
+       org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required
+       username="{{ .user.username }}"
+       password="{{ .user.password }}";
+    };
+{{- end -}}
+
+{{/*
+Consolidate SASL configuration logic into a single context.
+Returns a JSON object with:
+  internalSaslEnabled: bool
+  clientSaslEnabled: bool
+  internalUsers: list of {username, password}
+  clientUsers: list of {username, password}
+  internalUsersAutogenerated: bool
+Usage: $saslConfig := include "fluss.sasl.config" . | fromJson
+*/}}
+{{- define "fluss.sasl.config" -}}
+{{- $internalSaslEnabled := eq (include "fluss.listener.sasl.enabled" (dict "root" . "listener" "internal")) "true" -}}
+{{- $clientSaslEnabled := eq (include "fluss.listener.sasl.enabled" (dict "root" . "listener" "client")) "true" -}}
+{{- $internalSecurity := default (dict) .Values.listeners.internal.security -}}
+{{- $clientSecurity := default (dict) .Values.listeners.client.security -}}
+{{- $internalUsersProvided := default (list) $internalSecurity.users -}}
+{{- $internalUsersAutogenerated := and $internalSaslEnabled (eq (len $internalUsersProvided) 0) -}}
+{{- $internalUsers := include "fluss.sasl.autoGenerateInternalUsers" (dict "root" . "existingUsers" $internalUsersProvided) | fromJsonArray -}}
+{{- $clientUsers := default (list) $clientSecurity.users -}}
+{{- $internalMechanism := include "fluss.listener.sasl.mechanism" (dict "root" . "listener" "internal") -}}
+{{- $clientMechanism := include "fluss.listener.sasl.mechanism" (dict "root" . "listener" "client") -}}
+{{- include "fluss.sasl.validateListener" (dict "listener" "internal" "mechanism" $internalMechanism "users" $internalUsers "saslEnabled" $internalSaslEnabled) -}}
+{{- if and $clientSaslEnabled (eq (len $clientUsers) 0) -}}
+{{- fail "listeners.client.security.users must contain at least one user when listeners.client.protocol is SASL" -}}
+{{- end -}}
+{{- include "fluss.sasl.validateListener" (dict "listener" "client" "mechanism" $clientMechanism "users" $clientUsers "saslEnabled" $clientSaslEnabled) -}}
+{{- dict "internalSaslEnabled" $internalSaslEnabled "clientSaslEnabled" $clientSaslEnabled "internalUsers" $internalUsers "clientUsers" $clientUsers "internalUsersAutogenerated" $internalUsersAutogenerated | toJson -}}
+{{- end -}}

helm/templates/secret-jaas-config.yaml

@@ -0,0 +1,72 @@
+#
+# 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.
+#
+
+{{- if (include "fluss.jaas.required" .) }}
+{{- $saslEnabled := include "fluss.sasl.enabled" . -}}
+{{- $zookeeperSaslConfigured := include "fluss.zookeeper.sasl.enabled" . -}}
+{{- $zkSaslUsername := default "" .Values.security.zookeeperSasl.username -}}
+{{- $zkSaslPassword := default "" .Values.security.zookeeperSasl.password -}}
+{{- $zkSaslLoginModuleClass := default "" .Values.security.zookeeperSasl.loginModuleClass -}}
+{{- if and $zookeeperSaslConfigured (or (eq (trim $zkSaslUsername) "") (eq (trim $zkSaslPassword) "")) -}}
+{{- fail "security.zookeeperSasl.username and security.zookeeperSasl.password must be set together" -}}
+{{- end -}}
+{{- if and $zookeeperSaslConfigured (eq (trim $zkSaslLoginModuleClass) "") -}}
+{{- fail "security.zookeeperSasl.loginModuleClass must be set when ZooKeeper SASL is configured" -}}
+{{- end -}}
+{{/* --- Render Secret --- */}}
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "fluss.fullname" . }}-sasl-jaas-config
+  labels:
+    {{- include "fluss.labels" . | nindent 4 }}
+type: Opaque
+stringData:
+{{- if $saslEnabled }}
+{{- $saslConfig := include "fluss.sasl.config" . | fromJson -}}
+{{- if $saslConfig.internalUsersAutogenerated }}
+  internal-generated-username: {{ (first $saslConfig.internalUsers).username | quote }}
+  internal-generated-password: {{ (first $saslConfig.internalUsers).password | quote }}
+{{- end }}
+{{- end }}
+  jaas.conf: |
+{{- if $saslEnabled }}
+{{- $saslConfig := include "fluss.sasl.config" . | fromJson -}}
+{{- if $saslConfig.internalSaslEnabled }}
+{{- include "fluss.sasl.jaasServerBlock" (dict "name" "internal" "users" $saslConfig.internalUsers) }}
+{{- end }}
+{{- if $saslConfig.clientSaslEnabled }}
+{{- include "fluss.sasl.jaasServerBlock" (dict "name" "client" "users" $saslConfig.clientUsers) }}
+{{- end }}
+{{- if $saslConfig.internalSaslEnabled }}
+{{- include "fluss.sasl.jaasClientBlock" (dict "user" (first $saslConfig.internalUsers)) }}
+{{- end }}
+{{- end }}
+{{- if $zookeeperSaslConfigured }}
+    ZookeeperClient {
+       {{ $zkSaslLoginModuleClass }} required
+       username="{{ $zkSaslUsername }}"
+       password="{{ $zkSaslPassword }}";
+    };
+{{- end }}
+{{- if $zookeeperSaslConfigured }}
+  zookeeper-client.properties: |
+    zookeeper.sasl.client=true
+    zookeeper.sasl.clientconfig=ZookeeperClient
+{{- end }}
+{{- end -}}

helm/templates/sts-coordinator.yaml

@@ -77,6 +77,21 @@ spec:
               echo "" >> $FLUSS_HOME/conf/server.yaml && \
               echo "bind.listeners: ${BIND_LISTENERS}" >> $FLUSS_HOME/conf/server.yaml && \
               echo "advertised.listeners: ${ADVERTISED_LISTENERS}" >> $FLUSS_HOME/conf/server.yaml && \
+              echo "security.protocol.map: INTERNAL:{{ upper .Values.listeners.internal.protocol }},CLIENT:{{ upper .Values.listeners.client.protocol }}" >> $FLUSS_HOME/conf/server.yaml && \
+
+              {{- if (include "fluss.sasl.enabled" .) }}
+              echo "security.sasl.enabled.mechanisms: PLAIN" >> $FLUSS_HOME/conf/server.yaml && \
+              {{ if eq (include "fluss.listener.sasl.enabled" (dict "root" . "listener" "internal")) "true" }}
+              echo "client.security.protocol: SASL" >> $FLUSS_HOME/conf/server.yaml && \
+              echo "client.security.sasl.mechanism: {{ include "fluss.listener.sasl.mechanism" (dict "root" . "listener" "internal") }}" >> $FLUSS_HOME/conf/server.yaml && \
+              {{ end }}
+              {{- end }}
+              {{- if (include "fluss.zookeeper.sasl.enabled" .) }}
+              echo "zookeeper.client.config-path: /etc/fluss/conf/zookeeper-client.properties" >> $FLUSS_HOME/conf/server.yaml && \
+              {{- end }}
+              {{- if (include "fluss.jaas.required" .) }}
+              export FLUSS_ENV_JAVA_OPTS="-Djava.security.auth.login.config=/etc/fluss/conf/jaas.conf ${FLUSS_ENV_JAVA_OPTS}" && \
+              {{- end }}
 
               bin/coordinator-server.sh start-foreground
           livenessProbe:
@@ -94,12 +109,17 @@ spec:
             tcpSocket:
               port: {{ .Values.listeners.client.port }}
           resources:
-            {{- toYaml .Values.resources.tabletServer | nindent 12 }}
+            {{- toYaml .Values.resources.coordinatorServer | nindent 12 }}
           volumeMounts:
             - name: fluss-conf
               mountPath: /opt/conf
             - name: data
               mountPath: /tmp/fluss/data
+            {{- if (include "fluss.jaas.required" .) }}
+            - name: sasl-config
+              mountPath: /etc/fluss/conf
+              readOnly: true
+            {{- end }}
       volumes:
         - name: fluss-conf
           configMap:
@@ -108,6 +128,11 @@ spec:
         - name: data
           emptyDir: {}
         {{- end }}
+        {{- if (include "fluss.jaas.required" .) }}
+        - name: sasl-config
+          secret:
+            secretName: {{ include "fluss.fullname" . }}-sasl-jaas-config
+        {{- end }}
   {{- if .Values.coordinator.storage.enabled }}
   volumeClaimTemplates:
     - metadata: