filebeat收集nginx日志发送到kafka，Logstash 读取 Kafka 日志消息队列到Elasticsearch，kibana做图形展示

es集群：10.0.0.70、10.0.0.71、10.0.0.72
kibana：10.0.0.73
mysql：10.0.0.73
filebeat1-web1：10.0.0.74
filebeat2-web2：10.0.0.76
kafka1-logstash1：10.0.0.77
kafka2-logstash2：10.0.0.78
kafka3-logstash3：10.0.0.79

#修改kafka和logstash的主机名称
hostnamectl set-hostname kafka1-logstash1-node1
hostnamectl set-hostname kafka2-logstash2-node2
hostnamectl set-hostname kafka3-logstash3-node3
#修改filebeat的主机名称
hostnamectl set-hostname filebeat1-web1
hostnamectl set-hostname filebeat2-web2

#脚本安装kafka和logstash
root@kafka1-logstash1-node1:/tools# bash /tools/install_kafka.sh 
请输入node编号(默认为 1): 1
root@kafka2-logstash2-node2:/tools# bash /tools/install_kafka.sh 
请输入node编号(默认为 1): 2
root@kafka3-logstash3-node3:/tools# bash /tools/install_kafka.sh 
请输入node编号(默认为 1): 3

#检查kafka集群
root@kafka1-logstash1-node1:/tools# /usr/local/zookeeper/bin/zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
root@kafka2-logstash2-node2:~# /usr/local/zookeeper/bin/zkServer.sh status
/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader
root@kafka3-logstash3-node3:~# /usr/local/zookeeper/bin/zkServer.sh status
/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower

#配置filebeat1-web1和filebeat2-web2的nginx日志为json格式
root@filebeat1-web1:~# cat /etc/nginx/nginx.conf
user www-data;
worker_processes auto;
pid /run/nginx.pid;
include /etc/nginx/modules-enabled/*.conf;

events {
worker_connections 768;
}

http {
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;

include /etc/nginx/mime.types;
default_type application/octet-stream;

ssl_protocols TLSv1 TLSv1.1 TLSv1.2 TLSv1.3; # Dropping SSLv3, ref: POODLE
ssl_prefer_server_ciphers on;

log_format access_json '{"@timestamp":"$time_iso8601",'
'"host":"$server_addr",'
'"clientip":"$remote_addr",'
'"size":$body_bytes_sent,'
'"responsetime":$request_time,'
'"upstreamtime":"$upstream_response_time",'
'"upstreamhost":"$upstream_addr",'
'"http_host":"$host",'
'"uri":"$uri",'
'"domain":"$host",'
'"xff":"$http_x_forwarded_for",'
'"referer":"$http_referer",'
'"tcp_xff":"$proxy_protocol_addr",'
'"http_user_agent":"$http_user_agent",'
'"status":"$status"}';
access_log /var/log/nginx/access_json.log access_json ;
error_log /var/log/nginx/error.log;
gzip on;

include /etc/nginx/conf.d/*.conf;
include /etc/nginx/sites-enabled/*;
}

#配置2台filebeat链接kafka
root@filebeat1-web1:~# cat /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access_json.log
  json.keys_under_root: true
  json.overwrite_keys: true
  tags: ["nginx-access"]

- type: log
  enabled: true
  paths:
    - /var/log/nginx/error.log
  tags: ["nginx-error"]

- type: log
  enabled: true
  paths:
    - /var/log/syslog
  tags: ["syslog"]

output.kafka:
  hosts: ["10.0.0.77:9092","10.0.0.78:9092","10.0.0.79:9092"]
  topic: filebeat-log   #指定kafka的topic
  partition.round_robin:
    reachable_only: true #true表示只发布到可用的分区，false时表示所有分区，如果一个节点down，会block
  required_acks: 1  #如果为0，错误消息可能会丢失，1等待写入主分区（默认），-1等待写入副本分区
  compression: gzip  #实现gzip的压缩
  max_message_bytes: 1000000 #每条消息最大长度，以字节为单位，如果超过将丢弃



#重启
systemctl restart nginx

#重启
systemctl restart filebeat.service

#前台启动查看日志
filebeat -c /etc/filebeat/filebeat.yml -e