企业级日志中心(ELK)

一、基础概念

随着用户系统规模的增大以及时间的积累，用户系统产生的日志将会越来越多，对日志的管理和查询就可以使用 ELK 以及 Kafka 来实现日志数据采集，处理最终存储到移动云 Elasticsearch 中，并可以通过 Kibana 连接 Elasticsearch 来实现对日志数据的搜索和可视化的展示。

• Filebeat：一个轻量级开源日志文件数据搜集器，用来收集日志的，因为原来收集日志的工具Logstash是java开发的，占用资源过多，影响性能会对服务器产生压力影响性能，
• Kafka:数据缓冲队列，作为消息队列解耦了处理过程，同时提高了可扩展性，具有削峰处理能力。
• Logstash:主要用来日志的搜集、分析、过滤日志。可以用它去搜集日志、转换日志、解析日志并将他们作为数据提供给其他模块调用，例如搜索，存储等
• Elasticsearch: 是一个级域lucene的搜索服务器，提供搜集、分析、存储数据三大功能，它提供了一个分布式多用户能力的全文搜索引擎
• Kibana：用于展示，将存储在es中的索引数据转化为各种图标，为用户提供强大的数据可视化支持

安装版本

filebeat-7.4.0-linux-x86_64.tar.gz
elasticsearch-7.4.0-linux-x86_64.tar.gz
kibana-7.4.0-linux-x86_64.tar.gz
logstash-7.4.0.tar.gz
kafka_2.12-2.3.0.tgz
软件分享链接
通过百度网盘分享的文件：EFK
链接：https://pan.baidu.com/s/15loB80kETOqqo_q6Bb9qCg?pwd=aura
提取码：aura

二、filebeat安装配置

1.把filebeat包传至想要收集日志的应用机器上，解压到指定目录

#将filebeat包解压到指定目录
tar -xzf filebeat-7.4.0-linux-x86_64.tar.gz -C /opt/app
改名
mv filebeat-7.4.0 filebeat
2.修改filebeat.yml
#进入配置文件目录
cd /opt/app/filebeat

