Kafka upgrade (#2871) (#2803) (#2990)

Kafka: 
* upgrade to 2.8.1 version
* move upgrade tasks to Kafka role
* refactor installation and upgrade tasks
* refactor: rename 'kafka_var' setting

Zookeeper:
* refactor installation and upgrade
* move upgrade tasks to Zookeeper role

Kafka-exporter:
* verify if service definition needs to be updated even if exporter itself was not updated
* remove `kafka.version` from configuration

ansible/playbooks/kafka.yml

@@ -2,8 +2,8 @@
 # Ansible playbook that makes sure the base items for all nodes are installed
 
 - hosts: all
-  gather_facts: yes
-  tasks: [ ]
+  gather_facts: true
+  tasks: []
 
 - hosts: kafka
   become: true

ansible/playbooks/roles/kafka/defaults/main.yml

@@ -1,4 +1,6 @@
-kafka_version: 2.6.0
+kafka_version: 2.8.1
 scala_version: 2.12
-kafka_bin_filename: "kafka_2.12-2.6.0.tgz"
+kafka_bin_filename: "kafka_2.12-2.8.1.tgz"
+kafka_install_dir: "/opt/kafka_{{ scala_version }}-{{ kafka_version }}"
+
 prometheus_jmx_exporter_path: /opt/jmx-exporter/jmx_prometheus_javaagent.jar

ansible/playbooks/roles/kafka/files/jmx-kafka-config.yml

@@ -85,4 +85,4 @@ rules:
   name: kafka_$1_$2_$3
   type: GAUGE
   labels:
-    quantile: "0.$4"
+    quantile: "0.$4"

ansible/playbooks/roles/kafka/handlers/main.yml

@@ -1,20 +1,22 @@
 ---
 # Handlers for Kafka
 
-- name: restart kafka
+- name: Restart kafka
   service:
     name: kafka
     state: restarted
-    enabled: yes
+    enabled: true
   retries: 10
   delay: 10
 
-- name: restart prometheus
+- name: Restart prometheus
   become: true
   systemd:
     daemon_reload: true
     name: prometheus
     state: restarted
-  delegate_to: "{{ item }}"
-  with_inventory_hostnames:
-    - prometheus
+  delegate_to: "{{ node }}"
+  loop_control:
+    loop_var: node
+  loop: "{{ groups.prometheus }}"
+  when: groups.prometheus is defined

ansible/playbooks/roles/kafka/tasks/common/download_and_unpack_binary.yml

@@ -0,0 +1,22 @@
+---
+- name: Download Kafka binaries
+  include_role:
+    name: download
+    tasks_from: download_file
+  vars:
+    file_name: "{{ kafka_bin_filename }}"
+
+- name: Uncompress the Kafka tar
+  unarchive:
+    remote_src: true
+    creates: "{{ kafka_install_dir }}"
+    src: "{{ download_directory }}/{{ kafka_bin_filename }}"
+    dest: /opt
+
+- name: Change ownership on Kafka directory
+  file:
+    path: "{{ kafka_install_dir }}"
+    state: directory
+    mode: u=rwx,go=rx
+    owner: "{{ specification.user }}"
+    group: "{{ specification.group }}"

ansible/playbooks/roles/kafka/tasks/start.yml → ansible/playbooks/roles/kafka/tasks/common/start.yml

@@ -1,10 +1,10 @@
 ---
-
 - name: Enable and Start Kafka
   service:
     name: kafka
     state: started
-    enabled: yes
+    enabled: true
+    daemon-reload: true
 
 # - name: wait for kafka port
 #   wait_for: host={{kafka.listen_address| default('localhost')}} port={{kafka.port}} state=started timeout={{ kafka.wait_for_period }}

ansible/playbooks/roles/kafka/tasks/common/stop.yml

@@ -0,0 +1,5 @@
+---
+- name: Stop Kafka
+  systemd:
+    name: kafka
+    state: stopped

ansible/playbooks/roles/kafka/tasks/generate-certificates.yml

@@ -1,56 +1,63 @@
 - name: Create stores directory
   file:
-    path: "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}"
+    path: "{{ specification.security.ssl.server.keystore_location | dirname }}"
     state: directory
-    owner: "{{ specification.kafka_var.user }}"
-    group: "{{ specification.kafka_var.group }}"
-    mode: "0755"
+    owner: "{{ specification.user }}"
+    group: "{{ specification.group }}"
+    mode: u=rwx,go=rx
 
 - name: Check if keystore exists on broker
   stat:
-    path: "{{ specification.kafka_var.security.ssl.server.keystore_location }}"
+    path: "{{ specification.security.ssl.server.keystore_location }}"
+    get_attributes: false
+    get_checksum: false
+    get_mime: false
   changed_when: false
   register: keystore_exists
 
 - name: Generate keystore for each server
-  shell: keytool -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} \
-         -alias localhost -validity {{ specification.kafka_var.security.ssl.server.cert_validity }} -genkey -keyalg RSA \
-         -noprompt -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }} \
-         -keypass {{ specification.kafka_var.security.ssl.server.passwords.key }} \
-         -dname "CN={{ inventory_hostname }}" -ext SAN="DNS:{{ inventory_hostname }}"
+  command: keytool -keystore {{ specification.security.ssl.server.keystore_location }} \
+           -alias localhost -validity {{ specification.security.ssl.server.cert_validity }} -genkey -keyalg RSA \
+           -noprompt -storepass {{ specification.security.ssl.server.passwords.keystore }} \
+           -keypass {{ specification.security.ssl.server.passwords.key }} \
+           -dname "CN={{ inventory_hostname }}" -ext SAN="DNS:{{ inventory_hostname }}"
   when:
     - not keystore_exists.stat.exists
 
 - name: Check if signing certificate exists
   stat:
