1.背景

高日志压力情况下，为了避免Logstash直接向ElasticSearch写入的压力过大导致中间过程堆积，采用Logstash——Kafka——Logstash——ElasticSearch的方案，由Kafka缓冲一下数据压力。

2.ELK的配置

参见《ELK搭建及Java程序接入》

2.1.下载

官网下载Elasticsearch、Logstash、Kibana

https://www.elastic.co/

https://www.elastic.co/cn/downloads/

我下载的都是7.1.1版本，最好统一版本

（注意，这几个都是试用版，过一段时间会过期，如果需要长期使用，得下载文末的oss版本）

2.2.关闭防火墙

systemctl stop firewalld

systemctl disable firewalld

2.3.安装elasticsearch

在 Linux 环境中，elasticsearch 不允许以 root 权限来运行。

如果以 root 身份运行 elasticsearch，会提示这样的错误：

can not run elasticsearch as root

解决方法：使用非 root 权限账号运行 elasticsearch

# 创建用户组

groupadd elk

# 创建新用户，-g elk 设置其用户组为 elk，-p elk 设置其密码为 elk

useradd elk -g elk -p elk

# 更改 /opt 文件夹及内部文件的所属用户及组为 elk:elk

chown -R elk:elk /opt # 假设你的 elasticsearch 安装在 opt 目录下

# 切换账号

su elk

解压elasticsearch

tar xzvf elasticsearch-7.1.1-linux-x86_64.tar.gz

运行

cd elasticsearch-7.1.1

bin/elasticsearch

测试

curl http://localhost:9200

默认只能用localhost访问，如需开启外部访问，需修改配置

vi config/elasticsearch.yml

去掉network.host的注释并修改为本机ip

network.host: 0.0.0.0

discovery.seed_hosts: ["127.0.0.1","[::1]"]

此时启动可能会报错

ERROR: [3] bootstrap checks failed

[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65535]

