[bitnami/kafka] Allow external access

bitnami/kafka/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v1
 name: kafka
-version: 7.1.3
+version: 7.2.0
 appVersion: 2.4.0
 description: Apache Kafka is a distributed streaming platform.
 keywords:

bitnami/kafka/README.md

@@ -122,6 +122,12 @@ The following tables lists the configurable parameters of the Kafka chart and th
 | `service.nodePort`                      | Kubernetes Service nodePort                                                                                                                               | `nil`                                                   |
 | `service.loadBalancerIP`                | loadBalancerIP for Kafka Service                                                                                                                          | `nil`                                                   |
 | `service.annotations`                   | Service annotations                                                                                                                                       | ``                                                      |
+| `externalAccess.enabled`               | Enable Kubernetes external cluster access to Kafka brokers                                                                                                 | `false`                                                 |
+| `externalAccess.service.type`          | Kubernetes Servive type for external access. It can be NodePort or LoadBalancer                                                                            | `LoadBalancer`                                          |
+| `externalAccess.service.port`          | Kafka port used for external access when service type is LoadBalancer                                                                                      | `19092`                                                 |
+| `externalAccess.service.loadBalancerIP`| Array of load balancer IPs for Kafka brokers.                                                                                                              | `[]`                                                    |
+| `externalAccess.service.domain`        | Domain or external ip used to configure Kafka external listener when service type is NodePort                                                              | `nil`                                                   |
+| `externalAccess.service.nodePort`      | Array of node ports used to configure Kafka external listener when service type is NodePort                                                                | `[]`                                                    |
 | `serviceAccount.create`                 | Enable creation of ServiceAccount for kafka pod                                                                                                           | `false`                                                 |
 | `serviceAccount.name`                   | Name of the created serviceAccount                                                                                                                        | Generated using the `kafka.fullname` template           |
 | `persistence.enabled`                   | Enable Kafka persistence using PVC, note that Zookeeper perisstency is unaffected                                                                         | `true`                                                  |

bitnami/kafka/templates/NOTES.txt

@@ -48,26 +48,39 @@ To start a kafka consumer run the following command:
     kubectl --namespace {{ .Release.Namespace }} exec -it $POD_NAME -- kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
     {{- end }}
 
-To connect to your Kafka server from outside the cluster execute the following commands:
+{{- if .Values.externalAccess.enabled }}
 
-{{- if contains "NodePort" .Values.service.type }}
+To connect to your Kafka server from outside the cluster check the following information:
 
-    export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
-    export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ template "kafka.fullname" . }})
-    echo "Kafka Broker Endpoint: $NODE_IP:$NODE_PORT"
+{{- if contains "NodePort" .Values.externalAccess.service.type }}
 
-{{- else if contains "LoadBalancer" .Values.service.type }}
+{{- if .Values.externalAccess.service.domain }}
+
+    Kafka brokers domain: Use your provided hostname to reach Kafka brokers, {{ .Values.externalAccess.service.domain }}
+
+{{- else }}
+
+    Kafka brokers domain: You can get the external node ip from the Kafka configuration file with the following command (Check the EXTERNAL listener)
+
+        kubectl exec -it KAFKA_POD -- cat /opt/bitnami/kafka/conf/server.properties | grep advertised.listeners
+
+{{- end }}
+
+    Kafka brokers port: You should have a different node port for each Kafka broker. The list of configured node ports is {{ .Values.externalAccess.service.nodePort }}
+
+{{- else if contains "LoadBalancer" .Values.externalAccess.service.type }}
 
   NOTE: It may take a few minutes for the LoadBalancer IP to be available.
         Watch the status with: 'kubectl get svc --namespace {{ .Release.Namespace }} -w {{ template "kafka.fullname" . }}'
 
-    export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ template "kafka.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
-    echo "Kafka Broker Endpoint: $SERVICE_IP:9092"
+    Kafka Brokers domain: You will have a different external IP for each Kafka broker. Get the external ip from `-external` suffixed services: `kubectl get svc`.
+    Kafka Brokers port: {{ .Values.externalAccess.service.port }}
 
-{{- else if contains "ClusterIP" .Values.service.type }}
+{{- else }}
 