-    path: "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-cert"
+    path: "{{ specification.security.ssl.server.keystore_location | dirname }}/ca-cert"
+    get_attributes: false
+    get_checksum: false
+    get_mime: false
   register: signing_certificate_exists
   changed_when: false
   when:
     - groups['kafka'][0] == inventory_hostname
 
 - name: Generate signing certificate
-  shell: openssl req -new -x509 -keyout {{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-key \
-         -out {{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-cert \
-         -days {{ specification.kafka_var.security.ssl.server.cert_validity }} \
-         -subj "/CN={{ inventory_hostname }}" \
-         --passout pass:{{ specification.kafka_var.security.ssl.server.passwords.key }}
+  command: openssl req -new -x509 -keyout {{ specification.security.ssl.server.keystore_location | dirname }}/ca-key \
+           -out {{ specification.security.ssl.server.keystore_location | dirname }}/ca-cert \
+           -days {{ specification.security.ssl.server.cert_validity }} \
+           -subj "/CN={{ inventory_hostname }}" \
+           --passout pass:{{ specification.security.ssl.server.passwords.key }}
   when:
     - groups['kafka'][0] == inventory_hostname
     - not signing_certificate_exists.stat.exists
 
 - name: Create kafka certificates directory on Epiphany host
   become: false
   file:
-    path: "{{ specification.kafka_var.security.ssl.server.local_cert_download_path }}"
+    path: "{{ specification.security.ssl.server.local_cert_download_path }}"
     state: directory
+    mode: u=rwx,go=
   delegate_to: localhost
 
 - name: Fetching files
-  fetch:
-    src: "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/{{ item }}"
-    dest: "{{ specification.kafka_var.security.ssl.server.local_cert_download_path }}/{{ item }}"
-    flat: yes
+  slurp:
+    src: "{{ specification.security.ssl.server.keystore_location | dirname }}/{{ item }}"
+    dest: "{{ specification.security.ssl.server.local_cert_download_path }}/{{ item }}"
+    flat: true
   loop:
     - "ca-cert"
     - "ca-key"
@@ -59,8 +66,9 @@
 
 - name: Copy signing certificate and key to brokers
   copy:
-     src: "{{ specification.kafka_var.security.ssl.server.local_cert_download_path }}/{{ item }}"
-     dest: "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/"
+    src: "{{ specification.security.ssl.server.local_cert_download_path }}/{{ item }}"
+    dest: "{{ specification.security.ssl.server.keystore_location | dirname }}/"
+    mode: preserve
   loop:
     - "ca-cert"
     - "ca-key"
@@ -69,71 +77,74 @@
 
 - name: Check if trustore exists
   stat:
-    path: "{{ specification.kafka_var.security.ssl.server.truststore_location }}"
+    path: "{{ specification.security.ssl.server.truststore_location }}"
+    get_attributes: false
+    get_checksum: false
+    get_mime: false
   register: trustore_exists
 
 - name: Create trustore
-  shell: keytool -noprompt -keystore "{{ specification.kafka_var.security.ssl.server.truststore_location }}" -alias CARoot \
-         -import -file "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-cert" \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }} \
-         -keypass {{ specification.kafka_var.security.ssl.server.passwords.key }}
+  command: keytool -noprompt -keystore "{{ specification.security.ssl.server.truststore_location }}" -alias CARoot \
+           -import -file "{{ specification.security.ssl.server.keystore_location | dirname }}/ca-cert" \
+           -storepass {{ specification.security.ssl.server.passwords.keystore }} \
+           -keypass {{ specification.security.ssl.server.passwords.key }}
   when:
     - not trustore_exists.stat.exists
 
 - name: Check if CA certificate is already imported
-  shell: keytool -list -v -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }} \
+  shell: set -o pipefail && keytool -list -v -keystore {{ specification.security.ssl.server.keystore_location }} \
+         -storepass {{ specification.security.ssl.server.passwords.keystore }} \
          | grep -i "Alias name" | grep -i "caroot"
   failed_when: "caroot_exists.rc == 2"
   changed_when: false
   register: caroot_exists
 
 - name: Check if certificate signed by CA is already imported
   shell: |-
