ELK之使用filebeat收集java运行日志

　　安装filebeat修改配置文件/etc/filebeat/filebeat.yml

filebeat.prospectors:
- type: log
  enabled: true
#日志路径
  paths:
    - /home/ekp/linux64/tomcat/logs/catalina.out
#日志tags
  tags: [ekp-tomcat]
#排除空行
  exclude_lines: ['^$']
#java多行日志合并
  multiline:
    pattern: '^d{4}-d{1,2}-d{1,2}sd{1,2}:d{1,2}:d{1,2}'
    negate: true
    match: after
 
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 3
setup.kibana:
#输出至logstash
output.logstash:
  hosts: ["10.1.4.242:5044"]

　　传输用logstash配置/etc/logstash/conf.d/beat-redis.conf 如下

input{
    beats{
        port => 5044
    }
}

output{
    if "nginx-ekp-log" in [tags]{
        redis {
	    host => "10.1.4.243"
  	    port => "6379"
	    password => "123456"
 	    db => "1"
	    data_type => 'list'
	    key => "nginx-ekp-log"
        }
    }
    if "ekp-tomcat" in [tags]{
        redis {
	    host => "10.1.4.243"
  	    port => "6379"
	    password => "123456"
 	    db => "2"
	    data_type => 'list'
	    key => "ekp-tomcat"
        }
	#stdout{
	#    codec => rubydebug
	#}
    }
}

　　　　过滤分析logstash配置/etc/logstash/conf.d/redis-elastic.conf如下

input{
    redis {
	host => "10.1.4.243"
	port => "6379"
	password => "123456"
	db => "1"
	data_type => "list"
	key => "nginx-ekp-log"
    }
    redis {
	host => "10.1.4.243"
	port => "6379"
	password => "123456"
	db => "2"
	data_type => "list"
	key => "ekp-tomcat"
    }
}

filter{
    if "nginx-ekp-log" in [tags] {
        json {
	source => "message"
        }
    }
    if [user_ua] != "-" {
	useragent {
            target => "agent"
	    source => "user_ua"
	}
    }
    if [lan_ip] != "-" {
      geoip {
            source => "lan_ip"
            target => "geoip"
            # database => "/usr/share/GeoIP/GeoIPCity.dat"
            add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
            add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
            }
            mutate {
                        convert => [ "[geoip][coordinates]", "float"]
            }
     }
}

output{
   if "nginx-ekp-log" in [tags] {
	elasticsearch{
	hosts => ["10.1.4.244:9200"]
	index => "nginx-ekp-log-%{+YYYY.MM}"
        }
   }
   if "ekp-tomcat" in [tags] {
    elasticsearch{
    hosts => ["10.1.4.244:9200"]
    index => "ekp-tomcat-%{+YYYY.MM}"
    }
   }

 # stdout{
 #      codec => rubydebug
 #   }
}

　　启动filebeat，logstash即可把java日志多行合并进行收集