修改filebeat.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
     - /opt/applog/ldp/daily.log  # 采集日志的路径
  fields:
    log_topic: "test-ldp"　　　　　# topic名字 
  multiline.pattern: ^\[
  multiline.negate: true
  multiline.match: after
# 输出到kafka
output.kafka:
  hosts: ["xxx.xxx.xxx.xxx:9092"]     #kafka服务器ip
  topic: '%{[fields][log_topic]}'
# 启动filebeat（前提：filebeat安装机器与安装kafka的机器网络层相通）
cd /opt/app/filebeat
nohup ./filebeat -e -c filebeat.yml > /opt/applog/filebeat.log &

# 如果启动日志出现Failed to connect to broker vm-vmwxxx-app:9092: dial tcp: lookup vm-vmwxxx-app on [::1]:53: read udp [::1]:33342->[::1]:53: read: connection refused
# 解决方法
切换到root用户执行：vi /etc/hosts
在文件末尾添加部署kafka机器的IP与主机名即可

三、kafka的安装配置

1.解压安装包到指定目录

启动kafka需要提前安装jdk环境

# 解压文件并改名
tar -xzf kafka_2.12-2.3.0.tgz -C /DATA/elk
mv kafka_2.12-2.3.0 kafka
# 修改kafka配置
cd /DATA/elk/kafka/config
vi server.properties

broker.id=0 
port=9092 #端口号 
listeners=PLAINTEXT://127.0.0.1:9092    #监听地址
advertised.listeners=PLAINTEXT://127.0.0.1:9092
log.dirs=/DATA/logs/kafka #日志存放路径可修改可不修改
zookeeper.connect=xxx.xxx.xxx.xxx:2181 #zookeeper地址和端口

2.Zookeeper配置（使用其自带的Zookeeper）

#zk配置文件名：zookeeper.properties
cd /DATA/elk/kafka/config
vi zookeeper.properties
# zk配置
dataDir=/DATA/logs/zk_data
dataLogDir=/DATA/logs/zk_log
clientPort=2181
maxClientCnxns=100
tickTime=2000
initLimit=10

3.编写启停脚本

# 启动脚本
#!/bin/bash
#启动zookeeper
/DATA/elk/kafka/bin/zookeeper-server-start.sh /DATA/elk/kafka/config/zookeeper.properties &
# 3s后启动kafka
sleep 3
#启动kafka
/DATA/elk/kafka/bin/kafka-server-start.sh /DATA/elk/kafka/config/server.properties &

# 停止脚本
#!/bin/bash
#关闭kafka
/DATA/elk/kafka/bin/kafka-server-stop.sh /DATA/elk/kafka/config/server.properties &
# 3s后关闭zk
sleep 3
#关闭zookeeper
/DATA/elk/kafka/bin/zookeeper-server-stop.sh /DATA/elk/kafka/config/zookeeper.properties &



并给脚本赋予权限
chmod +x 脚本名称.sh

4.执行启动脚本，待服务起来后可以创建topic试下效果：

# 进入到kafka bin目录
cd /DATA/elk/kafka/bin
# 创建topic 出现Created topic ldp-test.说明创建成功
sudo ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic ldp-test
# 查看topic
sudo ./kafka-topics.sh --list --zookeeper localhost:2181
# 删除topic
sudo ./kafka-topics.sh --delete --zookeeper localhost:2181 --topic ldp-test

5.使用终端打开两个窗口来验证生产者和消费者的通信

# 生产者
sudo ./kafka-console-producer.sh --broker-list localhost:9092 --topic ldp-test

#消费者
sudo ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ldp-test --from-beginning

生产者发送消息

消费者接收消息

重启filebeat，执行查看topic命令看之前filebeat中配置的topic名字有没有出现在列表里

查看消费者信息，有出现信息说明filebeat到kafka数据流通成功。

四、logstash安装与配置

1.将logstash包解压到指定目录，编辑文件
#解压logstash  解压到指定目录
tar -zxf logstash-7.4.0.tar.gz -C /DATA/elk/

#更名
mv logstash-7.4.0 logstash

cd config/
# conf文件放置在该目录下
mkdir conf.d
cd conf.d
# 创建conf文件，名字自定义
touch ldp-pipeline.conf
vi ldp-pipeline.conf

input {
  kafka {
    bootstrap_servers => "xxx.xxx.xxx.xxx:9092"
    topics => "test-ldp"
    group_id => "ldp-group"
    client_id => "ldp-client"
    codec => "json"
    id => "test-ldp"
  }
}

output {
  elasticsearch {
    hosts => ["xxx.xxx.xxx.xxx:9200"]
    index => "test-ldp-index-%{+YYYY.MM.dd}"
  }
}

2.修改logstash.yml文件

node.name: xxx.xxx.xxx.xxx
path.data: /DATA/elk/logstash/data
http.host: xxx.xxx.xxx.xxx
http.port: 9600
log.level: info
path.logs: /DATA/logs/logstash

3.修改pipelines.yml文件

- pipeline.id: test-ldp
  path.config: "/DATA/elk/logstash/config/conf.d/ldp-pipeline.conf"
  pipeline.workers: 1
  pipeline.batch.size: 125
  pipeline.batch.delay: 50
  queue.type: memory

五、es安装与配置

1.创建新用户，因为es启动不能是root用户，所以我们创建一个新的用户

添加组：groupadd elk
添加用户：useradd elk -m -s /bin/bash -d /home/elk -g elk
给elk添加密码：passwd elk
授权： chown -R elk:elk /DATA/elk

2.将es包解压至指定目录，配置jdk环境，本例采用jdk1.8

#将es解压到指定目录
tar -zxvf elasticsearch-7.4.0-linux-x86_64.tar.gz -C /DATA/elk

#更名
mv elasticsearch-7.4.0 elasticsearch


# 修改es配置文件
cd /DATA/elk/elasticsearch/config
vi elasticsearch.yml

node.name: node-1
path.logs: /DATA/logs/es
network.host: xxx.xxx.xxx.xxx
http.port: 9200
cluster.initial_master_nodes: ["node-1"]

3.变更jdk，es自带jdk我们配置使用es自带jdk

vi elasticsearch-env 

# now set the path to java
#if[!-z"$JAVA HOME”];then
# JAVA="$JAVA HOME/bin/java
  # JAVA TYPE="JAVA HOME
#else
 if["$(uname s)"="Darwin" ]; then
    # macOs has a different structure
    JAVA="$ES HOME/jdk.app/Contents/Home/bin/java
 else
    JAVA="$ES HOME/jdk/bin/java'
 fi
#fi

4.启动es

#在bin目录下启动es
nohup ./elasticsearch &
启动时报错解决方案：
#a.# 若出现ERROR: bootstrap checks failed[1]: max file descriptors [4096] for elasticsearch process is too low。。。。
#解决方案:
   # 切换到root用户修改
    vim /etc/security/limits.conf
   # 在最后面追加下面内容
    *               soft    nofile          65536
    *               hard    nofile          65536
    *               soft    nproc           4096
    *               hard    nproc           4096
   # 退出重新登录检测配置是否生效:
    ulimit -Hn
    ulimit -Sn
    ulimit -Hu
    ulimit -Su

#b.重新启动出现如下错误
#ERROR: max number of threads [3802] for user [chenyn] is too low,increase to at least [4096]**
#   解决方案:
   #进入limits.d目录下修改配置文件。
    vim /etc/security/limits.d/20-nproc.conf 
   # 修改为 启动ES用户名 soft nproc 4096
   
#c.重新启动出现如下错误
#ERROR: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]**
   解决方案:
    vim /etc/sysctl.conf
    vm.max_map_count=655360
   #执行以下命令生效：
    sysctl -p

5.启动logstash

#es完成后启动logstash
cd /DATA/elk/logstash/bin
nohup ./logstash &

观察日志有无报错信息

通过elasticsearch看整个流程数据是否正确流通
查看索引：curl ‘xxx.xxx.xxx.xxx:9200/_cat/indices?v’

根据索引查询数据：curl -XGET ‘xxx.xxx.xxx.xxx:9200/test-ldp-index/_search?pretty’

这就证明整个流程下来数据流通无误。

六、kibana安装

1.解压包到指定目录，修改kibana.yml配置文件。
#解压到指定目录
 tar -zxvf kibana-7.4.0-linux-x86_64.tar.gz -C /DATA/elk
#更名
 mv kibana-7.4.0-linux-x86_64 kibana

cd config
vi kibana.yml

server.host: "xxx.xxx.xxx.xxx"   #允许访问ip  如果不限制则 0.0.0.0
elasticsearch.hosts: ["http://esip:9200"]

2.启动kibana

cd  /DATA/elk/kibana/bin
./kibana &

创建索引

查看数据