-         keytool -list -v -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }} \
+         set -o pipefail && keytool -list -v -keystore {{ specification.security.ssl.server.keystore_location }} \
+         -storepass {{ specification.security.ssl.server.passwords.keystore }} \
          -alias localhost \
          | grep -i 'Certificate chain length: 2'
   failed_when: "signed_cert_exists.rc == 2"
   changed_when: false
   register: signed_cert_exists
 
 - name: Export certificate to sign certificate with CA
-  shell: keytool -noprompt -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} \
-         -alias localhost -certreq \
-         -file "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/cert-file" \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }} \
-         -keypass {{ specification.kafka_var.security.ssl.server.passwords.key }}
+  command: keytool -noprompt -keystore {{ specification.security.ssl.server.keystore_location }} \
+           -alias localhost -certreq \
+           -file "{{ specification.security.ssl.server.keystore_location | dirname }}/cert-file" \
+           -storepass {{ specification.security.ssl.server.passwords.keystore }} \
+           -keypass {{ specification.security.ssl.server.passwords.key }}
   when:
     - signed_cert_exists.rc == 1
 
 - name: Signing certificate with CA
-  shell: openssl x509 -req -CA "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-cert" \
-         -CAkey "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-key" \
-         -in "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/cert-file" \
-         -out "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/cert-signed" \
-         -days {{ specification.kafka_var.security.ssl.server.cert_validity }} -CAcreateserial \
-         -passin pass:{{ specification.kafka_var.security.ssl.server.passwords.key }}
+  command: openssl x509 -req -CA "{{ specification.security.ssl.server.keystore_location | dirname }}/ca-cert" \
+           -CAkey "{{ specification.security.ssl.server.keystore_location | dirname }}/ca-key" \
+           -in "{{ specification.security.ssl.server.keystore_location | dirname }}/cert-file" \
+           -out "{{ specification.security.ssl.server.keystore_location | dirname }}/cert-signed" \
+           -days {{ specification.security.ssl.server.cert_validity }} -CAcreateserial \
+           -passin pass:{{ specification.security.ssl.server.passwords.key }}
   when:
     - signed_cert_exists.rc == 1
 
 - name: Import certificate CA
-  shell: keytool -noprompt -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} -alias CARoot \
-         -import -file "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/ca-cert" \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }}
+  command: keytool -noprompt -keystore {{ specification.security.ssl.server.keystore_location }} -alias CARoot \
+           -import -file "{{ specification.security.ssl.server.keystore_location | dirname }}/ca-cert" \
+           -storepass {{ specification.security.ssl.server.passwords.keystore }}
   when:
     - caroot_exists.rc == 1
 
 - name: Import certificate signed by CA
-  shell: keytool -noprompt -keystore {{ specification.kafka_var.security.ssl.server.keystore_location }} -alias localhost \
-         -import -file "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/cert-signed" \
-         -storepass {{ specification.kafka_var.security.ssl.server.passwords.keystore }}
+  command: keytool -noprompt -keystore {{ specification.security.ssl.server.keystore_location }} -alias localhost \
+           -import -file "{{ specification.security.ssl.server.keystore_location | dirname }}/cert-signed" \
+           -storepass {{ specification.security.ssl.server.passwords.keystore }}
   when:
     - signed_cert_exists.rc == 1
 
 - name: Remove extracted key and cert from others than root node
   file:
-    path: "{{ specification.kafka_var.security.ssl.server.keystore_location | dirname }}/{{ item }}"
+    path: "{{ specification.security.ssl.server.keystore_location | dirname }}/{{ item }}"
     state: absent
   loop:
     - "ca-cert"

ansible/playbooks/roles/kafka/tasks/main.yml

@@ -3,6 +3,9 @@
 
 - name: Check if jmx exporter is available
   stat:
+    get_attributes: false
+    get_checksum: false
+    get_mime: false
     path: "{{ prometheus_jmx_exporter_path }}"
   register: exporter
 
@@ -13,5 +16,4 @@
 - include_tasks: metrics.yml
   when: exporter.stat.exists
 
-- include_tasks: start.yml
-
+- include_tasks: common/start.yml