merge dev

salt/logstash/defaults.yaml

@@ -35,6 +35,7 @@ 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/enabled.sls

@@ -75,10 +75,13 @@ so-logstash:
       {% else %}
       - /etc/pki/tls/certs/intca.crt:/usr/share/filebeat/ca.crt:ro
       {% endif %}
-      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-import', 'so-heavynode', 'so-searchnode'] %}
+      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-import', 'so-heavynode', 'so-searchnode' ] %}
       - /opt/so/conf/ca/cacerts:/etc/pki/ca-trust/extracted/java/cacerts:ro
       - /opt/so/conf/ca/tls-ca-bundle.pem:/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem:ro
       {% endif %}
+      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-searchnode'] %}
+      - /etc/pki/kafka-logstash.p12:/usr/share/logstash/kafka-logstash.p12:ro
+      {% endif %}
       {% if GLOBALS.role == 'so-eval' %}
       - /nsm/zeek:/nsm/zeek:ro
       - /nsm/suricata:/suricata:ro

salt/logstash/init.sls

@@ -4,9 +4,13 @@
 # Elastic License 2.0.
 
 {% from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+{% from 'kafka/map.jinja' import KAFKAMERGED %}
 
 include:
-{% if LOGSTASH_MERGED.enabled %}
+{# 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 %}
   - logstash.enabled
 {% else %}
   - logstash.disabled

salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja

@@ -0,0 +1,38 @@
+{%- set kafka_password = salt['pillar.get']('kafka:password') %}
+{%- set kafka_brokers = salt['pillar.get']('kafka:nodes', {}) %}
+{%- set brokers = [] %}
+
+{%- if kafka_brokers %}
+{%-   for key, values in kafka_brokers.items() %}
+{%-     if 'broker' in values['role'] %}
+{%-       do brokers.append(key ~ ':9092') %}
+{%-     endif %}
+{%-   endfor %}
+{%-   set bootstrap_servers = ','.join(brokers) %}
+
+input {
+    kafka {
+        codec => json
+        topics_pattern => '.*-securityonion$'
+        group_id => 'searchnodes'
+        consumer_threads => 3
+        client_id => '{{ GLOBALS.hostname }}'
+        security_protocol => 'SSL'
+        bootstrap_servers => '{{ bootstrap_servers }}'
+        ssl_keystore_location => '/usr/share/logstash/kafka-logstash.p12'
+        ssl_keystore_password => '{{ kafka_password }}'
+        ssl_keystore_type => 'PKCS12'
+        ssl_truststore_location => '/etc/pki/ca-trust/extracted/java/cacerts'
+        ssl_truststore_password => 'changeit'
+        decorate_events => true
+        tags => [ "elastic-agent", "input-{{ GLOBALS.hostname}}", "kafka" ]
+    }
+}
+filter {
+  if ![metadata] {
+    mutate {
+      rename => { "@metadata" => "metadata" }
+    }
+  }
+}
+{% endif %}