Setup kafka reactor for managing kafka controllers globally

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

salt/kafka/controllers.sls

@@ -0,0 +1,20 @@
+# 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.
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+{%   import_yaml 'kafka/defaults.yaml' as KAFKADEFAULTS %}
+
+{% set process_x_roles = salt['pillar.get']('kafka:config:server:process_x_roles', KAFKADEFAULTS.kafka.config.server.process_x_roles, merge=true) %}
+
+{# Send an event to the salt master at every highstate. Containing the minions process_x_roles.
+    if no value is set for this minion then the default in kafka/defaults.yaml is used #}
+push_event_to_master:
+  event.send:
+    - name: kafka/controllers_update
+    - data:
+        id: {{ grains['id'] }}
+        process_x_roles: {{ process_x_roles }}
+{% endif %}

salt/kafka/enabled.sls

@@ -13,6 +13,7 @@ include:
   {% if grains.role in ['so-manager', 'so-managersearch', 'so-standalone'] %}
   - kafka.nodes
   {% endif %}
+  - kafka.controllers
   - elasticsearch.ca
   - kafka.sostatus
   - kafka.config

salt/reactor/kafka.sls

@@ -0,0 +1,16 @@
+{# 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. #}
+
+{% set minionid = data['id'].split('_')[0] %}
+{% set role = data['data']['process_x_roles'] %}
+
+{# Run so-yaml to replace kafka.node.<minionID>.role with the value from kafka/controllers.sls #}
+
+update_global_kafka_pillar:
+  local.cmd.run:
+    - tgt: 'G@role:so-manager or G@role:so-managersearch or G@role:so-standalone'
+    - tgt_type: compound
+    - arg:
+      - '/usr/sbin/so-yaml.py replace /opt/so/saltstack/local/pillar/kafka/nodes.sls kafka.nodes.{{ minionid }}.role {{ role }}'

salt/salt/files/reactor.conf

@@ -0,0 +1,3 @@
+reactor:
+  - 'kafka/controllers_update':
+    - salt://reactor/kafka.sls

salt/salt/master.sls

@@ -32,6 +32,11 @@ engines_config:
     - name: /etc/salt/master.d/engines.conf
     - source: salt://salt/files/engines.conf
 
+reactor_config:
+  file.managed:
+    - name: /etc/salt/master.d/reactor.conf
+    - source: salt://salt/files/reactor.conf
+
 salt_master_service:
   service.running:
     - name: salt-master