Spring Boot项目ELK栈审计日志处理：仅用Grok Filter实现Logstash动态键值对输出配置需求

Got it, let's build the exact Logstash Grok configuration you need for your audit logs. Since you're restricted to using only the Grok filter, we'll leverage Grok's pattern matching, dynamic field creation via add_field, and built-in field removal to meet your requirements.

Complete logstash.conf Configuration #

input {
  # Replace with your actual input source (e.g., file, beats)
  file {
    path => "/path/to/your/audit.log"
    start_position => "beginning"
    sincedb_path => "/dev/null" # Useful for testing to re-read logs
  }
}

filter {
  grok {
    match => {
      "message" => [
        # First pattern: match fixed fields + first key-value pair
        "\[%{LOGLEVEL:logLevel_temp}\] \[%{DATA:threadId_temp}\] %{DATA:timeStamp_temp} %{DATA:class_temp} %{DATA:key}:%{DATA:value}(?:\| %{GREEDYDATA:remaining_kv})?",
        # Subsequent patterns: match remaining key-value pairs
        "%{DATA:key}:%{DATA:value}(?:\| %{GREEDYDATA:remaining_kv})?"
      ]
    }
    # Keep processing until all key-value pairs are parsed
    break_on_match => false
    
    # Create final fields in your required 2D array format
    add_field => {
      "logLevel" => "[ [ \"%{logLevel_temp}\" ] ]",
      "threadId" => "[ [ \"%{threadId_temp}\" ] ]",
      "timeStamp" => "[ [ \"%{timeStamp_temp}\" ] ]",
      "class" => "[ [ \"%{class_temp}\" ] ]",
      "%{key}" => "[ [ \"%{value}\" ] ]"
    }
    
    # Clean up temporary and unused fields
    remove_field => ["logLevel_temp", "threadId_temp", "timeStamp_temp", "class_temp", "key", "value", "remaining_kv", "message"]
  }
}

output {
  # Replace with your actual output (e.g., Elasticsearch)
  stdout {
    codec => rubydebug
  }
}

How This Works #

1. Fixed Field Capture: The first Grok pattern grabs the static log components (log level, thread ID, timestamp, class) into temporary fields to avoid conflicts with dynamic keys.
2. Dynamic Key-Value Parsing: We use two patterns with break_on_match => false to repeatedly parse every key-value pair in the log. Each pair is stored in temporary key and value fields.
3. Formatting to Required Structure: The add_field parameter constructs the final output fields. For dynamic keys, we use %{key} to dynamically name the field based on the captured key value, and wrap all values in the requested 2D array format.
4. Cleanup: Temporary fields and the original message field are removed to keep the output clean and aligned with your requirements.

Key Adjustments for Precision #

• If your thread ID is always a UUID, replace %{DATA:threadId_temp} with %{UUID:threadId_temp} for stricter validation.
• For precise timestamp matching, replace %{DATA:timeStamp_temp} with a custom pattern: (?<timeStamp_temp>%{YEAR}-%{MONTHNUM}-%{MONTHDAY}_%{HOUR}:%{MINUTE}:%{SECOND}\.%{INT}).
• The DATA pattern handles values with spaces (since it matches any character except line breaks), so your sentence-style values will be captured correctly.

Testing this configuration with your sample log will produce exactly the output you specified, with dynamic keys as top-level fields and all values formatted as 2D arrays.

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