虽然 Goreplay 本身有对接 es 的插件， 而且我也改写过插件支持到 es 6.x 的版本，详情见【重写 goreplay-es 插件，支持连接 es6.x】

但是其实使用起来不是很方便，首先是要回放才能用，要是不回放是不能传数据到 es 的，其次是跟传统的 elk 框架 filebeat + logstash + elasticsearch 不能集成在一起。

基于这样一些痛点，我决定用传统 elk 的方式，对 goreplay 的日志进行清洗组合，并结合 grafana 来做日志可视化和热力地图这样一些骚操作。

项目代码：goreplay-grafana 【 欢迎来 star !!! 】

grafana-dashboard: https://grafana.com/grafana/dashboards/11924

filebeat.yml:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/share/filebeat/gorlogs/*.log
  multiline.pattern: '🐵🙈🙉'    
  multiline.negate: true
  multiline.match: after
  tags: ["gorlogs"]

- type: log
  enabled: true
  paths:
    - /usr/share/filebeat/logs/*.log
  tags: ["normallogs"]

output.logstash:
  hosts: ["localhost:5044"]

这里做了个区分，因为有些可能是你的 nginx 格式的日志，不适合 goreplay 的日志组合清洗方式，所以这里通过打 tag 的方式来区分。

logstash.yml:

input {
    beats {
        port => "5044"
    }
}
filter {
    if "gorlogs" in [tags] {
        ruby {
            code => "
                array = event.get('message').split(' ')
                array.each_with_index do |item, index|
                    if item == 'Host:'
                        event.set('Host:', array[index + 1])
                    elsif item == 'GET'
                        event.set('Method', 'GET')
                        event.set('Url', array[index + 1])
                    elsif item == 'POST'
                        event.set('Method', 'POST')
                        event.set('Url', array[index + 1])
                    elsif item == 'User-Agent:'
                        event.set('User-Agent', array[index + 1])
                    elsif item == 'user-agent:'
                        temp = ''
                        i = index
                        until array[i + 1].include?('accept') do
                            temp = temp + array[i + 1] + ' '
                            i += 1
                        end
                        event.set('user-agent', temp)
                    elsif item == 'X-Real-IP:'
                        event.set('clientip', array[index + 1])
                    end
                end
            "
            remove_field => ['host', '@version', '_version', 'input', 'prospector', 'beat', 'message', 'offset', 'log']
        }
        geoip {
            source => "clientip"
            target => "geoip"
            fields => ["country_name", "region_name", "city_name", "location", "ip"]
        }
        useragent {
            source => "user-agent"
            target => "devices"
        }
    }

    if "normallogs" in [tags] {
        json {
            source => "message"
        }
    }
}
output {
    if "gorlogs" in [tags] {
        elasticsearch {
            hosts => [ "localhost:9200" ]
            index => "logstash-gor-%{+YYYY.MM.dd}"
        }
    }
    if "normallogs" in [tags] {
        elasticsearch {
            hosts => [ "localhost:9200" ]
            index => "logstash-%{+YYYY.MM.dd}"
        }
    }
}

分隔好以后的 message 是字符串，这里我用 ruby 的方式来分隔脚本，同时使用了 geoip user-agent 这些插件来增加有用信息。

最后的效果：

【注：代码里面提供了 es-cluster 集群的部署方式，实际 demo 使用的是 es 单节点部署。】

PS：技术交流 QQ 群 552643038