-    kubectl port-forward --namespace {{ .Release.Namespace }} svc/{{ template "kafka.fullname" . }} 9092:9092 &
-    echo "Kafka Broker Endpoint: 127.0.0.1:9092"
+    Unsupported service type for external access.
+
+{{- end }}
 
 {{- end }}
 {{ if .Values.auth.enabled }}
@@ -98,3 +111,5 @@ WARNING: Rolling tag detected ({{ .Values.image.repository }}:{{ .Values.image.t
 +info https://docs.bitnami.com/containers/how-to/understand-rolling-tags-containers/
 
 {{- end }}
+
+{{ include "kafka.validateValues" . }}

bitnami/kafka/templates/_helpers.tpl

@@ -259,3 +259,47 @@ Usage:
         {{- tpl (.value | toYaml) .context }}
     {{- end }}
 {{- end -}}
+
+{{/*
+Compile all warnings into a single message, and call fail.
+*/}}
+{{- define "kafka.validateValues" -}}
+{{- $messages := list -}}
+{{- $messages := append $messages (include "kafka.validateValues.loadBalancerIPListLength" .) -}}
+{{- $messages := append $messages (include "kafka.validateValues.nodePortListLength" .) -}}
+{{- $messages := append $messages (include "kafka.validateValues.externalAccessServiceType" .) -}}
+{{- $messages := without $messages "" -}}
+{{- $message := join "\n" $messages -}}
+
+{{- if $message -}}
+{{-   printf "\nVALUES VALIDATION:\n%s" $message | fail -}}
+{{- end -}}
+{{- end -}}
+
+{{/* Validate values of Kafka - number of replicas must be the same than loadBalancerIP list */}}
+{{- define "kafka.validateValues.loadBalancerIPListLength" -}}
+{{- $replicaCount := int .Values.replicaCount }}
+{{- $loadBalancerIPListLength := len .Values.externalAccess.service.loadBalancerIP }}
+{{- if and ( .Values.externalAccess.enabled ) ( not (eq $replicaCount $loadBalancerIPListLength )) (eq .Values.externalAccess.service.type "LoadBalancer") -}}
+kafka: .Values.externalAccess.service.loadBalancerIP
+    Number of replicas and loadBalancerIP array length must be the same.
+{{- end -}}
+{{- end -}}
+
+{{/* Validate values of Kafka - number of replicas must be the same than NodePort list */}}
+{{- define "kafka.validateValues.nodePortListLength" -}}
+{{- $replicaCount := int .Values.replicaCount }}
+{{- $nodePortListLength := len .Values.externalAccess.service.nodePort }}
+{{- if and ( .Values.externalAccess.enabled ) ( not (eq $replicaCount $nodePortListLength )) (eq .Values.externalAccess.service.type "NodePort") -}}
+kafka: .Values.externalAccess.service.nodePort
+    Number of replicas and nodePort array length must be the same.
+{{- end -}}
+{{- end -}}
+
+{{/* Validate values of Kafka - service type for external access */}}
+{{- define "kafka.validateValues.externalAccessServiceType" -}}
+{{- if and (not (eq .Values.externalAccess.service.type "NodePort")) (not (eq .Values.externalAccess.service.type "LoadBalancer")) -}}
+kafka: .Values.externalAccess.service.type
+    Available servive type for external access are NodePort or LoadBalancer.
+{{- end -}}
+{{- end -}}

bitnami/kafka/templates/statefulset.yaml

@@ -59,6 +59,39 @@ spec:
           {{- if .Values.resources }}
           resources: {{ toYaml .Values.resources | nindent 12 }}
           {{- end }}
+          command:
+            - bash
+            - -ec
+            - |
+                HOSTNAME=$(hostname -s)
+                ID=${HOSTNAME:(-1)}
+
+                # Configure external ip and port
+                {{- if eq .Values.externalAccess.service.type "LoadBalancer" }}
+                export EXTERNAL_ACCESS_IP=$(echo '{{ .Values.externalAccess.service.loadBalancerIP }}' | tr -d '[]' | cut -d ' ' -f "$(($ID + 1))")
+                export EXTERNAL_ACCESS_PORT={{ .Values.externalAccess.service.port }}
+                {{- else if eq .Values.externalAccess.service.type "NodePort" }}
+                {{- if .Values.externalAccess.service.domain }}
+                export EXTERNAL_ACCESS_IP={{ .Values.externalAccess.service.domain }}
+                {{- else }}
+                export EXTERNAL_ACCESS_IP=$(curl -s https://ipinfo.io/ip)
+                {{- end }}
+                export EXTERNAL_ACCESS_PORT=$(echo '{{ .Values.externalAccess.service.nodePort }}' | tr -d '[]' | cut -d ' ' -f "$(($ID + 1))")
+                {{- end }}
+
+                # Configure Kafka internal and external listeners
+                {{- if .Values.externalAccess.enabled }}
+                export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+                export KAFKA_CFG_LISTENERS=INTERNAL://:{{ .Values.service.port }},EXTERNAL://:${EXTERNAL_ACCESS_PORT}
+                export KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
+                {{- if .Values.advertisedListeners }}
+                export KAFKA_CFG_ADVERTISED_LISTENERS={{ .Values.advertisedListeners }}
+                {{- else }}
+                export KAFKA_CFG_ADVERTISED_LISTENERS="INTERNAL://${MY_POD_NAME}.{{ template "kafka.fullname" . }}-headless.{{.Release.Namespace}}.svc.{{ .Values.clusterDomain }}:{{ .Values.service.port }},EXTERNAL://${EXTERNAL_ACCESS_IP}:${EXTERNAL_ACCESS_PORT}"
+                {{- end }}
+                {{- end }}
+
+                exec /entrypoint.sh /run.sh
           env:
             - name: BITNAMI_DEBUG
               value: {{ ternary "true" "false" .Values.image.debug | quote }}
@@ -78,6 +111,7 @@ spec:
               {{- end }}
             - name: KAFKA_PORT_NUMBER
               value: {{ .Values.service.port | quote }}
+            {{- if not .Values.externalAccess.enabled }}
             - name: KAFKA_CFG_LISTENERS
               {{- if .Values.listeners }}
               value: {{ .Values.listeners }}
@@ -98,6 +132,7 @@ spec:
               {{- else }}
               value: 'PLAINTEXT://$(MY_POD_NAME).{{ template "kafka.fullname" . }}-headless.{{.Release.Namespace}}.svc.{{ .Values.clusterDomain }}:$(KAFKA_PORT_NUMBER)'
               {{- end }}
+            {{- end }}
             {{- if .Values.listenerSecurityProtocolMap }}
             - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
               value: {{ .Values.listenerSecurityProtocolMap }}
@@ -207,6 +242,10 @@ spec:
             - name: kafka-ssl
               containerPort: 9093
             {{- end }}
+            {{- if and .Values.externalAccess.enabled (eq .Values.externalAccess.service.type "LoadBalancer") }}
+            - name: kafka-external
+              containerPort: 19092
+            {{- end }}
           {{- if .Values.livenessProbe.enabled }}
           livenessProbe:
             tcpSocket:

bitnami/kafka/templates/svc-external-access.yaml

@@ -0,0 +1,44 @@
+{{- if .Values.externalAccess.enabled -}}
+  {{- $fullName := include "kafka.fullname" . }}
+  {{- $replicaCount := .Values.replicaCount | int }}
+  {{- $root := . }}
+
+{{- range $i, $e := until $replicaCount }}
+  {{- $targetPod := printf "%s-%d" (printf "%s" $fullName) $i }}
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ template "kafka.fullname" $ }}-{{ $i }}-external
+  labels: {{- include "kafka.labels" $ | nindent 4 }}
+    app.kubernetes.io/component: kafka
+    pod: {{ $targetPod }}
+spec:
+  type: {{ $root.Values.externalAccess.service.type }}
+  {{- if eq $root.Values.externalAccess.service.type "LoadBalancer" }}
+  {{- if $root.Values.externalAccess.service.loadBalancerIP }}
+  loadBalancerIP: {{ index $root.Values.externalAccess.service.loadBalancerIP $i }}
+  {{- end }}
+  {{- end }}
+  ports:
+    - name: kafka
+      {{- if eq $root.Values.externalAccess.service.type "LoadBalancer" }}
+      port: {{ $root.Values.externalAccess.service.port }}
+      {{- else }}
+      port: {{ index $root.Values.externalAccess.service.nodePort $i }}
+      {{- end }}
+      {{- if and $root.Values.externalAccess.service.nodePort (eq $root.Values.externalAccess.service.type "NodePort") }}
+      nodePort: {{ index $root.Values.externalAccess.service.nodePort $i }}
+      {{- end }}
+      {{- if eq $root.Values.externalAccess.service.type "LoadBalancer" }}
+      targetPort: kafka-external
+      {{- else }}
+      targetPort: {{ index $root.Values.externalAccess.service.nodePort $i }}
+      {{- end }}
+  selector: {{- include "kafka.matchLabels" $ | nindent 4 }}
+    app.kubernetes.io/component: kafka
+    statefulset.kubernetes.io/pod-name: {{ $targetPod }}
+---
+{{- end }}
+{{- end -}}

bitnami/kafka/values-production.yaml

@@ -35,7 +35,7 @@ image:
 ## String to partially override kafka.fullname template (will maintain the release name)
 # nameOverride:
 
- ## String to fully override kafka.fullname template
+## String to fully override kafka.fullname template
 # fullnameOverride:
 
 ## Init containers parameters:
@@ -207,16 +207,16 @@ auth:
   enabled: true
 
   ## Enable SSL to be used with brokers and consumers
-  #ssl: false
+  # ssl: false
 
   ## Name of the existing secret containing credentials for brokerUser, interBrokerUser and zookeeperUser.
-  #existingSecret:
+  # existingSecret:
 
   ## Name of the existing secret containing the certificate files that will be used by Kafka.
-  #certificatesSecret:
+  # certificatesSecret:
 
   ## Password for the above certificates if they are password protected.
-  #certificatesPassword:
+  # certificatesPassword:
 
   ## Kafka client user.
   brokerUser: user
@@ -231,10 +231,10 @@ auth:
   # interBrokerPassword:
 
   ## Kafka Zookeeper user.
-  #zookeeperUser:
+  # zookeeperUser:
 
   ## Kafka Zookeeper password.
-  #zookeeperPassword:
+  # zookeeperPassword:
 
 ## Kubernetes Security Context
 ## https://kubernetes.io/docs/tasks/configure-pod-container/security-context/
@@ -265,6 +265,25 @@ service:
   ## Service annotations done as key:value pairs
   annotations: {}
 
+externalAccess:
+  enabled: false
+  service:
+    ## Type of service for external access. It can be LoadBalancer or NodePort.
+    ##
+    type: LoadBalancer
+    ## Port used when service type is LoadBalancer
+    ##
+    port: 19092
+    ## Array of load balancer IPs for each Kafka broker. Length must be the same as replicas
+    ##
+    loadBalancerIP: []
+    ## When service type is NodePort, you can specify the domain used for Kafka advertised listeners.
+    ## If not specified, the container will try to get the kubernetes node external IP using: 'curl -s https://ipinfo.io/ip'
+    # domain: mydomain.com
+    ## Array of node ports used for each Kafka broker. Length must be the same as replicas
+    ##
+    nodePort: []
+
 ## Service account for Kafka to use.
 ## ref: https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/
 ##
@@ -496,12 +515,12 @@ metrics:
       ## `whitelistObjectNames: []`
       ##
       overrideConfig: {}
-        # jmxUrl: service:jmx:rmi:///jndi/rmi://127.0.0.1:5555/jmxrmi
-        # lowercaseOutputName: true
-        # lowercaseOutputLabelNames: true
-        # ssl: false
-        # rules:
-        # - pattern: ".*"
+      #   jmxUrl: service:jmx:rmi:///jndi/rmi://127.0.0.1:5555/jmxrmi
+      #   lowercaseOutputName: true
+      #   lowercaseOutputLabelNames: true
+      #   ssl: false
+      #   rules:
+      #   - pattern: ".*"
       ## If you would like to supply your own ConfigMap for JMX metrics, supply the name of that
       ## ConfigMap as an `overrideName` here.
       ##