基于ELK栈，如何用Logstash处理单日志文件的三种格式日志？

Hey there! Let's walk through exactly how to handle your three distinct log formats using Logstash's conditional logic. First, we need to tackle the multiline stacktrace logs (your first format) at the Filebeat level—it's way more efficient than handling it in Logstash. Then we'll set up Logstash conditionals to parse each format correctly.

Step 1: Configure Filebeat for Multiline Logs #

Your first log type has stacktraces that span multiple lines, so we need to tell Filebeat to merge those lines into a single event. Add this to your filebeat.yml:

filebeat.inputs:
- type: log
  paths:
    - /path/to/your/target/logs/*.log
  # Match lines that DON'T start with your timestamp pattern (these are stacktrace lines)
  multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
  multiline.negate: true
  multiline.match: after
  # Optional: If your stacktraces include "Caused by:" lines, update the pattern to catch those too
  # multiline.pattern: '^(?:\s+at|\s+Caused by:|^\d{4}-\d{2}-\d{2})'

This setup merges any line that doesn't start with your timestamp (like stacktrace lines) into the previous timestamped line, creating one complete event for the entire error log.

Step 2: Logstash Configuration with Conditionals #

Now let's set up Logstash to parse each log format using if/else if logic. Here's a complete logstash.conf example tailored to your use case:

input {
  beats {
    port => 5044 # Default port Filebeat uses to send logs to Logstash
  }
}

filter {
  # First, parse the timestamp from all log types (since they all start with a date)
  grok {
    match => { "message" => "^%{TIMESTAMP_ISO8601:log_timestamp} " }
    add_tag => ["date_parsed"]
  }

  # Convert the parsed timestamp to Logstash's @timestamp field
  date {
    match => ["log_timestamp", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
    remove_field => ["log_timestamp"] # Clean up the temporary field
  }

  # Handle Format 1: Date + Params + JSON + Stacktrace (multiline)
  if [message] =~ /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}.*\{.*\}\s+at / {
    # Extract params, JSON, and stacktrace using grok
    grok {
      match => { "message" => "^%{TIMESTAMP_ISO8601} %{DATA:request_params} %{GREEDYDATA:error_json}\s+%{GREEDYDATA:stacktrace}" }
    }
    # Parse the JSON part into structured fields
    json {
      source => "error_json"
      target => "error_details"
      remove_field => ["error_json"] # Clean up after parsing
    }
    add_tag => ["multiline_error_log"] # Tag for easy filtering in Kibana
  }
  # Handle Format 2: Date + GET/POST + Text Content
  else if [message] =~ /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} (GET|POST) / {
    grok {
      match => { "message" => "^%{TIMESTAMP_ISO8601} %{WORD:request_method} %{GREEDYDATA:request_content}" }
    }
    add_tag => ["http_request_log"]
  }
  # Handle Format 3: Date + Module Name (e.g., paymentAdmin) + JSON
  else if [message] =~ /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} %{WORD:module_name} \{/ {
    grok {
      match => { "message" => "^%{TIMESTAMP_ISO8601} %{WORD:module_name} %{GREEDYDATA:module_json}" }
    }
    # Parse the module-specific JSON
    json {
      source => "module_json"
      target => "module_details"
      remove_field => ["module_json"]
    }
    add_tag => ["module_specific_log"]
  }
  # Catch-all for unrecognized logs (helps with debugging)
  else {
    add_tag => ["unmatched_log"]
  }
}

output {
  # Send parsed logs to Elasticsearch
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "application-logs-%{+YYYY.MM.dd}" # Rotate index daily
  }
  # Optional: Print parsed logs to console for debugging
  stdout { codec => rubydebug }
}

Key Notes to Customize for Your Logs #

• Adjust Grok Patterns: The regex in the grok filters is tailored to your description, but you'll need to tweak it to match your actual log syntax. For example, if your "params" field has specific delimiters, replace %{DATA:request_params} with a more precise pattern.
• Test Regex: Use Logstash's built-in grok debugger (run logstash -f your.conf --config.test_and_exit) or local regex tools to refine your patterns until they match all your log lines.
• Tagging: The tags we added (multiline_error_log, http_request_log, etc.) let you quickly filter and visualize different log types in Kibana.

内容的提问来源于stack exchange，提问作者user84592