From b2af20690c7552028f190986a65fb035e5512bc0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=C4=BDubo=C5=A1=20Hozz=C3=A1n?= <hozzan@gmail.com>
Date: Mon, 13 May 2024 11:07:52 +0200
Subject: [PATCH] Minor improvement for Kafka (#31)

---
 chart/templates/init-configmap.yaml          | 27 +++++++++++++++-----
 chart/templates/kafka/kafka-service.yaml     | 10 +++++---
 chart/templates/kafka/kafka-statefulset.yaml | 26 +++++++------------
 chart/values.yaml                            |  4 +--
 4 files changed, 38 insertions(+), 29 deletions(-)

diff --git a/chart/templates/init-configmap.yaml b/chart/templates/init-configmap.yaml
index 1f5f02d..449900a 100644
--- a/chart/templates/init-configmap.yaml
+++ b/chart/templates/init-configmap.yaml
@@ -98,16 +98,31 @@ data:
   check_job.sh: |-
     #!/bin/bash
 
-    set -e
+    set -o pipefail
 
-    echo "waiting for $job_name job completion"
+    echo -n "INFO: Checking Kubernetes API availability ... "
     TOKEN=$(cat /run/secrets/kubernetes.io/serviceaccount/token)
 
+    response=$(curl -sS -X GET https://$KUBERNETES_SERVICE_HOST/apis/batch/v1/namespaces/$job_namespace/jobs/$job_name/status \
+      --header "Authorization: Bearer $TOKEN" --cacert /run/secrets/kubernetes.io/serviceaccount/ca.crt \
+      -w "%{http_code}" -o /dev/null)
+
+    if [[ $response != "200" ]]; then
+      echo -e "FAILED\nERROR: Response:"
+      curl -S -X GET https://$KUBERNETES_SERVICE_HOST/apis/batch/v1/namespaces/$job_namespace/jobs/$job_name/status \
+      --header "Authorization: Bearer $TOKEN" --cacert /run/secrets/kubernetes.io/serviceaccount/ca.crt
+      exit 1
+    else
+      echo -e "OK"
+    fi
+
     state="down"
     while [[ ! $state =~ "1" ]]; do
-      state=`curl -sS -X GET https://$KUBERNETES_SERVICE_HOST/apis/batch/v1/namespaces/$job_namespace/jobs/$job_name/status \
-      --header "Authorization: Bearer $TOKEN" --cacert /run/secrets/kubernetes.io/serviceaccount/ca.crt | jq '.status.succeeded'`
-      sleep 3
+      state=$(curl -sS -X GET https://$KUBERNETES_SERVICE_HOST/apis/batch/v1/namespaces/$job_namespace/jobs/$job_name/status \
+      --header "Authorization: Bearer $TOKEN" --cacert /run/secrets/kubernetes.io/serviceaccount/ca.crt | \
+      jq '.status.succeeded')
+      echo "INFO: Waiting for job $job_name (\"$state\") ..."
+      sleep 5
     done
 
-    echo "$job_name job status.succeeded $state"
\ No newline at end of file
+    echo "INFO: Initialized"
\ No newline at end of file
diff --git a/chart/templates/kafka/kafka-service.yaml b/chart/templates/kafka/kafka-service.yaml
index 7d1b75e..e5c8240 100644
--- a/chart/templates/kafka/kafka-service.yaml
+++ b/chart/templates/kafka/kafka-service.yaml
@@ -8,22 +8,23 @@ spec:
   clusterIP: None
   ports:
   - port: 9092
-    name: kafka
+    name: listener
   selector: {{ include "logging.selectorLabels" . | indent 4 }}
     type: kafka
-{{ range $kafkaId := until (.Values.kafka.replicas | int) }}
 ---
+{{ range $kafkaId := until (.Values.kafka.replicas | int) }}
 apiVersion: v1
 kind: Service
 metadata:
   name: {{ $.Release.Name }}-kafka-{{ $kafkaId }}
   labels: {{ include "logging.labels" $ | indent 4 }}
+    type: kafka
 spec:
   ports:
     - port: 9092
-      name: kafka
+      name: listener
     - port: 9093
-      name: broker
+      name: controller
     - port: 9094
       name: internal
     - port: 12345
@@ -31,5 +32,6 @@ spec:
   selector: {{ include "logging.selectorLabels" $ | indent 4 }}
     statefulset.kubernetes.io/pod-name: {{ $.Release.Name }}-kafka-{{ $kafkaId }}
     type: kafka
+---
 {{ end }}
 {{- end }}
\ No newline at end of file
diff --git a/chart/templates/kafka/kafka-statefulset.yaml b/chart/templates/kafka/kafka-statefulset.yaml
index e5f73ea..ac50e5c 100644
--- a/chart/templates/kafka/kafka-statefulset.yaml
+++ b/chart/templates/kafka/kafka-statefulset.yaml
@@ -63,24 +63,14 @@ spec:
         imagePullPolicy: {{ .Values.kafka.imagePullPolicy }}
         ports:
         - containerPort: 9092
-          name: kafka
+          name: listener
         - containerPort: 9093
-          name: broker
+          name: controller
         - containerPort: 9094
           name: internal
         - containerPort: 12345
           name: jmx
         resources: {{ toYaml .Values.kafka.resources | nindent 10 }}
-        ### Workaround for missing but required variable
-        ### Details: https://github.com/bitnami/containers/issues/33271
-        command:
-        - /bin/bash
-        args:
-        - -ec
-        - |
-          export KAFKA_CFG_NODE_ID="$(echo "$MY_POD_NAME" | grep -o -E '[0-9]*$')"
-          /opt/bitnami/scripts/kafka/entrypoint.sh /opt/bitnami/scripts/kafka/run.sh
-        ### --- ###
         env:
         # - name: BITNAMI_DEBUG # get more information from Kafka pod logs
         #   value: "true" # default "false"
@@ -94,10 +84,12 @@ spec:
           value: "PLAINTEXT://:9092,CONTROLLER://:9093,INTERNAL://:9094"
         - name: KAFKA_CFG_INITIAL_BROKER_REGISTRATION_TIMEOUT_MS
           value: "270000" # default 60000 (1 minute)
+        - name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES
+          value: "CONTROLLER"
         - name: KAFKA_CFG_LOG_DIRS
           value: "/kafka/data"
-        # - name: NODE_ID_COMMAND # need for get broker ID, broken, see workaround above
-        #   value: 'hostname | cut -d "-" -f 3'
+        - name: KAFKA_NODE_ID_COMMAND # need for get broker ID
+          value: "echo ${MY_POD_NAME} | grep -o -E '[0-9]*$'"
         - name: KAFKA_HEAP_OPTS
           value: "-Xmx{{ .Values.kafka.heapSize }} -Xms{{ .Values.kafka.heapSize }}"
         - name: ALLOW_PLAINTEXT_LISTENER
@@ -107,7 +99,7 @@ spec:
             fieldRef:
               fieldPath: metadata.name
         - name: KAFKA_CFG_ADVERTISED_LISTENERS
-          value: "INTERNAL://$(MY_POD_NAME).logging.svc.cluster.local:9094,PLAINTEXT://$(MY_POD_NAME).logging.svc.cluster.local:9092"
+          value: "INTERNAL://$(MY_POD_NAME).{{ .Release.Namespace }}.svc.cluster.local:9094,PLAINTEXT://$(MY_POD_NAME).{{ .Release.Namespace }}.svc.cluster.local:9092"
         - name: KAFKA_INTER_BROKER_LISTENER_NAME
           value: "INTERNAL"
         - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
@@ -122,7 +114,7 @@ spec:
               command: ["/opt/bitnami/kafka/bin/kafka-server-stop.sh"]
         readinessProbe:
           tcpSocket:
-            port: kafka
+            port: controller
           initialDelaySeconds: 10
           periodSeconds: 5
           timeoutSeconds: 3
@@ -130,7 +122,7 @@ spec:
           failureThreshold: 24
         livenessProbe:
           tcpSocket:
-            port: kafka
+            port: controller
           initialDelaySeconds: 15
           periodSeconds: 10
           timeoutSeconds: 5
diff --git a/chart/values.yaml b/chart/values.yaml
index e4ede63..db75cf2 100644
--- a/chart/values.yaml
+++ b/chart/values.yaml
@@ -25,7 +25,7 @@ opensearch:
   imagePullPolicy: IfNotPresent
   # -- Set [deployment layout](https://nickytd.github.io/kubernetes-logging-helm/docs/deployments/) for OpenSearch
   singleNode: false
-  # -- TODO
+  # -- If `true`, timestamp will be collected and showed with full precision.
   timeNanoSeconds: false
   # externalOpensearch.disabled designates an external managed opensearch stack.
   externalOpensearch:
@@ -198,7 +198,7 @@ opensearch_dashboards:
     url: ""
   # -- Replicas count for OpenSearch Dashboard pods.
   replicas: 1
-  extraEnvs: {}
+  extraEnvs: []
   # -- Set user name with *admin* privileges.
   user: "opensearch"
   # -- Set password for user with *admin* privileges.