ELK – winlogbeat dynamically splitting windows logs to various kafka topics

So in this next post I will cover an examples of how I can use winlogbeat, with kafka output in combination with a dynamic variable “fields” to direct the kafka output as I please.

So lets say you have a requirement to split windows logs to different kafka topics or even streams which ends up in indices with different security or ILM policies in Elastic.. well this solution might just work for you.

This example has 5 have kafka topics and 6 windows event logs which I want to split, thanks to the dynamic field I can direct them as required.

• logs-windows-application

• logs-windows-system

• logs-windows-security

• logs-windows-sysmon

• logs-windows-powershell

Now using a field called kafka_topic as a dynamic variable I direct it to topic: “%{[kafka_topic]}”, this way any logs coming in from those sources will be tagged with the specific field and processed accordingly with thanks to “%{[kafka_topic]}”

---
# Winlogbeat config

winlogbeat.event_logs:
  - name: Application
    fields:
      kafka_topic: logs-windows-application
    fields_under_root: true

  - name: System
    fields:
      kafka_topic: logs-windows-system
    fields_under_root: true

  - name: Security
    fields:
      kafka_topic: logs-windows-security
    fields_under_root: true

  - name: Microsoft-Windows-Sysmon/Operational
    fields:
      kafka_topic: logs-windows-sysmon
    fields_under_root: true

  - name: Windows PowerShell
    fields:
      kafka_topic: logs-windows-powershell
    fields_under_root: true

  - name: Microsoft-Windows-PowerShell/Operational
    fields:
      kafka_topic: logs-windows-powershell
    fields_under_root: true

# Outputs

output.kafka:
  hosts:
    - saveme-1.vman.ch:6969
    - saveme-2.vman.ch:6969
    - saveme-3.vman.ch:6969
  topic: "%{[kafka_topic]}"

  ssl.enabled: true

# Logging

# Sets log level. The default log level is info.
# Available log levels are: error, warning, info, debug
logging.level: warning

So this works well for unauthenticated kafka topcis, if your kafka topics need to use a username and password the account used must have access to write to all the topics in question.

---
# Winlogbeat config

winlogbeat.event_logs:
  - name: Application
    fields:
      kafka_topic: logs-windows-application
    fields_under_root: true

  - name: System
    fields:
      kafka_topic: logs-windows-system
    fields_under_root: true

  - name: Security
    fields:
      kafka_topic: logs-windows-security
    fields_under_root: true

  - name: Microsoft-Windows-Sysmon/Operational
    fields:
      kafka_topic: logs-windows-sysmon
    fields_under_root: true

  - name: Windows PowerShell
    fields:
      kafka_topic: logs-windows-powershell
    fields_under_root: true

  - name: Microsoft-Windows-PowerShell/Operational
    fields:
      kafka_topic: logs-windows-powershell
    fields_under_root: true

# Outputs

output.kafka:
  hosts:
    - saveme-1.vman.ch:6969
    - saveme-2.vman.ch:6969
    - saveme-3.vman.ch:6969
  topic: "%{[kafka_topic]}"
  username: superduperuser
  password: userduperuserpassword123
  ssl.enabled: true

# Logging

# Sets log level. The default log level is info.
# Available log levels are: error, warning, info, debug
logging.level: warning

While I know this exists in documentation, more specifically for filebeat i thought the community might appreicate a simple example.

Hope it was helpful

vMan