工作时候遇到了日志解析入库的配置，过程不复杂但是很经典，这里记录一下 Logstash 的日志解析的配置过程，以及设计到的主键的替换。

Grok 原始日志处理

原始日志的内容如下面给出：

[error][2020-08-07T00:00:24.520Z] responseError:{"message":"timeout of 20000ms exceeded","name":"Error","stack":"Error: timeout of 20000ms exceeded\n    at createError (/data/web-ssr/node_modules/axios/lib/core/createError.js:16:15)\n    at ClientRequest.handleRequestTimeout (/data/web-ssr/node_modules/axios/lib/adapters/http.js:256:16)\n    at Object.onceWrapper (events.js:286:20)\n    at ClientRequest.emit (events.js:203:15)\n    at ClientRequest.EventEmitter.emit (domain.js:448:20)\n    at Socket.emitRequestTimeout (_http_client.js:662:40)\n    at Object.onceWrapper (events.js:286:20)\n    at Socket.emit (events.js:198:13)\n    at Socket.EventEmitter.emit (domain.js:448:20)\n    at Socket._onTimeout (net.js:442:8)","config":{"url":"http://9.59.2.110:3035/http_operation/music_hit/get_home_data","method":"get","params":{"activity_id":"5f23b8c81174149d1c8518e7","rank_id":"5f23baaf1174149d1c8518e8","rank_type":"2","wmid":"191890335","lang":"en","region":"mm"},"data":null,"headers":{"Accept":"application/json, text/plain, */*","User-Agent":"axios/0.19.2"},"transformRequest":[null],"transformResponse":[null],"timeout":20000,"xsrfCookieName":"XSRF-TOKEN","xsrfHeaderName":"X-XSRF-TOKEN","maxContentLength":-1,"needModule":true},"code":"ECONNABORTED"}

根据日志的内容，宽衣根据特征来进行正则提取，分别是前面的level， time，和 后面的Json 的body。根据Grok 的语法来进行解析，这里使用 GREEDYDATA 可以理解为就是 .*的形式来进行正则的提取。之后，通过提取到的字段，来json 的解析。

filter{
        grok {
            match => [ "message", "\[%{GREEDYDATA:level}\]\[%{GREEDYDATA:time}\] responseError:%{GREEDYDATA:body}"]
        }
        date {
            match => ["time_local","yyyy-MMM-ddTHH:mm:ss Z"]
            target => "@timestamp"
        }
        json {
                source => "body"
        }
}

Replcae into

关于Elastic的根据主键的replace，直觉告诉我是不可以的，但是在查阅了不少资料之后发现，实际上是可行的。根据es 的文档里写，决定一个Document 的唯二的字段是 index 和 document_id。 因为都是写在同一个index 里面，所以这里的核心就是，在 ducument id。所以就有了下面的配置，来动态的更新docid 的值。从而实现了 es 里面的doc 的replace into 的功能。

output {
    if "parsefailure" in [tags] {
        stdout {codec => rubydebug}
    }else{
        if [type] == "xxx_trace" {
            elasticsearch {
                #...
                document_id => "%{sid}"
            }
        }
    }
}

参考

• Logstash最佳实践-Grok 正则捕获