主要解决生产上越来越多的平台应用访问的可数据化显示

---

一、logstash 的 client

就是从 /data/logs/access.log 进行写入到 kafka 里面

logstash配置

---

input{

file {

type => "apachelogs"

path => "/data/logs/access.log"

start_position => "beginning"

}

}

filter {

if [type] == "apachelogs" {

grok {

match => {"message" => "%{COMBINEDAPACHELOG}"}

}

geoip {

source => 'cilentip'

}

date {

match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]

}

}

}

output{

kafka {

bootstrap_servers => "192.168.158.129:9092,192.168.158.130:9092,192.168.158.132:9092"

topic_id => "apachelogs"

compression_type => "snappy"

}

}

---

二、logstash 配置server

---

input {

kafka {

zk_connect => "192.168.158.130:2181,192.168.158.129:2181,192.168.132:2181"

topic_id => "apachelogs"

codec =>  json

reset_beginning => false

consumer_threads => 5

decorate_events => true

}

}

output {

elasticsearch {

hosts => ["192.168.158.128:9200","192.168.158.131:9200"]

index => "apachelogs-%{+YYYY-MM-dd}"

}

}

---

三、查看信息

1、登 录elk

http://192.168.158.128:9200/_plugin/head/

  如上图所示：可以看到很多个字段内容  

2、在kafka监控查看到

可以看到 apachelogs的在kafka里创建了

3、登 录kibana

http://192.168.158.128:5601/

可以通过id 查对应的错误信息：

选择饼图

能进行统计信息

这里值得提醒的是： /data/logs/access.log的文件必须是今天的日期， logstash 估计会认为是旧的日志文件而不进行提交操作，导致我kafka一直没有数据。因为我是从服务器拿的数据，比较旧的数据，所以要改变一下文件的时间。