如何通过Logstash/Filebeat解析历史Tomcat访问日志并按日志日期生成Elasticsearch索引

我来帮你一步步解决这两个核心问题：解析Tomcat日志提取目标字段，以及按日志文件日期自动生成Elasticsearch索引。下面分两种场景给出具体配置：

方案一：Logstash直接读取日志文件 #

1. 配置Filter解析日志字段 #

你的日志格式是典型的Tomcat访问日志（注意其中POST后的URL为空，我会按通用格式适配），推荐用grok插件来解析，它比mutate更适合结构化日志提取。在filter块中添加以下配置：

filter {
  # 第一步：用grok匹配日志格式，提取核心字段
  grok {
    match => { "message" => '%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:request_type} %{URIPATH:url} %{DATA:protocol}" %{NUMBER:response_code:int} %{NUMBER:bytes_returned:int}' }
    # 如果你的日志里确实没有URL（比如POST后面是空），可以把上面的%{URIPATH:url}换成%{DATA:url}，调整为：
    # match => { "message" => '%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:request_type} %{DATA:url} %{DATA:protocol}" %{NUMBER:response_code:int} %{NUMBER:bytes_returned:int}' }
    add_tag => ["tomcat_access_log"]
  }

  # 第二步：把提取的timestamp转为ES可识别的日期格式，覆盖@timestamp字段
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
    timezone => "Europe/Berlin" # 根据你的日志时区调整，这里对应+0100时区
  }

  # 可选：删除不需要的原始字段，精简数据
  mutate {
    remove_field => ["message", "timestamp"]
  }
}

2. 按日志文件日期生成动态索引 #

要实现localhost_access_log.2022-12-19.txt对应tomcat_access-2022.12.19索引，需要先从文件路径中提取日期，再在output中引用这个字段：

先在filter中添加提取日期的逻辑：

filter {
  # ... 保留上面的grok和date配置 ...

  # 从文件路径提取日期（匹配localhost_access_log.2022-12-19.txt格式）
  dissect {
    mapping => { "path" => "%{dir}/localhost_access_log.%{file_date}.txt" }
  }
  # 把file_date的格式从2022-12-19转为2022.12.19，适配索引命名习惯
  mutate {
    gsub => ["file_date", "-", "."]
  }
}

然后修改output的index配置：

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "tomcat_access-%{file_date}" # 用提取的file_date生成动态索引
    data_stream => "false"
  }
  stdout {codec => "rubydebug"}
}

方案二：Filebeat + Logstash组合方案 #

你当前的Filebeat Tomcat模块出现了dissect_parsing_error，说明默认的解析规则和你的日志格式不匹配。可以通过两种方式解决：

方法1：修改Filebeat Tomcat模块的解析规则 #

编辑Filebeat的Tomcat模块配置文件（通常在modules.d/tomcat.yml），自定义解析规则覆盖默认逻辑：

- module: tomcat
  log:
    enabled: true
    var.input: file
    var.paths:
      - /path/localhost_access_log.*.log
    # 添加自定义解析处理器，适配你的日志格式
    processors:
      - dissect:
          tokenizer: '%{client_ip} - - [%{timestamp}] "%{request_type} %{url} %{protocol}" %{response_code} %{bytes_returned}'
          field: "message"
          target_prefix: "tomcat"
      - date:
          field: "tomcat.timestamp"
          formats: ["dd/MMM/yyyy:HH:mm:ss Z"]
          target_field: "@timestamp"

方法2：在Logstash中添加解析逻辑（更灵活） #

保持Filebeat的配置不变，在Logstash的filter块中添加和方案一类似的解析规则：

filter {
  grok {
    match => { "event.original" => '%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:request_type} %{URIPATH:url} %{DATA:protocol}" %{NUMBER:response_code:int} %{NUMBER:bytes_returned:int}' }
    add_tag => ["tomcat_access_log"]
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
    timezone => "Europe/Berlin"
  }

  # 从log.file.path中提取日期
  dissect {
    mapping => { "log.file.path" => "%{dir}/localhost_access_log.%{file_date}.txt" }
  }
  mutate {
    gsub => ["file_date", "-", "."]
    remove_field => ["event.original", "timestamp"]
  }
}

调整Filebeat+Logstash的Output索引配置 #

同样在Logstash的output中使用动态索引：

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "tomcat_access-%{file_date}"
    data_stream => "false"
  }
  stdout {codec => "rubydebug"}
}

备注：内容来源于stack exchange，提问作者F.M.