ELK+kafka+filebeat搭建生产ELFK集群

1
2
3

hostnamectl set-hostname elk-master
hostnamectl set-hostname elk-node1
hostnamectl set-hostname elk-node2

1
2
3
4
5
6
7
8

cat >>/etc/security/limits.conf<< EOF
*               soft      nofile          65536
*               hard      nofile          65536
*               soft      nproc           65536
*               hard      nproc           65536
*               hard      memlock         unlimited
*               soft      memlock         unlimited
EOF

1
2
3
4
5

cat >>/etc/systemd/system.conf<< EOF
DefaultLimitNOFILE=65536
DefaultLimitNPROC=32000
DefaultLimitMEMLOCK=infinity
EOF

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

cat >>/etc/sysctl.conf<< EOF
# 关闭交换内存
vm.swappiness =0
# 影响java线程数量，建议修改为262144或者更高
vm.max_map_count= 262144
# 优化内核listen连接
net.core.somaxconn=65535
# 最大打开文件描述符数，建议修改为655360或者更高
fs.file-max=655360
# 开启ipv4转发
net.ipv4.ip_forward= 1
EOF

1
2
3
4
5

cat >>/etc/hosts<< EOF
elk-master  10.0.11.172
elk-node1   10.0.21.117
elk-node2   10.0.11.208
EOF

1

reboot

1
2
3
4

#存放快照日志
mkdir zkdata 
#存放事物日志
mkdir zklogs 

1
2
3

wget  http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.6/apache-zookeeper-3.5.6-bin.tar.gz
tar -zxvf apache-zookeeper-3.5.6-bin.tar.gz 
mv apache-zookeeper-3.5.6-bin zookeeper

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

[root@elk-master zookeeper]# cat conf/zoo.cfg  |grep  -v ^#
# 服务器之间或客户端与服务器之间维持心跳的时间间隔
# tickTime以毫秒为单位。
tickTime=2000 
# 集群中的follower服务器(F)与leader服务器(L)之间的初始连接心跳数
initLimit=10
# 集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数
syncLimit=5
# 数据保存目录
dataDir=../zkdata
# 日志保存目录
dataLogDir=../zklogs
# 客户端连接端口
clientPort=2181
# 客户端最大连接数。# 根据自己实际情况设置，默认为60个
maxClientCnxns=60
# 三个接点配置，格式为： server.服务编号=服务地址、LF通信端口、选举端口
server.1=10.0.11.172:2888:3888
server.2=10.0.21.117:2888:3888
server.3=10.0.11.208:2888:3888

1
2
3
4
5

[root@elk-master zookeeper]# cd /home/tools/zookeeper/bin/
[root@elk-master bin]# ./zkServer.sh start 
ZooKeeper JMX enabled by default
Using config: /home/tools/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

1
2
3
4
5

[root@elk-master bin]# sh /home/tools/zookeeper/bin/zkServer.sh status 
ZooKeeper JMX enabled by default
Using config: /home/tools/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader

1
2
3
4
5

cat >>/etc/profile<< EOF
export ZOOKEEPER_INSTALL=/home/tools/zookeeper/
export PATH=$PATH:$ZOOKEEPER_INSTALL/bin
export PATH
EOF

1

source /etc/profile

1
2
3
4
5
6

[root@elk-master tools]# mkdir kafka
[root@elk-master tools]# cd kafka/
[root@elk-master kafka]# wget https://www-eu.apache.org/dist/kafka/2.3.0/kafka_2.12-2.3.0.tgz
[root@elk-master kafka]# tar xf kafka_2.12-2.3.0.tgz 
[root@elk-master kafka]# mv kafka_2.12-2.3.0 kafka
[root@elk-master kafka]# cd kafka/config/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

[root@elk-master config]# cat /home/tools/kafka/kafka/config/server.properties
############################# Server Basics ############################# 
# broker的id，值为整数，且必须唯一，在一个集群中不能重复
broker.id=1

############################# Socket Server Se：ttings ############################# 
# kafka默认监听的端口为9092 (默认与主机名进行连接)
listeners=PLAINTEXT://:9092

# 处理网络请求的线程数量，默认为3个
num.network.threads=3

# 执行磁盘IO操作的线程数量，默认为8个 
num.io.threads=8

# socket服务发送数据的缓冲区大小，默认100KB
socket.send.buffer.bytes=102400

# socket服务接受数据的缓冲区大小，默认100KB
socket.receive.buffer.bytes=102400

# socket服务所能接受的一个请求的最大大小，默认为100M
socket.request.max.bytes=104857600

############################# Log Basics ############################# 
# kafka存储消息数据的目录
log.dirs=../kfkdata

