(refactor)kafka: add enhanced kafka health checks with support for ka…

…fka sasl auth (#958)

* (refactor)kafka: add support for improved kafka health checks and sasl auth

Signed-off-by: ksatchit <[email protected]>

experiments/kafka/kafka-broker-pod-failure/kafka-broker-pod-failure-ansible-logic.yml

@@ -13,6 +13,7 @@
     kafka_ns: "{{ lookup('env','KAFKA_NAMESPACE') }}"
     kafka_label: "{{ lookup('env','KAFKA_LABEL') }}"
     kafka_kind: "{{ lookup('env','KAFKA_KIND') }}"
+    kafka_sasl_auth: "{{ lookup('env','KAFKA_SASL_AUTH') }}"
     kafka_broker: "{{ lookup('env','KAFKA_BROKER') }}"
     kafka_stream: "{{ lookup('env','KAFKA_LIVENESS_STREAM') }}" 
     kafka_liveness_image: "{{ lookup('env','KAFKA_LIVENESS_IMAGE') }}" 

experiments/kafka/kafka-broker-pod-failure/kafka-broker-pod-failure-k8s-job.yml

@@ -30,6 +30,10 @@ spec:
           - name: KAFKA_LIVENESS_IMAGE
             value: 'litmuschaos/kafka-client:ci'
 
+          # set to 'enabled' if you have auth set up
+          - name: KAFKA_SASL_AUTH
+            value: 'disabled'
+
           # in milliseconds
           - name: KAFKA_CONSUMER_TIMEOUT
             value: '30000'

utils/apps/kafka/kafka_cluster_health.yml

@@ -14,3 +14,47 @@
     retries: 60
     a_ns: "{{ zk_ns }}"
     a_label: "{{ zk_label }}"
+
+- name: Obtain pods name of any one of the zookeeper pods
+  shell: >
+    kubectl get pods -l {{ zk_label }} -n {{ zk_ns }} 
+    --no-headers -o custom-columns=:metadata.name | head -n 1
+  args:
+    executable: /bin/bash
+  register: zk_result
+
+- set_fact: 
+    zk_pod_name: "{{ zk_result.stdout }}"
+
+- name: Obtain the desired replica count of the Kafka statefulset
+  shell: >
+    kubectl get sts -l {{ kafka_label }} -n {{ kafka_ns }}
+    --no-headers -o custom-columns=:spec.replicas
+  args:
+    executable: /bin/bash
+  register: kafka_sts_replicas
+
+- name: Derive the available kafka brokers from zookeeper
+  shell: >
+    kubectl exec {{ zk_pod_name }} -n {{ zk_ns }} bash 
+    -- zkCli.sh -server {{ zk_service }}:{{ zk_port }}/{{ kafka_instance }}
+    ls /brokers/ids | tail -n 1 | tr -d '[],' | tr ' ' '\n'  | wc -l
+  args: 
+    executable: /bin/bash
+  register: kafka_available_brokers
+  when: "kafka_instance is defined and kafka_instance != ''"
+
+- name: Derive the available kafka brokers from zookeeper
+  shell: >
+    kubectl exec {{ zk_pod_name }} -n {{ zk_ns }} bash 
+    -- zkCli.sh -server {{ zk_service }}:{{ zk_port }} 
+    ls /brokers/ids | tail -n 1 | tr -d '[],' | tr ' ' '\n'  | wc -l
+  args: 
+    executable: /bin/bash
+  register: kafka_available_brokers
+  when: "kafka_instance is not defined or kafka_instance == ''"
+
+- name: Check whether all kafka brokers are alive
+  debug: 
+    msg: "All Kafka brokers are alive"
+  failed_when: "kafka_sts_replicas.stdout != kafka_available_brokers.stdout"

utils/apps/kafka/kafka_liveness_sasl_auth.j2

@@ -0,0 +1,88 @@
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: kafka-liveness
+  labels: 
+    name: kafka-liveness 
+spec:
+  restartPolicy: Never
+  initContainers:
+  - name: kafka-topic-creator
+    image: {{ kafka_liveness_image }}
+    imagePullPolicy: Always
+    env: 
+      - name: TOPIC_NAME
+        value: {{ kafka_topic }}
+      - name: KAFKA_INSTANCE_NAME
+        value: {{ kafka_instance }} 
+      - name: ZOOKEEPER_SERVICE
+        value: {{ zk_service }}
+      - name: ZOOKEEPER_PORT
+        value: "{{ zk_port }}"
+      - name: REPLICATION_FACTOR
+        value: "{{ kafka_replication_factor  }}"
+    command: 
+      - sh
+      - -c
+      - "./topic.sh"
+  containers:
+  - name: kafka-producer
+    image: {{ kafka_liveness_image }}
+    imagePullPolicy: Always
+    env:
+      - name: TOPIC_NAME
+        value: {{ kafka_topic }}
+      - name: KAFKA_SERVICE
+        value: {{ kafka_service }} 
+      - name: KAFKA_PORT
+        value: "{{ kafka_port }}"
+      - name: KAFKA_OPTS
+        value: "-Djava.security.auth.login.config=/opt/jaas.conf"
+    command:
+      - sh
+      - -c
+      - "stdbuf -oL ./producer.sh"
+    volumeMounts:
+      - name: jaas-properties
+        mountPath: /etc
+      - name: client-properties
+        mountPath: /opt
+      - name: jaas-conf
+        mountPath: /opt
+  - name: kafka-consumer
+    image: {{ kafka_liveness_image }}
+    imagePullPolicy: Always
+    env:
+      - name: KAFKA_CONSUMER_TIMEOUT
+        value: "{{ kafka_consumer_timeout }}"
+      - name: TOPIC_NAME
+        value: {{ kafka_topic }}
+      - name: KAFKA_SERVICE
+        value: {{ kafka_service }}
+      - name: KAFKA_PORT
+        value: "{{ kafka_port }}"
+      - name: KAFKA_OPTS
+        value: "-Djava.security.auth.login.config=/opt/jaas.conf"
+    command:
+      - sh
+      - -c
+      - "stdbuf -oL ./consumer.sh"
+    volumeMounts:
+      - name: jaas-properties
+        mountPath: /etc
+      - name: client-properties
+        mountPath: /opt
+      - name: jaas-conf
+        mountPath: /opt
+  volumes: 
+    - name: jaas-properties
+      configMap: 
+        name: jaas-properties
+    - name: client-properties
+      configMap: 
+        name: client-properties
+    - name: jaas-conf
+      configMap: 
+        name: jaas-conf
+

utils/apps/kafka/kafka_liveness_stream.yml

@@ -10,8 +10,15 @@
 
 - name: Generate the kafka liveness spec from template
   template: 
-    src: /utils/apps/kafka/kafka_liveness.j2
+    src: /utils/apps/kafka/kafka_liveness_non_auth.j2
     dest: kafka_liveness.yml
+  when: kafka_sasl_auth is not defined or kafka_sasl_auth == '' or kafka_sasl_auth == 'disabled'
+
+- name: Generate the kafka liveness spec from template
+  template: 
+    src: /utils/apps/kafka/kafka_liveness_sasl_auth.j2
+    dest: kafka_liveness.yml
+  when: kafka_sasl_auth is defined and kafka_sasl_auth == 'enabled' 
 
 - name: Apply the pub-sub kafka liveness applicaton
   shell: 

utils/apps/kafka/kafka_select_broker.yml

@@ -1,7 +1,7 @@
 - name: select leader broker as per the liveness topic (partition) 
   set_fact: 
     kafka_broker: "{{ liveness_topic_leader }}"
-  when: kafka_stream is defined and kafka_stream != ''
+  when: kafka_stream is defined and kafka_stream == 'enabled'
 
 - name: allow random pod selection by chaosutil 
   set_fact: