elasticsearch - 使用 filebeat 多行时重复的消息内容

Question

我正在使用 filebeat 将应用程序日志发送到 elasticsearch，我的应用程序由 java (log4j) 编写我的日志存储在 elasticsearch 上，但我看到消息字段中的内容是重复的。示例：我的日志文件中的内容：

2021-07-25 09:52:55,154 INFO  c.m.v.w.r.QrService_v_4_0_0 - [210725104196] Payment notify to client {
  "resCode" : "00",
  "resDesc" : "Giao dịch thành công",
  "qrTrace" : "210725104196"
}

elasticsearch 上的内容： img1

2021-07-25 09:52:55,154 INFO  c.m.v.w.r.QrService_v_4_0_0 - [210725104196] Payment notify to client {
  "resCode" : "00",
  "resDesc" : "Giao dịch thành công",
  "qrTrace" : "210725104196"
},
[210725104196] Payment notify to client {
  "resCode" : "00",
  "resDesc" : "Giao dịch thành công",
  "qrTrace" : "210725104196"
}

FileBeat 创建重复消息的原因可能是什么？有我的filebeat.yml：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /home/Payment/logs/qrcode_api*.log
  fields:
     service: payapp
     app_id: qrcode-api
  multiline.pattern: '^202[0-9]\-[0-9][0-9]\-[0-9][0-9] [0-9][0-9]\:[0-9][0-9]\:[0-9][0-9],[0-9][0-9][0-9] [A-Z]+'
  multiline.negate: true
  multiline.match: after
  max_lines: 5000
  timeout: 60s
filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: false
output.redis:
  hosts: ["10.144.139.116"]
  key: "elasticsearchqueue"

# filebeat version
#filebeat version 7.9.2 (amd64), libbeat 7.9.2 [2ab907f5ccecf9fd82fe37105082e89fd871f684 built 2020-09-22 23:19:45 +0000 UTC]

我的日志：

input {
  redis {
    host => "10.144.139.116"
    key => "elasticsearchqueue"
    data_type => "list"
    add_field => { "inputsource" => "redis1" }
  }
}

filter {
    if [inputsource] == "redis1"  {
        if [fields][service] == "payapp" and [fields][app_id] == "qrcode-api" {
            grok {
                match => {
                    "message" => [
                        "(?<tmptime>20[0-9]{2}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2},[0-9]{3}) %{LOGLEVEL:loglevel} (?<category>[A-Za-z0-9$_.]+) \[%{NOTSPACE:thread}\]:%{NONNEGINT:line} %{GREEDYDATA:message}$",
                        "(?<tmptime>%{TIMESTAMP_ISO8601}) * %{LOGLEVEL:loglevel} * (?<logger>[A-Za-z0-9$_.]+) - %{GREEDYDATA:message}$"
                    ]
                }
            }
            mutate {
                add_field => { "logdate" => "%{[tmptime]}" }
                
                #add_field => { "created_time" => "%{[@timestamp]}" }
                remove_field => [ "[host][architecture]","[host][containerized]","[host][id]","[host][ip]","[host][mac]","[host][os][family]","[host][os][kernel]","[host][os][name]","[host][os][platform]","[host][os][version]","[log][offset]" , "[host][name]","[agent][type]","[agent][id]","[agent][ephemeral_id]","[ecs][version]","[fields][max_bytes]","[agent][version]","[logResponse]","[category]","[ErrorDesc]","[ErrorCode]","[loglevel]","[requestId]","[thread]","[line]" ]
            }
            date {
                match => ["tmptime", "yyyy-MM-dd HH:mm:ss,SSS", "ISO8601" ]
                timezone => "Asia/Ho_Chi_Minh"
                target => "@timestamp"
                remove_field => ["tmptime"]
            }
            ruby {
            code => "event.set('[index_day]', event.get('@timestamp').time.localtime('+07:00').strftime('%Y.%m.%d'))"
            }
        }
    }
}


output {
  if [fields][service] == "payapp" {
  elasticsearch {
    hosts => ["https://172.16.26.50:9200"]
       index => "%{[fields][app_id]}_%{index_day}"
      user => "filebeatuser"
      password => "filebeatpassword"
      cacert  =>  "/etc/logstash/elasticsearch-ca.pem"
  }
}
}

标签： elasticsearchfilebeat