使用Filebeat和Logstash集中归档游戏日志

方 案

Filebeat->Logstash->Files
Filebeat->Redis->Logstash->Files
Nxlog(Rsyslog、Logstash)->Kafka->Flink(Logstash->ES-Kibana)
其他方案(可根据自己需求，选择合适的架构,作者选择了第二种方案)

注释: 由于Logstash无法处理输出到文件乱序的问题，可通过不同的文件使用不同的Logstash；或者直接写入ES(不存在乱序问题)、通过Flink输出到文件

部 署

系统环境
Debian8 x64
logstash-6.1.1
filebeat-6.1.1-amd64
Redis-3.2

Filebeat配置

1234567891011121314151617181920212223

/etc/filebeat/filebeat.ymlfilebeat.prospectors:- type: log paths: - /home/data/log/* - /home/data/*.log scan_frequency: 20s encoding: utf-8 tail_files: true harvester_buffer_size: 5485760fields: ip_address: 192.168.2.2 env: qaoutput.redis: hosts: ["192.168.1.1:6379"] password: "geekwolf" key: "filebeat" db: 0 timeout: 5 max_retires: 3 worker: 2 bulk_max_size: 4096

Logstash配置

12345678910111213141516171819202122232425262728293031323334353637383940

input { #Filebeat # beats { # port => 5044 # } #Redis redis { batch_count => 4096 data_type => "list" key => "filebeat" host => "127.0.0.1" port => 5044 password => "geekwolf" db => 0 threads => 2 }}filter { ruby { code => 'event.set("filename",event.get("source").split("/")[-1])' }}output { if [filename] =~ "nohup" { file { path => "/data/logs/%{[fields][env]}/%{+YYYY-MM-dd}/%{[fields][ip_address]}/%{filename}" flush_interval => 3 codec => line { format => "%{message}"} } } else { file { path => "/data/logs/%{[fields][env]}/%{+YYYY-MM-dd}/%{[fields][ip_address]}/logs/%{filename}" flush_interval => 3 codec => line { format => "%{message}"} } } #stdout { codec => rubydebug }}

生产日志目录

123456789101112

.├── prod│   └── 2018-01-13│   └── 2.2.2.2│   ├── logs│   │   ├── rpg_slow_db_.27075│   └── nohup_service.log└── qa ├── 2018-01-12 │   ├── 192.168.3.1 └── 2018-01-13 ├── 192.168.3.2