1、原始数据源服务器，从日志拉取数据，并转化为utf8编码，增加ip地址，保存到Redis中，上配置如下：

[plain] view plain copy 

1. input {  

2.     file {  

3.         path => "/tmp/123server.log"  

4.         codec => json {  

5.             charset => "GBK"  

6.         }  

7.         start_position => "beginning"  

8.         sincedb_path => "/dev/null"  

9.         type => "activitysun"  

10.     }  

11. }  

12.   

13. filter {  

14.     date {  

15.         match => ["timestamp", "UNIX"]  

16.         remove_field => ["time"]  

17.     }  

18.     #ruby {  

19.     #    code => "event.timestamp.time.localtime"  

20.     #}  

21.     mutate {  

22.         #some pc no host  

23.     replace => { "host" => "192.168.10.11" }     

24.     }  

25. }  

26.   

27. output {  

28.     #stdout {  

29.     #    codec => plain {  

30.     #        charset => "UTF-8"  

31.     #    }  

32.     #}  

33.     file {  

34.         path => "/tmp/logstash.log"  

35.         codec => json {  

36.             charset => "UTF-8"  

37.         }  

38.     }  

39.     redis {  

40.         host => ["192.168.10.18"]  

41.         port => 16378  

42.         data_type => "list"  

43.         key => "123server"  

44.         codec => json {  

45.             charset => "UTF-8"  

46.         }  

47.     }  

48. }  

采用如下命令启动

[plain] view plain copy 

1. /data/logstash-2.3.4/bin/logstash -f activitylog.conf  

2、收集服务器，从redis拉取数据，放入elasticsearch，时间做一下处理，方便分析。配置如下：

[plain] view plain copy 

1. input {  

2.   redis {  

3.     host => ["192.168.10.18"]  

4.     port => 26378  

5.     data_type => "list"  

6.     key => "123server"  

7.     codec => json {  

8.       charset => "UTF-8"  

9.     }  

10.     #type => "activitysun"  

11.   }  

12. }  

13.   

14. filter {  

15.     ruby {  

16.         code=>"event['daytag']=event.timestamp.time.localtime.strftime('%Y.%m.%d')"  

17.     }  

18. }  

19.   

20. output {  

21.   elasticsearch {  

22.     hosts => ["127.0.0.1:19200"]  

23.     index => "%{type}-%{daytag}"  

24.     #index => "%{type}-%{+yyyy.MM.dd}"  

25.   }  

26. }  

3、kibana展示

filebeat.yml

filebeat.prospectors:
- input_type: log
  paths:
    - /log/123.log
    #- /home/haoren/data/filebeat-5.0.2-linux-x86_64/test.log
  encoding: gbk
  symlinks: true
  include_lines: ['\[.*?统计\]','\[.*?结算\]']
  document_type: pchannelserver
  fields_under_root: true
  fields:
    host: 192.168.10.15


processors:
- drop_fields:
    #fields: ["beat.hostname", "beat.name", "beat.version", "input_type", "beat"]
    fields: ["input_type", "beat", "offset", "source"]


output.redis:
  enabled: true
  hosts: ["192.168.10.18"]
  password: "8841c09BAD52E63067C4DA"
  port: 15377
  datatype: list
  key: "filebeat"
  db: 0


output.file:
  enabled: true
  path: "/tmp/filebeat"


output.console:
  enabled: false