# 每个topic默认的partition数量
num.partitions=3

# 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
num.recovery.threads.per.data.dir=1

############################# Log Flush Policy ############################# 

# 消息刷新到磁盘中的消息条数阈值
#log.flush.interval.messages=10000

# 消息刷新到磁盘中的最大时间间隔,1s
#log.flush.interval.ms=1000

############################# Log Retention Policy ############################# 

# 日志保留小时数，超时会自动删除，默认为7天
log.retention.hours=168

# 日志保留大小，超出大小会自动删除，默认为1G
#log.retention.bytes=1073741824

# 日志分片策略，单个日志文件的大小最大为1G，超出后则创建一个新的日志文件
log.segment.bytes=1073741824

# 每隔多长时间检测数据是否达到删除条件,300s
log.retention.check.interval.ms=300000

############################# Zookeeper ############################# 
# Zookeeper连接信息，如果是zookeeper集群，则以逗号隔开
zookeeper.connect=10.0.11.172,10.0.21.117,10.0.11.208

# 连接zookeeper的超时时间,6s
zookeeper.connection.timeout.ms=6000

1

[root@elk-master config]# mkdir ../kfkdata

1
2
3
4
5

cd /home/tools/kafka/kafka/bin/
#启动测试
./kafka-server-start.sh ../config/server.properties
#放入后台
./kafka-server-start.sh -daemon ../config/server.properties

1
2

wget https://pan.cnsre.cn/d/Package/Linux/ELK/elasticsearch-7.10.1-x86_64.rpm
[root@elk-master package]#  rpm -ivh elasticsearch-7.10.1-x86_64.rpm 

1
2

cd /etc/elasticsearch
cp elasticsearch.yml  elasticsearch.yml.bak

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

cat >/etc/elasticsearch/elasticsearch.yml << EOF
#集群名
cluster.name: elk-cluster

#node名
node.name: elk-1

#数据存储路径
path.data: /home/elasticsearch/esdata

#数据快照路径
path.repo: /home/backup/essnapshot

#日志存储路径
path.logs: /home/elasticsearch/eslogs

#es绑定的ip地址，根据自己机器ip进行修改
network.host: 0.0.0.0

#服务端口
http.port: 9200

#集群master需要和node名设置一致
discovery.seed_hosts: ["10.0.11.172", "10.0.21.117", "10.0.11.208"]
cluster.initial_master_nodes: ["10.0.11.172","10.0.21.117","10.0.11.208"]

#允许跨域请求
http.cors.enabled: true
#* 表示支持所有域名
http.cors.allow-origin: "*"
#添加请求header
http.cors.allow-headers: Authorization,X-Requested-With,Content-Length,Content-Type

#生产必须为true，内存锁定检查，目的是内存地址直接映射，减少一次copy时间
bootstrap.memory_lock: true
#系统过滤检查，防止数据损坏，考虑集群安全，生产设置成false
bootstrap.system_call_filter: false

#xpack配置
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: /etc/elasticsearch/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: /etc/elasticsearch/elastic-certificates.p12
EOF

1
2
3

[root@elk-node1 elasticsearch]# cat -n  jvm.options |grep 8g
    22  -Xms8g
    23  -Xmx8g

1
2
3
4