[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

如果不配置discovery.seed_hosts会报错

[3]: the default discovery settings are unsuitable for production use; at least one of [discovery.seed_hosts, discovery.seed_providers, cluster.initial_master_nodes] must be configured

vm.max_map_count 不低于 262144

问题：vm.max_map_count 表示虚拟内存大小，它是一个内核参数。elasticsearch 默认要求 vm.max_map_count 不低于 262144。

max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

解决方法：

你可以执行以下命令，设置 vm.max_map_count ，但是重启后又会恢复为原值。

sysctl -w vm.max_map_count=262144

持久性的做法是在 /etc/sysctl.conf 文件中修改 vm.max_map_count 参数：

echo "vm.max_map_count=262144" > /etc/sysctl.conf sysctl -p

nofile 不低于 65536

问题： nofile 表示进程允许打开的最大文件数。elasticsearch 进程要求可以打开的最大文件数不低于 65536。

解决方法：

echo "* soft nofile 65536" > /etc/security/limits.conf

echo "* hard nofile 131072" > /etc/security/limits.conf

reboot重启后即可正常启动

后台启动

nohup bin/elasticsearch > service.out&

跟踪后台启动的输出

tail -f service.out

看进程

[elk@localhost xx]$ ps -ef|grep opt

elk 2095 1750 11 13:42 pts/0 00:02:51 /opt/elasticsearch-7.1.1/jdk/bin/java -Xms1g -Xmx1g -XX:+UseConc……/opt/elasticsearch-7.1.1/lib/* org.elasticsearch.bootstrap.Elasticsearch

elk 2183 2095 0 13:42 pts/0 00:00:00 /opt/elasticsearch-7.1.1/modules/x-pack-ml/platform/linux-x86_64/bin/controller

2.4.安装Logstash

解压

tar xzvf logstash-7.1.1.tar.gz

运行

cd logstash-7.1.1

bin/logstash -f config/logstash-sample.conf

或后台启动

nohup bin/logstash -f config/logstash-sample.conf > service.out&

如果没有JAVA_HOME，需设置JAVA_HOME，注意使用root用户

vi /etc/profile

添加如下内容

export JAVA_HOME=/opt/elasticsearch-7.1.1/jdk

export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

export PATH=$PATH:$JAVA_HOME/bin

配置立即生效执行

source /etc/profile

看进程

$ ps -ef|grep opt

elk 2460 2317 68 14:01 pts/1 00:03:37 /opt/elasticsearch-7.1.1/jdk/bin/java ……/opt/logstash-7.1.1/logstash-core/lib/jars/slf4j-api-1.7.25.jar org.logstash.Logstash -f config/logstash-sample.conf

2.5.安装Kibana

解压

tar xzvf kibana-7.1.1-linux-x86_64.tar.gz

cd kibana-7.1.1-linux-x86_64

需要修改elasticsearch的ip端口则改一下配置

vi config/kibana.yml

去掉这两行的注释并修改

server.host: "192.168.91.149" #需修改才能从其他机器访问

elasticsearch.hosts: ["http://localhost:9200"]

运行

bin/kibana

或后台启动

nohup bin/kibana > service.out&

访问地址

http://localhost:5601

看进程

[elk@localhost kibana-7.1.1-linux-x86_64]$ ps -ef |grep node

elk 2658 1750 99 14:09 pts/0 00:00:12 bin/../node/bin/node --no-warnings --max-http-header-size=65536 bin/../src/cli

2.6.Java日志输出到Logstash

cd logstash-7.1.1

vi config/logstash.conf

添加如下内容

input {
beats {
port => 5044
}
tcp {
# host:port就是上面appender中的 destination，
# 这里其实把logstash作为服务，开启9250端口接收logback发出的消息
host => "0.0.0.0" port => 9250 mode => "server" tags => ["tags"] codec => json_lines
}
}

output {
elasticsearch {
hosts => ["http://localhost:9200"]
}
}
kill掉原来的logstash后启动

nohup bin/logstash -f config/logstash.conf > service.out&

Java代码修改

pom.xml增加依赖

<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>6.1</version>
</dependency>
logback.xml按需调整

<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">

<destination>192.168.91.149:9250</destination>
<encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"/>
</appender>
<root level="INFO">

<appender-ref ref="STDOUT"/>
<appender-ref ref="LOGSTASH" />
</root>
此时启动java程序，如果启动成功，日志就进入es了

日志查看

进入kibana web页面

点击左侧Discover按钮，会自动进入创建索引，创建索引logstash-*

点击下一步Next step

Time Filter field name选择@timestamp，点击Create index pattern按钮等待创建索引成功

此时再次点击左侧Discover按钮，进入熟悉的索引查询页面了

区分应用

实际应用过程中，多个app日志混在一起，需要增加额外入库字段进行区分，此时修改java程序的logback.xml，添加customFields配置节即可。

<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">

<destination>192.168.91.149:9250</destination>
<encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder">
<customFields>{"appname":"whqtest"}</customFields>
</encoder>
</appender>

logback.xml也可以采用变量的写法

<configuration>
   
   <property name="APP_ID" value="whqtest"/>
   ………………
   <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
   
   <destination>192.168.91.149:9250</destination>
   <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder">
   <customFields>{"appname":"${APP_ID}"}</customFields>
   </encoder>
   </appender>
   ………………
</configuration>

2.7.OSS版本

一段时间后启动kibana报错

Elasticsearch cluster did not respond with license information.

需要下载oss版本
https://www.elastic.co/cn/downloads/elasticsearch-oss

https://www.elastic.co/cn/downloads/logstash-oss

https://www.elastic.co/cn/downloads/kibana-oss

OSS版部署问题
如果连接es超时

vi es的目录/config/jvm.options

增大-Xms和-Xmx到2g

增大es连接超时时间

vi config/kibana.yml

elasticsearch.requestTimeout: 90000

连接es返回错误：master_not_discovered_exception

es服务器修改

vi /etc/hostname 设置机器名为node1

node1

vi /etc/hosts

127.0.0.1 node1

vi es目录/config/elasticsearch.yml

node.name: node1

discovery.seed_hosts: ["127.0.0.1"]

cluster.initial_master_nodes: ["node1"]
————————————————
版权声明：本文为CSDN博主「洪岐」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/whq12789/article/details/92832436

3.Kafka的配置

3.1.zookeeper搭建

参见《centos7 zookeeper3.5.6单点部署》

官网下载

https://zookeeper.apache.org/
下载
https://zookeeper.apache.org/releases.html
直接部署的包
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/current/apache-zookeeper-3.5.6-bin.tar.gz
源码包（下载这个比较麻烦，需要安装maven，然后mvn package进行编译）
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/current/apache-zookeeper-3.5.6.tar.gz

部署
#安装jdk1.8
yum install -y java-1.8.0-openjdk

#部署zookeeper
tar xvzf apache-zookeeper-3.5.6-bin.tar.gz
mv apache-zookeeper-3.5.6-bin /opt
cd /opt/apache-zookeeper-3.5.6-bin/conf
cp zoo_sample.cfg zoo.cfg
vi zoo.cfg
修改
dataDir=/data/zookeeper
#该目录为数据保存目录
mkdir -p /data/zookeeper

cd /opt/apache-zookeeper-3.5.6-bin/
bin/zkServer.sh start
输出如下，STARTED表示成功。

ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.5.6-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

连接测试

bin/zkCli.sh -server 127.0.0.1:2181
出现如下命令输入框则表示连接成功
WatchedEvent state:SyncConnected type:None path:null
[zk: 127.0.0.1:2181(CONNECTING) 0]
此时输入
ls /
可以查看zookeeper根目录有一个zookeeper文件：
[zk: 127.0.0.1:2181(CONNECTED) 2] ls /
[zookeeper]
————————————————
版权声明：本文为CSDN博主「洪岐」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/whq12789/article/details/103122786

3.2.kafka搭建

参见《centos7 kafka2.3.1单点部署》

kafka下载

http://kafka.apache.org/downloads
http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.1/kafka_2.12-2.3.1.tgz

部署

tar xvzf kafka_2.12-2.3.1.tgz
mv kafka_2.12-2.3.1 /opt/
cd /opt/kafka_2.12-2.3.1/
vi config/server.properties
修改

listeners=PLAINTEXT://192.168.81.145:9092 #注意，这里一定要有客户端可访问的ip，否则非本机连接调用第二次会异常
log.dirs=/data/kafka-logs

启动
bin/kafka-server-start.sh config/server.properties &
如果输出出现
[2019-11-18 15:17:08,052] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
表示成功启动

停止服务
bin/kafka-server-stop.sh

测试

启动生产者
bin/kafka-console-producer.sh --broker-list 192.168.81.145:9092 --topic test
然后随便输入几行消息

启动消费者
bin/kafka-console-consumer.sh --bootstrap-server 192.168.81.145:9092 --topic test --from-beginning
消费者会获取到生产者输入的几行消息
————————————————
版权声明：本文为CSDN博主「洪岐」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/whq12789/article/details/103123372

4.整合

接下来，将logstash拷贝出一份改名logstash2
logstash目录的配置修改为：
vi config/logstash.conf
input {
tcp {
# 这里其实把logstash作为服务，开启9250端口接收logback发出的消息
host => "0.0.0.0" port => 9250 mode => "server" tags => ["tags"] codec => json_lines
}
}

output {
kafka {
bootstrap_servers => "192.168.81.145:9092,192.168.81.145:9093,192.168.81.145:9094" # kafka 集群主机
topic_id => "web-log" # 主题名称，会自动创建
#compression_type => "snappy" # 压缩方式
codec => json #注意这里向kafka输出日志为json格式
}
}

logstash2目录的配置修改：
vi config/logstash.conf
input {
kafka {
bootstrap_servers => "192.168.81.145:9092,192.168.81.145:9093,192.168.81.145:9094" # kafka 集群主机
topics_pattern => "web-log"
consumer_threads => 5
decorate_events => true
auto_offset_reset => "latest"
group_id => "logstash"
codec => json #从kafka中直接取json格式日志
}
}

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

接下来kill掉logstash，然后两个目录下分别执行
nohup bin/logstash -f config/logstash.conf > service.out&
启动两个logstash服务

启动测试的java程序后
bin/kafka-console-consumer.sh --bootstrap-server 192.168.81.145:9092 --topic web-log
这个就能接收到第一个logstash输入的json日志

kibana中也可以看到elasticsearch中“sc-日期”的索引数据了。

配置过程中参见了
https://www.elastic.co/guide/en/logstash/current/index.html
展开Input plugins和Output plugins有相应的配置说明。