如何在ELK中提取PostgreSQL日志的duration字段值？

要拆分message字段中的duration值并生成独立可聚合字段，核心是在Logstash的filter阶段完成字段提取，以下是两种可行方案：

方案一：使用Grok插件（适配复杂日志格式） #

Grok适合匹配结构多变但有规律的日志，假设你的PostgreSQL日志格式类似：

2024-05-20 10:00:00 UTC [12345]: [1-1] user=postgres,db=test,app=psql,client=127.0.0.1 LOG: duration: 12.34 ms statement: SELECT * FROM users;

在Logstash配置文件的filter块中添加：

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{NUMBER:pid}:\] \[%{NUMBER:session_id}\] user=%{USER:user},db=%{USER:db},app=%{DATA:app},client=%{IP:client} LOG:  duration: %{NUMBER:duration:float} ms  statement: %{GREEDYDATA:statement}" }
    add_tag => ["postgresql_log"]
  }
}

• 该配置会将duration提取为float类型字段，支持后续聚合、排序操作
• 如果日志格式更简单，可简化grok规则，比如仅匹配duration片段：

  filter {
    grok {
      match => { "message" => "LOG:  duration: %{NUMBER:duration:float} ms" }
    }
  }
  

方案二：使用Dissect插件（高效适配固定格式日志） #

如果PostgreSQL日志中duration: X ms的位置固定，Dissect的性能比Grok更优，配置如下：

filter {
  dissect {
    mapping => { "message" => "%{*} duration: %{duration:float} ms %{*}" }
  }
}

• %{*}表示忽略该位置的任意内容，精准提取中间的duration值并转为float类型

验证与生效步骤 #

1. 测试Logstash配置：执行bin/logstash -f your_config.conf --config.test_and_exit，确保配置无语法错误
2. 重启Logstash服务，让新配置生效
3. 在Kibana中刷新对应索引的字段列表，之后即可在可视化、Discover中使用独立的duration字段进行聚合分析

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