Only interact with logstash configuration when Kafka pipeline is enabled otherwise leave it default

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

salt/kafka/defaults.yaml

@@ -3,6 +3,7 @@ kafka:
   cluster_id:
   controllers:
   reset:
+  logstash: []
   config:
     password:
     trustpass:

salt/kafka/soc_kafka.yaml

@@ -16,6 +16,12 @@ kafka:
     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
+  logstash:
+    description: By default logstash is disabled when Kafka is enabled. This option allows you to specify any hosts you would like to re-enable logstash on alongside Kafka.
+    forcedType: "[]string"
+    multiline: True
+    advanced: True
+    helpLink: kafka.html
   config:
     password:
       description: The password used for the Kafka certificates.

salt/logstash/defaults.yaml

@@ -35,7 +35,6 @@ logstash:
       - so/0900_input_redis.conf.jinja
       - so/9805_output_elastic_agent.conf.jinja
       - so/9900_output_endgame.conf.jinja
-      - so/0800_input_kafka.conf.jinja
     custom0: []
     custom1: []
     custom2: []

salt/logstash/init.sls

@@ -4,13 +4,9 @@
 # Elastic License 2.0.
 
 {% from 'logstash/map.jinja' import LOGSTASH_MERGED %}
-{% from 'kafka/map.jinja' import KAFKAMERGED %}
 
 include:
-{# Disable logstash when Kafka is enabled except when the role is standalone #}
-{% if LOGSTASH_MERGED.enabled and grains.role == 'so-standalone' %}
-  - logstash.enabled
-{% elif LOGSTASH_MERGED.enabled and not KAFKAMERGED.enabled %}
+{% if LOGSTASH_MERGED.enabled %}
   - logstash.enabled
 {% else %}
   - logstash.disabled

salt/logstash/map.jinja

@@ -6,6 +6,7 @@
 {% from 'vars/globals.map.jinja' import GLOBALS %}
 {% import_yaml 'logstash/defaults.yaml' as LOGSTASH_DEFAULTS %}
 {% set LOGSTASH_MERGED = salt['pillar.get']('logstash', LOGSTASH_DEFAULTS.logstash, merge=True) %}
+{% set KAFKA_LOGSTASH = salt['pillar.get']('kafka:logstash', []) %}
 
 {# used to store the redis nodes that logstash needs to know about to pull from the queue #}
 {% set LOGSTASH_REDIS_NODES = [] %}
@@ -30,3 +31,14 @@
 {%   endfor %}
 {% endfor %}
 
+{# Append Kafka input pipeline when Kafka is enabled #}
+{% if GLOBALS.pipeline == 'KAFKA' %}
+{%   do LOGSTASH_MERGED.defined_pipelines.search.append('so/0800_input_kafka.conf.jinja') %}
+{%   do LOGSTASH_MERGED.defined_pipelines.manager.append('so/0800_input_kafka.conf.jinja') %}
+{# Disable logstash on manager & receiver nodes unless it has an override configured #}
+{%   if not KAFKA_LOGSTASH %}
+{%     if GLOBALS.role in ['so-manager', 'so-receiver'] and GLOBALS.hostname not in KAFKA_LOGSTASH %}
+{%       do LOGSTASH_MERGED.update({'enabled': False}) %}
+{%     endif %}
+{%   endif %}
+{% endif %}