From 25d63f751676b9fc708d04b4e4a5f60dc0e392e2 Mon Sep 17 00:00:00 2001
From: reyesj2 <94730068+reyesj2@users.noreply.github.com>
Date: Mon, 22 Apr 2024 16:42:59 -0400
Subject: [PATCH] Setup kafka reactor for managing kafka controllers globally

Signed-off-by: reyesj2 <94730068+reyesj2@users.noreply.github.com>
---
 salt/kafka/controllers.sls   | 20 ++++++++++++++++++++
 salt/kafka/enabled.sls       |  1 +
 salt/reactor/kafka.sls       | 16 ++++++++++++++++
 salt/salt/files/reactor.conf |  3 +++
 salt/salt/master.sls         |  5 +++++
 5 files changed, 45 insertions(+)
 create mode 100644 salt/kafka/controllers.sls
 create mode 100644 salt/reactor/kafka.sls
 create mode 100644 salt/salt/files/reactor.conf

diff --git a/salt/kafka/controllers.sls b/salt/kafka/controllers.sls
new file mode 100644
index 000000000..c6df07b0c
--- /dev/null
+++ b/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 %}
diff --git a/salt/kafka/enabled.sls b/salt/kafka/enabled.sls
index 3c4f548f1..d05a49a0e 100644
--- a/salt/kafka/enabled.sls
+++ b/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
diff --git a/salt/reactor/kafka.sls b/salt/reactor/kafka.sls
new file mode 100644
index 000000000..879fb5431
--- /dev/null
+++ b/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 }}'
\ No newline at end of file
diff --git a/salt/salt/files/reactor.conf b/salt/salt/files/reactor.conf
new file mode 100644
index 000000000..129305572
--- /dev/null
+++ b/salt/salt/files/reactor.conf
@@ -0,0 +1,3 @@
+reactor:
+  - 'kafka/controllers_update':
+    - salt://reactor/kafka.sls
\ No newline at end of file
diff --git a/salt/salt/master.sls b/salt/salt/master.sls
index 0a65f3e01..76340fb3d 100644
--- a/salt/salt/master.sls
+++ b/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