Remove logstash output to kafka pipeline. Add additional topics for searchnodes to ingest and add partition/offset info to event

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

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

@@ -11,13 +11,10 @@
 {% endfor %}
 {% set bootstrap_servers = "','".join(broker_ips) %}
 
-
-#Run on searchnodes ingest kafka topic(s) group_id allows load balancing of event ingest to all searchnodes
 input {
     kafka {
         codec => json
-        #Can ingest multiple topics. Set to a value from SOC UI?
-        topics => ['ea-logs']
+        topics => ['default-logs', 'kratos-logs', 'soc-logs', 'strelka-logs', 'suricata-logs', 'zeek-logs']
         group_id => 'searchnodes'
         client_id => '{{ GLOBALS.hostname }}'
         security_protocol => 'SSL'
@@ -26,7 +23,15 @@ input {
         ssl_keystore_password => 'changeit'
         ssl_keystore_type => 'PKCS12'
         ssl_truststore_location => '/etc/pki/ca-trust/extracted/java/cacerts'
-        # Set password as a pillar to avoid bad optics? This is default truststore for grid
         ssl_truststore_password => 'changeit'
+        decorate_events => true
+        tags => [ "elastic-agent", "input-{{ GLOBALS.hostname}}", "kafka" ]
     }
+}
+filter {
+  if ![metadata] {
+    mutate {
+      rename => { "@metadata" => "metadata" }
+    }
+  }
 }

salt/logstash/pipelines/config/so/0899_output_kafka.conf.jinja

@@ -1,22 +0,0 @@
-{% set kafka_brokers = salt['pillar.get']('logstash:nodes:kafkanode', {}) %}
-{% set broker_ips = [] %}
-{% for node, node_data in kafka_brokers.items() %}
-  {% do broker_ips.append(node_data['ip'] + ":9092") %}
-{% endfor %}
-
-{% set bootstrap_servers = "','".join(broker_ips) %}
-
-#Run on kafka broker logstash writes to topic 'logstash-topic'
-output {
-    kafka {
-        codec => json
-        topic_id => 'logstash-topic'
-        bootstrap_servers => '{{ bootstrap_servers }}'
-        security_protocol => 'SSL'
-        ssl_keystore_location => '/usr/share/logstash/kafka-logstash.p12'
-        ssl_keystore_password => ''
-        ssl_keystore_type => 'PKCS12'
-        ssl_truststore_location => '/etc/pki/ca-trust/extracted/java/cacerts'
-        ssl_truststore_password => 'changeit'
-    }
-}