mkdir  -p /home/elasticsearch/{esdata,eslogs}
chown  elasticsearch:elasticsearch  /home/elasticsearch/*
mkdir  -p /home/backup/essnapshot
chown elasticsearch:elasticsearch /home/backup/essnapshot

1
2

systemctl start elasticsearch
systemctl enable  elasticsearch

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

Unexpected response code [503] from calling PUT http://10.0.11.172:9200/_security/user/apm_system/_password?pretty
Cause: Cluster state has not been recovered yet, cannot write to the [null] index

Possible next steps:
* Try running this tool again.
* Try running with the --verbose parameter for additional messages.
* Check the elasticsearch logs for additional error details.
* Use the change password API manually. 

ERROR: Failed to set password for user [apm_system].

1
2

/usr/share/elasticsearch/bin/elasticsearch-certutil ca
/usr/share/elasticsearch/bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

/usr/share/elasticsearch/bin/elasticsearch-setup-passwords interactive
Initiating the setup of passwords for reserved users elastic,apm_system,kibana,logstash_system,beats_system,remote_monitoring_user.
You will be prompted to enter passwords as the process progresses.
Please confirm that you would like to continue [y/N]y
Enter password for [elastic]:
Reenter password for [elastic]:
Enter password for [apm_system]:
Reenter password for [apm_system]:
Enter password for [kibana]:
Reenter password for [kibana]:
Enter password for [logstash_system]:
Reenter password for [logstash_system]:
Enter password for [beats_system]:
Reenter password for [beats_system]:
Enter password for [remote_monitoring_user]:
Reenter password for [remote_monitoring_user]:
Changed password for user [apm_system]
Changed password for user [kibana]
Changed password for user [logstash_system]
Changed password for user [beats_system]
Changed password for user [remote_monitoring_user]
Changed password for user [elastic]

1
2

wget https://pan.cnsre.cn/d/Package/Linux/ELK/cerebro-0.9.2-1.noarch.rpm
rpm -ivh cerebro-0.9.2-1.noarch.rpm 

1
2
3

systemctl  start cerebro.service
systemctl  enable cerebro.service
systemctl  status  cerebro.service

1
2

https://artifacts.elastic.co/downloads/kibana/kibana-7.10.1-x86_64.rpm
rpm -ivh kibana-7.10.1-x86_64.rpm

1
2

cd /etc/kibana/
mv kibana.yml kibana.yml.bak

1
2
3
4
5
6
7

vim kibana.yml
server.port: 5601
server.host: 0.0.0.0
elasticsearch.hosts: ["http://10.0.11.172:9200/","http://10.0.21.117:9200/","http://10.0.11.208:9200/"]
elasticsearch.username: "elastic"
elasticsearch.password: "123"
i18n.locale: "zh-CN"

1
2
3

systemctl  start   kibana.service
systemctl  enable  kibana.service
systemctl  status  kibana.service

1
2
3
4

wget https://pan.cnsre.cn/d/Package/Linux/ELK/filebeat-7.10.1-x86_64.rpm
rpm -ivh filebeat-7.10.1-x86_64.rpm 
cd /etc/filebeat/
cp filebeat.yml filebeat.yml.bak

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

#=========================== Filebeat inputs =============================
max_procs: 1                     #限制filebeat的进程数量,其实就是内核数，避免过多抢占业务资源
queue.mem.events: 256            # 存储于内存队列的事件数，排队发送 (默认4096)
queue.mem.flush.min_events: 128  # 小于 queue.mem.events ,增加此值可提高吞吐量 (默认值2048)
filebeat.inputs:                 # inputs为复数，表名type可以有多个
- type: log                      # 输入类型
  enable: true                   # 启用这个type配置
  paths:
    - /home/homeconnect/logs/AspectLog/aspect.log  # 监控tomcat  的业务日志
  json.keys_under_root: true     #默认Flase,还会将json解析的日志存储至messages字段
  json.overwrite_keys: true      #覆盖默认的key,使用自定义json格式的key
  max_bytes: 20480               # 单条日志的大小限制,建议限制(默认为10M,queue.mem.events * max_bytes 将是占有内存的一部)
  fields:                        # 额外的字段
     source: test-prod-tomcat-aspect-a    # 自定义source字段，用于es建议索引（字段名小写，我记得大写好像不行）

- type: log      # 输入类型
  enable: true   # 启用这个type配置
  paths:
    - /home/tools/apache-tomcat-8.5.23/logs/localhost_access_log.*.log  # 监控tomcat access日志
  json.keys_under_root: true   #默认Flase,还会将json解析的日志存储至messages字段
  json.overwrite_keys: true    #覆盖默认的key,使用自定义json格式的key
  max_bytes: 20480             # 单条日志的大小限制,建议限制(默认为10M,queue.mem.events * max_bytes 将是占有内存的一部分)
  fields:                      # 额外的字段
     source: test-prod-tomcat-access-a    # 自定义source字段，用于es建议索引

# 自定义es的索引需要把ilm设置为false
setup.ilm.enabled: false

#=============================== output ===============================
output.kafka:         # 输出到kafka
    enabled: true     # 该output配置是否启用
    hosts: ["10.0.11.172:9092","10.0.21.117:9092","10.0.11.208:9092"] # kafka节点列表
    topic: 'logstash-%{[fields.source]}'  # kafka会创建该topic，然后logstash(可以过滤修改)会传给es作为索引名称
    partition.hash:
      reachable_only: true # 是否只发往可达分区
    compression: gzip      # 压缩
    max_message_bytes: 1000000  # Event最大字节数。默认1000000。应小于等于kafka broker message.max.bytes值
    required_acks: 1  # kafka ack等级
    worker: 1  # kafka output的最大并发数
    bulk_max_size: 2048    # 单次发往kafka的最大事件数
    logging.to_files: true   # 输出所有日志到file，默认true， 达到日志文件大小限制时，日志文件会自动限制替换

#=============================== other ===============================
close_older: 30m         # 如果文件在某个时间段内没有发生过更新，则关闭监控的文件handle。默认1h
force_close_files: false # 这个选项关闭一个文件,当文件名称的变化。只在window建议为true
# 没有新日志采集后多长时间关闭文件句柄，默认5分钟，设置成1分钟，加快文件句柄关闭
close_inactive: 1m
# 传输了3h后荏没有传输完成的话就强行关闭文件句柄，这个配置项是解决以上案例问题的key point
close_timeout: 3h
# 这个配置项也应该配置上，默认值是0表示不清理，不清理的意思是采集过的文件描述在registry文件里永不清理，在运行一段时间后，registry会变大，可能会带来问题
clean_inactive: 72h
# 设置了clean_inactive后就需要设置ignore_older，且要保证ignore_older < clean_inactive
ignore_older: 70h                                                                                         

1
2
3

systemctl  start  filebeat.service
systemctl  enable  filebeat.service
systemctl  status  filebeat.service

1
2
3

wget https://pan.cnsre.cn/d/Package/Linux/ELK/logstash-7.10.1-x86_64.rpm
rpm -ivh logstash-7.10.1-x86_64.rpm
mv logstash.yml logstash.yml.bak

1
2
3
4
5
6

vim logstash.yml
http.host: "0.0.0.0"
# 指发送到Elasticsearch的批量请求的大小，值越大，处理则通常更高效，但增加了内存开销
pipeline.batch.size: 3000
# 指调整Logstash管道的延迟，过了该时间则logstash开始执行过滤器和输出
pipeline.batch.delay: 200

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

[root@elk-kibana conf.d]# cat /etc/logstash/conf.d/get-kafka-logs.conf 
input {                                   # 输入组件
  kafka {                                  # 从kafka消费数据
    bootstrap_servers => ["10.0.11.172:9092,10.0.21.117:9092,10.0.11.208:9092"]
    codec => "json"                        # 数据格式
    #topics => ["3in1-topi"]                   # 使用kafka传过来的topic
    topics_pattern => "logstash-.*"             # 使用正则匹配topic
    consumer_threads => 3                  # 消费线程数量
    decorate_events => true                # 可向事件添加Kafka元数据，比如主题、消息大小的选项，这将向logstash事件中添加一个名为kafka的字段
    auto_offset_reset => "latest"          # 自动重置偏移量到最新的偏移量
    #group_id => "logstash-node"            # 消费组ID，多个有相同group_id的logstash实例为一个消费组
    #client_id => "logstash1"               # 客户端ID
    fetch_max_wait_ms => "1000"            # 指当没有足够的数据立即满足fetch_min_bytes时，服务器在回答fetch请求之前将阻塞的最长时间
  }
}

filter{
   # 当非业务字段时，无traceId则移除
   #if ([message] =~ "traceId=null") {          # 过滤组件，这里只是展示，无实际意义，根据自己的业务需求进行过滤
   #   drop {}
   #}
mutate {
    convert => ["Request time", "float"]
    }
        if [ip] != "-" {
        geoip {
                       source => "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 {           # 输出组件
  elasticsearch {   # Logstash输出到es
    hosts => ["10.0.11.172:9200","10.0.21.117:9200","10.0.11.208:9200"]
    index => "logstash-%{[fields][source]}-%{+YYYY-MM-dd}"  # 直接在日志中匹配
    #index => "%{[@metadata][topic]}-%{+YYYY-MM-dd}"  # 以日期建索引
    user => "elastic"
    password => "123"
  }
  #stdout {
  #    codec => rubydebug
 #}
}

1

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/get-kafka-logs.conf

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

[Unit]
Description=root

[Service]
Type=simple
User=root
Group=root
# Load env vars from /etc/default/ and /etc/sysconfig/ if they exist.
# Prefixing the path with '-' makes it try to load, but if the file doesn't
# exist, it continues onward.
EnvironmentFile=-/etc/default/logstash
EnvironmentFile=-/etc/sysconfig/logstash
ExecStart=/usr/share/logstash/bin/logstash "--path.settings" "/etc/logstash"
Restart=alway
WorkingDirectort=/
Nice=19
LimitNOFILE=16384

[Install]
WantedBy=multi-user.target

1
2
3
4

[root@elk-kibana ~]# cat  -n /usr/share/logstash/bin/logstash.lib.sh |grep  JAVACMD
    85    # set the path to java into JAVACMD which will be picked up by JRuby to launch itself
    86    JAVACMD="/home/tools/jdk1.8.0_221/bin/java"
    87    if [ -z "$JAVACMD" ]; then

1
2
3

systemctl reload logstash.service
systemctl restart  logstash.service
systemctl enable  logstash.service