求适用于Apache错误日志及系统日志的通用Logstash Grok模式

一、先搞定Apache错误日志的通用匹配 #

你之前碰到的grok_match_failure核心原因是两类日志结构不一致——一类带进程ID前缀[11446]，另一类没有；同时client字段也是可选存在的。我调整后的Grok模式可以完美兼容这两种场景：

首先保留你定义的基础时间模式：

APACHE_ERROR_TIME %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}

然后是通用的日志匹配规则，把可选的进程ID和client字段都用非捕获组(?:...)包裹，并标记为可选出现：

APACHE_ERROR_LOG (?:\[%{NUMBER:process_id}\] )?\[%{APACHE_ERROR_TIME:timestamp}\] \[%{LOGLEVEL:loglevel}\] (?:\[client %{IPORHOST:clientip}\] )?%{GREEDYDATA:errormsg}

这个模式能同时适配你给出的两类Apache错误日志：

• 带进程ID的：[11446] [Thu Jan 30 07:50:49 2020] [debug]: RT::Date used Time::ParseDate...
• 带client字段的：[Wed Jun 26 22:13:22 2013] [error] [client 10.10.10.100] PHP Fatal error...

另外别忘了在Logstash过滤器里给timestamp字段加上日期解析，确保时间格式能被正确识别：

filter {
  grok {
    match => { "message" => "%{APACHE_ERROR_LOG}" }
  }
  date {
    match => [ "timestamp", "EEE MMM dd HH:mm:ss yyyy" ]
    target => "@timestamp"
  }
}

二、适配两类系统日志的通用Grok模式 #

再看你新遇到的两类系统日志，它们的共同结构是时间 主机名 进程名: 日志内容，但日志内容的格式差异较大——一个是time="..." level=... msg="..." source="..."格式，另一个是[ISO时间][日志级别][组件] 日志信息格式。这里给你两个可选方案：

方案1：一次性通用Grok模式 #

先提取通用的日志前缀：

SYSLOG_PREFIX %{MONTH:month} %{MONTHDAY:day} %{TIME:time} %{HOSTNAME:hostname} %{DATA:process_name}:

然后针对两种日志内容，写一个包含可选分支的匹配规则：

GENERAL_SYSLOG %{SYSLOG_PREFIX} (?:time="%{TIMESTAMP_ISO8601:log_time}" level=%{LOGLEVEL:loglevel} msg="%{DATA:msg}" source="%{DATA:source}")|(?:\[%{TIMESTAMP_ISO8601:log_time}\]\[%{LOGLEVEL:loglevel}\s*\]\[%{DATA:component}\] %{GREEDYDATA:msg})

这个模式可以同时匹配你给出的两类系统日志，无需额外的条件判断。

方案2：Logstash条件分支（可读性更强） #

如果觉得上面的模式太冗长，也可以先用Grok提取公共前缀，再用条件分支匹配不同的日志内容格式，后续扩展新日志类型也更方便：

filter {
  grok {
    match => { "message" => "%{MONTH:month} %{MONTHDAY:day} %{TIME:time} %{HOSTNAME:hostname} %{DATA:process_name}: %{GREEDYDATA:log_content}" }
  }
  # 匹配prometheus_postgres_exporter风格的日志
  if [process_name] == "prometheus_postgres_exporter" {
    grok {
      match => { "log_content" => 'time="%{TIMESTAMP_ISO8601:log_time}" level=%{LOGLEVEL:loglevel} msg="%{DATA:msg}" source="%{DATA:source}"' }
    }
  }
  # 匹配logstash风格的日志
  else if [process_name] == "logstash" {
    grok {
      match => { "log_content" => '\[%{TIMESTAMP_ISO8601:log_time}\]\[%{LOGLEVEL:loglevel}\s*\]\[%{DATA:component}\] %{GREEDYDATA:msg}' }
    }
  }
  # 统一解析日志时间到@timestamp字段
  date {
    match => [ "log_time", "ISO8601" ]
    target => "@timestamp"
  }
}

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