From d67ebabc951cfaf226b176f6eb458ec2b4c35127 Mon Sep 17 00:00:00 2001
From: reyesj2 <94730068+reyesj2@users.noreply.github.com>
Date: Mon, 8 Apr 2024 16:38:03 -0400
Subject: [PATCH] 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>
---
 .../config/so/0800_input_kafka.conf.jinja     | 15 ++++++++-----
 .../config/so/0899_output_kafka.conf.jinja    | 22 -------------------
 2 files changed, 10 insertions(+), 27 deletions(-)
 delete mode 100644 salt/logstash/pipelines/config/so/0899_output_kafka.conf.jinja

diff --git a/salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja b/salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja
index 0260b774e..1391ce983 100644
--- a/salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja
+++ b/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" }
+    }
+  }
 }
\ No newline at end of file
diff --git a/salt/logstash/pipelines/config/so/0899_output_kafka.conf.jinja b/salt/logstash/pipelines/config/so/0899_output_kafka.conf.jinja
deleted file mode 100644
index ff9a6f6ee..000000000
--- a/salt/logstash/pipelines/config/so/0899_output_kafka.conf.jinja
+++ /dev/null
@@ -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'
-    }
-}
\ No newline at end of file