Update Kafka self reset & add initial Kafka wrapper scripts to build out

Signed-off-by: reyesj2 <94730068+reyesj2@users.noreply.github.com>

salt/kafka/config.sls

@@ -20,14 +20,13 @@ kafka:
     - uid: 960
     - gid: 960
 
-{# Future tools to query kafka directly / show consumer groups
 kafka_sbin_tools:
   file.recurse:
     - name: /usr/sbin
     - source: salt://kafka/tools/sbin
     - user: 960
     - group: 960
-    - file_mode: 755 #}
+    - file_mode: 755
 
 kafka_sbin_jinja_tools:
   file.recurse:
@@ -69,7 +68,7 @@ kafka_kraft_{{sc}}_properties:
 reset_quorum_on_changes:
   cmd.run:
     - name: rm -f /nsm/kafka/data/__cluster_metadata-0/quorum-state
-    - watch:
+    - onchanges:
       - file: /opt/so/conf/kafka/server.properties
 
 {% else %}

salt/kafka/nodes.map.jinja

@@ -16,7 +16,7 @@
          tgt_type='compound') %}
 
 {% set STORED_KAFKANODES = salt['pillar.get']('kafka:nodes', default=None) %}
-{% set KAFKA_CONTROLLERS_PILLAR = salt['pillar.get']('kafka:kafka_controllers', default=None) %}
+{% set KAFKA_CONTROLLERS_PILLAR = salt['pillar.get']('kafka:controllers', default=None) %}
 
 {% set existing_ids = [] %}
 

salt/kafka/soc_kafka.yaml

@@ -17,7 +17,7 @@ kafka:
     forcedType: "string"
     helpLink: kafka.html
   reset:
-    description: Disable and reset the Kafka cluster. This will remove all Kafka data including logs that may have not yet been ingested into Elasticsearch and reverts the grid to using REDIS as the global pipeline. This is useful when testing different Kafka configurations such as rearranging Kafka brokers / controllers allowing you to reset the cluster rather than manually fixing any issues arising from attempting to reassign a Kafka broker into a controller. Enter 'YES_reset' and submit to disable and reset Kafka. Make any configuration changes required and re-enable Kafka when ready. This action CANNOT be reversed.
+    description: Disable and reset the Kafka cluster. This will remove all Kafka data including logs that may have not yet been ingested into Elasticsearch and reverts the grid to using REDIS as the global pipeline. This is useful when testing different Kafka configurations such as rearranging Kafka brokers / controllers allowing you to reset the cluster rather than manually fixing any issues arising from attempting to reassign a Kafka broker into a controller. Enter 'YES_RESET_KAFKA' and submit to disable and reset Kafka. Make any configuration changes required and re-enable Kafka when ready. This action CANNOT be reversed.
     advanced: True
     helpLink: kafka.html
   config:

salt/kafka/tools/sbin/so-kafka-cli

@@ -0,0 +1,47 @@
+#! /bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+if [ -z "$NOROOT" ]; then
+	# Check for prerequisites
+	if [ "$(id -u)" -ne 0 ]; then
+		echo "This script must be run using sudo!"
+		exit 1
+	fi
+fi
+
+function usage() {
+  echo -e "\nUsage: $0 <script> [options]"
+  echo ""
+  echo "Available scripts:"
+  show_available_kafka_cli_tools
+}
+
+function show_available_kafka_cli_tools(){
+  docker exec so-kafka ls /opt/kafka/bin | grep kafka
+}
+
+if [ -z $1 ]; then
+  usage
+  exit 1
+fi
+
+available_tools=$(show_available_kafka_cli_tools)
+script_exists=false
+
+for script in $available_tools; do
+  if [ "$script" == "$1" ]; then
+    script_exists=true
+    break
+  fi
+done
+
+if [ "$script_exists" == true ]; then
+  docker exec so-kafka /opt/kafka/bin/$1 "${@:2}"
+else
+  echo -e  "\nInvalid script: $1"
+  usage
+  exit 1
+fi

salt/kafka/tools/sbin/so-kafka-config-update

@@ -0,0 +1,87 @@
+#! /bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+if [ -z "$NOROOT" ]; then
+	# Check for prerequisites
+	if [ "$(id -u)" -ne 0 ]; then
+		echo "This script must be run using sudo!"
+		exit 1
+	fi
+fi
+
+usage() {
+  cat <<USAGE_EOF
+
+  Usage: $0 <operation> [parameters]
+
+  Where <operation> is one of the following:
+
+    topic-partitions: Increase the number of partitions for a Kafka topic
+      Required arguments: topic-partitions <topic name> <# partitions>
+      Example: $0 topic-partitions suricata-topic 6
+
+	list-topics: List of Kafka topics
+	  Example: $0 list-topics
+
+USAGE_EOF
+  exit 1
+}
+
+if [[ $# -lt 1 || $1 == --help || $1 == -h ]]; then
+  usage
+fi
+
+kafka_client_config="/opt/kafka/config/kraft/client.properties"
+
+too_few_arguments() {
+  echo -e "\nMissing one or more required arguments!\n"
+  usage
+}
+
+get_kafka_brokers() {
+	brokers_cache="/opt/so/state/kafka_brokers"
+  broker_port="9092"
+  if [[ ! -f "$brokers_cache" ]] || [[ $(find "/$brokers_cache" -mmin +120) ]]; then
+    echo "Refreshing Kafka brokers list"
+	  salt-call pillar.get kafka:nodes --out=json | jq -r --arg broker_port "$broker_port" '.local | to_entries[] | select(.value.role | contains("broker")) | "\(.value.ip):\($broker_port)"' | paste -sd "," - > "$brokers_cache"
+  else
+    echo "Using cached Kafka brokers list"
+  fi
+  brokers=$(cat "$brokers_cache")
+}
+
+increase_topic_partitions() {
+  get_kafka_brokers
+	command=$(so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --alter --topic $topic --partitions $partition_count)
+  if $command; then
+    echo -e "Successfully increased the number of partitions for topic $topic to $partition_count\n"
+    so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --describe --topic $topic
+  fi
+}
+
+get_kafka_topics_list() {
+	get_kafka_brokers
+	so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --exclude-internal --list | sort
+}
+
+operation=$1
+case "${operation}" in
+  "topic-partitions")
+	if [[ $# -lt 3 ]]; then
+	  too_few_arguments
+	fi
+	topic=$2
+	partition_count=$3
+	increase_topic_partitions
+	;;
+  "list-topics")
+	get_kafka_topics_list
+	;;
+  *)
+	usage
+	;;
+esac
+

salt/salt/files/engines.conf

@@ -88,7 +88,7 @@ engines:
                   - cmd.run:
                       cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' saltutil.kill_all_jobs
                   - cmd.run:
-                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.apply kafka.disabled,kafka.reset_kafka
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.apply kafka.disabled,kafka.reset
                   - cmd.run:
                       cmd: /usr/sbin/so-yaml.py remove /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls kafka.reset
       interval: 10