fliebeat+kafka的ELK日志分析平台（上）

一.环境讲解

当前结构，Filebeat部署在需要收集日志的机器上，收集日志，输出到zk+kakfa集群这个中间件中。logstash从kafka集群消费信息，并根据配置内容，进行格式转化和过滤，整理好的数据会发给elastic进行存储。elastic能对大容量的数据进行接近实时的存储、搜索和分析操作。最后由kibana提供web界面，调用elastic做数据分析，然后展示出来。

注意点： 1.filebeat启动后，只会读取最新输入的日志，类似tail -f 2.kafka集群的内容，logstash读取一条，就会消失一条，不会造成重复读取 3.这套系统里，只有elastic存储数据

#当前用5台机器模拟，实际可使用2-3台，将zk+kafka集群每台都进行部署，也可以1台，部署多实例。
[nginx服务]
    主机名 = nginx-server
    系统 = centos-7.3
    地址 = 1.1.1.1
    软件 = nginx-1.8 80
           filebeat-7.4.2
[zk+kafka集群]
    主机名 = kafka-1
    系统 = centos-7.3
    地址 = 1.1.1.2
    软件 = jdk-1.8
           zookeeper-3.5 2181
           kafka-2.0.0 9092
    主机名 = kafka-2
    系统 = centos-7.3
    地址 = 1.1.1.3
    软件 = jdk-1.8
           zookeeper-3.5 2181
           kafka-2.0.0 9092
[elasticsearch]
    主机名 = elastic-server
    系统 = centos-7.3
    地址 = 1.1.1.4
    软件 = jdk-1.8
           elasticsearch-7.4.2 9200
[logstash+kibana]
    主机名 = lk-server
    系统 = centos-7.3
    地址 = 1.1.1.5
    软件 = jdk-1.8
           logstash-7.4.2 9600
           kibana-7.4.2 5601

注意事项： 1.filebeat的版本要一致，在官网都有对应的包 2.filebeat连接kafka的版本，是有支持范围的，可查看官方文档filebeat-kafka配置 3.不同版本elk需要的jdk版本也不同，需要看好说明

二.部署配置

配置kafka集群

操作服务器（kafka-1,kafka-2） 1.这里使用2台组建kafka集群，可根据需求添加或删减节点。部署可查看文章kafka集群部署

2.按照上述连接，启动并测试好kafka集群，确保可正常使用

配置日志输出端

操作服务器（nginx-server） 1.Nginx作为日志输出端，这里也可以用自定义文件代替，然后手动插入内容 yum -y install nginx-1.8

若没有包，可扩展epel或者源码安装nginx-1.6

2.启动并本地访问，促使产生日志 systemctl start nginx curl http://127.0.0.1 cat /var/log/nginx/access.log

3.安装filebeat curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.4.2-x86_64.rpm rpm -vi filebeat-7.4.2-x86_64.rpm

4.建立文件夹，这里使用自建立的配置文件，方便修改维护 mkdir conf cd conf

建立filebeat配置文件，格式一定要对齐，因为是yml格式的。启动后，filebeat会在kafka中建立一个叫nginx-log的topic vim filebeat-nginx.yml

filebeat.inputs:
- type: log
  paths:
    - /var/log/nginx/access.log
output.kafka:
  hosts: ["1.1.1.2:9092", "1.1.1.3:9092"]
  topic: 'nginx-log'

具体的参数解释，可以查看filebeat配置文件详解

5.启动filebeat，这里会把启动日志输出到当前目录filebeat.log文件中，方便查看 nohup /usr/share/filebeat/bin/filebeat -e -c filebeat-nginx.yml &>> filebeat.log &

若不是这个地址，可以用如下命令，去查找这个命令位置在哪 rpm -ql filebeat

6.等半分钟，然后查看filebeat进程和日志，是否启动有问题 ps -aux |grep filebeat

查看日志，没有异常情况，就可以进行下一步了 cat filebeat.log

7.写入日志 curl http://127.0.0.1

操作服务器（kafka-1） 8.kafka中应该有新的topic nginx-log产生，test是创建集群时，测试功能创建的 cd /usr/local/kafka bin/kafka-topics.sh --list --zookeeper localhost:2181

查看topic里的消息内容，可以看到日志信息的 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic nginx-log --from-beginning

配置logstash

操作服务器（lk-server） 1.先安装jdk-1.8 yum install java-1.8.0 -y

2.下载logstash的二进制包并部署 wget https://artifacts.elastic.co/downloads/logstash/logstash-7.4.2.tar.gz tar -xf logstash-7.4.2.tar.gz mv logstash-7.4.2 /usr/local/logstash

3.建立文件夹，这里使用自建立的配置文件，方便修改维护 mkdir conf cd conf

建立logstash配置文件，格式一定要对齐，因为是yml格式的。启动后会从kafka取数据，并传输给elasticsearch，中间是对nginx的日志数据，进行正则分段 vim logstash-nginx.yml

input {
  kafka  {
    auto_offset_reset => "latest"
    bootstrap_servers => "1.1.1.2:9092,1.1.1.3:9092"
    topics => ["nginx-log"]
    group_id => "logstash-file"
    codec => "json"
  }
}
filter {
  grok {
    match => {
      "message" => "%{IPORHOST:clientip} \[%{HTTPDATE:time}\] \"%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:http_status_code} %{NUMBER:bytes} \"(?<http_referer>\S+)\" \"(?<http_user_agent>\S+)\" \"(?<http_x_forwarded_for>\S+)\""
    }
  }
}
output {
  elasticsearch {
    hosts => ["192.168.56.106:9200"]
      index => "nginx-%{+YYYY.MM.dd}"
  }
}
具体的参数解释，可以查看logstash配置文件详解

4.启动logstash，这里会把启动日志输出到当前目录logstash.log文件中，方便查看 nohup /usr/local/logstash/bin/logstash -f logstash-nginx.yml &>> logstash.log &

5.等半分钟，然后查看logstash进程和日志，是否启动有问题 ps -aux|grep logstash

进程是启动着的，则查看日志，看看有没有异常 cat logstash.log