写在前面的话：这是本人的第一篇技术文章，主要是结合实际环境和个人经验总结而成，以此记录以备忘。内容如有谬误，请各位不吝指正。如果对于写文章有任何建议，欢迎交流，非常谢谢！
一、背景描述
由于私有云平台的基础架构涉及多地多中心，有大量的ESXi虚拟化主机，随着时间增长和体量越来越大，故障也越来越多。需要搭建一套完整的适应于分布式基础架构的日志管理系统，以实现所有ESXi主机日志的集中监控管理，并实现告警。出于成本和技术可控性出发，考虑选用ELK技术栈来实现日志管理平台建设。
二、系统架构设计和资源规划
（一）、架构考虑因素：可扩展性、资源合理利用、性能瓶颈、对其他业务可能产生的负面影响
1、可扩展性：可以更方便、快速扩展各不同类型节点的资源，且可扩展不影响现有系统稳定运行。
2、资源合理利用：各个不同类型的节点需要使用资源的不相同，如数据节点对内存需求较大、Coordinate查询消耗内存；大量数据写入，对磁盘的要求比较高；ES集群作为查询和存储数据对于CPU要求较高
3、性能瓶颈：大量消息传入需要缓冲和防止logstash挂掉，无法正常接收数据。
4、对其他业务影响：因为虚拟机化环境没有严格划分区域，可能会对其他业务产生影响，特别是vSAN的IO争用。
5、Index与Shard的设计，并不是一下子就可以完成的，需要不断预估、测试、调整、再测试来达到最终模型。整个过程大致如下图时间轴所示，分为以下几个阶段，当然期间有很多阶段是需要迭代的。
第一阶段，Index的设计。这一阶段是完成业务需求的分解与设计，之后剩下的就是如何构建Elasticsearch集群来承载这样的需求了。
第二阶段，预估数据量。预估单个Index的数据量，结合需要保留的数据周期得到整体的数据量。
第三阶段，预估机型。根据业务的场景，来决定选择CPU增强型、内存增强型还是通用型机器（针对云服务）。
第四阶段，预估机器数。知道了数据量，也就知道了最少需要多少台机器。在预估磁盘空间时，需要考虑另外两个因素：一个是Replica，即副本的个数，我们1个就够了；另一个是预留30%的磁盘空间，这部分空间既是预留给系统使用的，也是为磁盘告警的处理预留足够的时间。
第五阶段，预估Primary Shard个数。
第六阶段，部署、测试。根据性能测试和业务的特殊情况，做适当调整，调整可能包括Primary Shard个数、机型、机器个数等。
针对ESXi日志只能使用loginsite方式获取，且异地机房网络不稳定，故使用如下架构
Loginsite in clients àlogstash（proxy） à Kafka集群（proxy） à Logstash（proxy）àKafka集群（Server） àElasticSearch（Server） àKibana
其他操作系统日志采用如下方式：
Filebeat à Kafka集群 à Logstash àElasticSearch àKibana
架构图如下：
Kafka集群初步设定为3台服务器，架构如下：
（二）、异地多活架构参考，作为后续大规模扩展后的升级架构：
（三）、硬件配置：
Kafka集群：3台物理机（暂用虚拟机），CPU：8C，Mem：16GB，Disk1：100GB，Disk2：1TB
Logstash：3台虚拟机，CPU：8C，Mem：16GB，Disk1：100GB，Disk2：500GB
ES集群：8台物理机，CPU：32C，Mem：196GB，Disk：10TB
{ES规划参考标准：SN(分片数) = IS(索引大小) / 30
NN(节点数) = SN(分片数) + MNN(主节点数[无数据]) + NNN(负载节点数)}
Kibana集群：1台虚拟机（暂用一台，后续采用LVS+Keepalived方式进行高可用扩展），CPU：4C，Mem：8GB，Disk：100GB
（四）、操作系统和软件版本：
所有操作系统选择CentOS7.6，数据文件系统XFS
ELK选择当前7.6.2版本
（五）、网络规划：
需要ESXi管理网段到Kafka所在网段至少单向连通。
Kafka：12.168.13.201-203 主机名：Kafka1、Kafka2、Kafka3
Logstash：12.168.13.204-206 主机名：Logstash1、Logstash2、Logstash3
ES集群：12.168.13.207-209、211-217 主机名：ES1、ES2、ES3、ES4、ES5、ES6、ES7、ES8、ES9、ES10；水土物理机ES集群：17.21.30.1-8 主机名：ST-ES1、ST-ES2、ST-ES3、ST-ES4、ST-ES5、ST-ES6、ST-ES7、ST-ES8
Kibana：12.168.13.210 主机名：Kibana1
除水土机房外，其他各机房使用代理集群方式，避免网络丢包导致日志丢失问题，每个包含2台logstash和3台Kafka集群
三、安装部署
（一）、部署Kafka集群：
1、修改主机名
[root@localhost monitor]# hostnamectl set-hostname kafka1
2、配置代理或者直接下载软件包
kafka下载地址：http://mirror.bit.edu.cn/apache/kafka/2.5.0/
Java_JDK下载地址：https://www.oracle.com/java/technologies/javase-jdk8-downloads.html
3、安装JKD
在系统根目录下创建名称为java的文件夹
[root@kafka1 monitor]# mkdir /java
[root@kafka1 monitor]# mkdir /data
执行解压命令
[root@kafka1 java]# cd /java/
[root@ kafka1 java]# tar -xzvf jdk-8u161-linux-x64.tar.gz
4、配置Java环境变量
[root@kafka1 java]# vi /etc/profile
也需要在此文件中添加/etc/bashrc
添加：
# set Java environment
JAVA_HOME=/java/jdk1.8.0_161
PATH=$JAVA_HOME/bin:$PATH
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export PATH
export CLASSPATH
[root@kafka1 java]# source /etc/profile
5、检测JDK是否安装成功
[root@kafka1 java]# java –version
[root@kafka1 java]# javac
6、安装kafka
[root@kafka1 data]# mkdir /data/kafka
tar -xzvf kafka_2.13-2.5.0.tgz -C /data/kafka/
[root@kafka1 config]# cd /data/kafka/kafka_2.13-2.5.0/config
[root@kafka1 config]# vim server.properties
设置项如下：
broker.id=1
listeners=PLAINTEXT://12.168.13.201:9092
advertised.listeners=PLAINTEXT://12.168.13.201:9092
log.dirs=/data/kafka/log
# 注意这里的是 kafka 的存储目录。并不是存放日志的目录，当你启动kafka的时候，会自动在当前目录下生成一个 log 专门存放 kafka 的日志文件。如果没有需创建
zookeeper.connect=12.168.13.201:2181,12.168.13.202:2181,12.168.13.203:2181#这里可以是主机名，也可以是 ip，但是填写的是主机名，就必须在 /etc/hosts 文件下进行解析才可以生效
7、复制文件到slave节点上
[root@kafka1 config]# scp -r /data/ root@12.168.13.202:/
[root@kafka1 config]# scp -r /data/ root@12.168.13.203:/
[root@kafka1 config]# scp -r /java/ root@12.168.13.202:/
[root@kafka1 config]# scp -r /java/ root@12.168.13.203:/
8、修改slave节点上的配置文件：
[root@kafka1 config]# vim server.properties
修改以下选项：
broker.id=1
listeners=PLAINTEXT://12.168.13.201:9092
advertised.listeners=PLAINTEXT://12.168.13.201:9092
9、配置zookeeper集群
修改配置文件/data/kafka/kafka_2.13-2.5.0/config/zookeeper.properties
内容如下：
dataDir=/data/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false
server.1=kafka1:2888:3888
server.2=kafka2:2888:3888
server.3=kafka3:2888:3888
tickTime=1000
initLimit=10
syncLimit=5
10、在/data/zookeeper创建myid文件
touch myid
对应不同的brocker.id值在文件中添加对应数值
echo 1 >/data/zookeeper/myid
11、启动zookeeper服务
[root@kafka1 monitor]#/data/kafka/kafka_2.13-2.5.0/bin/zookeeper-server-start.sh -daemon /data/kafka/kafka_2.13-2.5.0/config/zookeeper.properties
12、启动kafka服务
[root@kafka1 monitor]#/data/kafka/kafka_2.13-2.5.0/bin/kafka-server-start.sh -daemon /data/kafka/kafka_2.13-2.5.0/config/server.properties
13、修改其他主机参数
（二）、部署Logstash：
1、上载安装包rpm -ivh logstash-7.6.2.rpm
2、安装软件包：
[root@localhost monitor]# rpm -ivh logstash-7.6.2.rpm
3、安装jdk，参见上面java环境设置
4、添加环境变量，由于是rpm包安装的，安装路径在/usr/share/logstash/
编辑/etc/bashrc添加如下内容：
export PATH=$PATH:/usr/share/logstash/bin
# set Java environment
JAVA_HOME=/java/jdk1.8.0_161
PATH=$JAVA_HOME/bin:$PATH
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export PATH
export CLASSPATH
5、设置服务开机自启动
systemctl enable logstash
systemctl start logstash
此处可能会报错，找不到环境变量，但检查java环境变量正常的：
这是由于rpm安装，需要建立logstash账户，服务通过logstash账号启动，且指定了环境变量文件导致的。
注释掉User和Group后可以root启动；/etc/sysconfig/logstash文件原本不存在，可以去掉，修改/etc/default/logstash，添加java环境变量。重载systemctl，重启服务即可。
6、修改配置文件：
[root@localhost monitor]# cat /etc/logstash/conf.d/test.conf
input{
stdin{}
}
output{
elasticsearch{
hosts => ["12.168.13.207:9200","12.168.13.208:9200","12.168.13.209:9200"]
}
}
（三）、部署ES集群：
1、修改主机名，创建目录/java、/data
[root@es1 /]# hostnamectl set-hostname ES1; mkdir –p /java /data/ES
2、安装java环境设置环境变量，这里通过下载gz包的方式，直接解压使用
[root@es1 /]#tar -xzvf jdk-8u161-linux-x64.tar.gz
3、设置jvm环境变量（rpm包安装已自带jdk环境）
参考kafka集群安装
4、安装ES软件
[root@es1 /]#rpm -ivh elasticsearch-7.6.2-x86_64.rpm
5、修改ES配置文件
[root@es1 /]#vim /etc/elasticsearch/elasticsearch.yml
修改项如下：
cluster.name: zyc-es-cluster
node.name: es1
path.data: /data/ES
path.logs: /var/log/elasticsearch
network.host: 12.168.13.207
bootstrap.memory_lock: true
discovery.seed_hosts: ["es2", "es3"]
cluster.initial_master_nodes: ["es1"] 此项为第一次启动前设置，后续重启集群和添加节点时不应设置。
6、设置java堆栈大小/etc/elasticsearch/ jvm.options
不应将以下像个参数值超过物理内存的50%，因JVM限制物理内存不要超过32GB，超过32GB会出现性能下降情况，我们初始搭建16GB内存，故设置如下：
-Xms7g
-Xmx7g
修改heap dump路径：
-XX:HeapDumpPath=/data/ES
7、重要的操作系统配置项：
a、禁用swap
vim /etc/fstab注释掉swap行
sysctl –w swappiness=0
sysctl –p
vim /etc/sysctl.conf添加vm.swappiness = 0
锁定bootstrap内存，这一步如果在修改ES配置中这一步已做，则不需要再设置：
vim /etc/elasticsearch/elasticsearch.yml
bootstrap.memory_lock: true
b、增加文件描述符
vim /etc/security/limit.d/30-nofile.conf
* soft nofile 65535
* hard nofile 100000
c、确保有足够的虚拟内存（可能默认参数已满足）
vim /etc/sysctl.conf
vm.max_map_count=262144
d、确保足够的线程（可能默认参数已满足）
vim /etc/security/limit.d/20-noproc.conf
* soft nproc 4096
root soft nproc unlimited
1、启动服务，我们此处用rpm安装的，直接用systemd控制
systemctl enable elasticsearch
systemctl start elasticsearch
此处可能会产生报错，需要将/data和/data/ES目录赋予777权限。
如果报以下错误：
May 06 10:16:51 es1 elasticsearch[129162]: ERROR: [1] bootstrap checks failed
May 06 10:16:51 es1 elasticsearch[129162]: [1]: memory locking requested for elasticsearch process but memory is not locked
则需要修改/usr/lib/systemd/system/elasticsearch.service配置文件如下
vim /usr/lib/systemd/system/elasticsearch.service
增加：
# Specifies the memory
LimitMEMLOCK=infinity
2、其他ES主机设置同上。
3、测试ES集群是搭建成功
[root@es1 monitor]# curl -i http://12.168.13.207:9200/_cluster/health?pretty
如上图所示则成功。
4、安装head插件（在更早版本中会默认安装）
（四）、搭建Kibana：
1、到官网下载rpm安装包https://www.elastic.co/
2、安装rpm包
rpm -ivh kibana-7.6.2-x86_64.rpm
3、添加环境变量vim /etc/bashrc
export PATH=$PATH:/usr/share/kibana/bin
source /etc/bashrc
4、修改配置文件
vim /etc/kibana/kibana.yml
修改如下内容：
elasticsearch.hosts: ["http://12.168.13.207:9200"]
如果需要配置集群方式则需要做如下操作：
并将kibana.yml中的对应项修改为：
elasticsearch.hosts:
- http://12.168.13.207:9200
- http://12.168.13.208:9200
- http://12.168.13.209:9200
5、启动服务和设置开机自启动
systemctl start kibana
systemctl enable kibana.service
6、通过浏览器进行web测试：
http://12.168.13.210:5601/
（五）、配置logstash将日志存储到Kafka
[root@localhost monitor]# cat /etc/logstash/conf.d/
input {
stdin {}
}
output {
kafka {
bootstrap_servers => "12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092"
topic_id => "test"
}
}
四、测试整个集群数据流通性
（一）、在logstash1上，测试
1、编辑测试配置文件/etc/logstash/conf.d/：
input {
stdin {}
}
output {
kafka {
bootstrap_servers => "12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092"
topic_id => "test"
}
}
2、运行测试配置文件：
logstash -f /etc/logstash/conf.d/test.conf
（二）、在kafka1上测试：
1、查看运行进程命令：jps可以列出kafka进程和zookeeper的进程ID号
2、创建测试topic命令如下：
/data/kafka/kafka_2.13-2.5.0/bin/kafka-topics.sh --zookeeper 12.168.13.201:2181, 12.168.13.202:2181, 12.168.13.203:2181 --create --topic test --replication-factor 3 --partitions 3
3、、查看topic命令如下：
/data/kafka/kafka_2.13-2.5.0/bin/kafka-topics.sh --list esxi --zookeeper kafka1:2181,kafka2:2181,kafka3:2181
删除topic命令如下：
/data/kafka/kafka_2.13-2.5.0/bin/kafka-topics.sh --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --delete --topic test
4、在kafka1上模拟消息发送
/data/kafka/kafka_2.13-2.5.0/bin/kafka-console-producer.sh --broker-list 12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092 --topic test
5、在kafka2上查看消息接收
/data/kafka/kafka_2.13-2.5.0/bin/kafka-console-consumer.sh --bootstrap-server 12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092 --topic test --from-beginning
6、在此记录一个遇到的错误：
需要进入到/data/kafka/log/meta.properties中将对应的cluster.id=值改为：1bBL5WrLQ4Klu4GEEqvCbg，然后重启zookeeper和kafka。
如果遇到如下问题：
Error while fetching metadata with correlation id xx …..
需到server.properties中添加auto.create.topics.enable=true，虽然官方文档显示，此项为默认开启，但不知为何要显示设置此项。如果设置后，则无法手动删除topic。
五、启用x-packt模块
（一）、ES上设置，参见https://www.elastic.co/guide/en/elasticsearch/reference/7.6/configuring-tls.html#node-certificates
1、目前ES7已免费开放base版本的x-pack，只需要在elasticsearch.yml的配置文件中增加如下行就可以开启：
xpack.security.enabled: true
2、在一个节点上创建CA证书：TLS需要X.509证书（X.509 证书是一个数字证书，它使用 X.509 公有密钥基础设施标准将公有密钥与证书中包含的身份相关联。X.509 证书由一家名为证书颁发机构 (CA) 的可信实体颁发。CA 持有一个或多个名为 CA 证书的特殊证书，它使用这种证书来颁发 X.509 证书。只有证书颁发机构才有权访问 CA 证书）才能对与之通信的应用程序执行加密和身份验证。 为了使节点之间的通信真正安全，必须对证书进行验证。在Elasticsearch集群中验证证书真实性的推荐方法是信任签署证书的证书颁发机构（CA）。这样，只需要使用由同一CA签名的证书，即可自动允许该节点。
bin/elasticsearch-certutil ca -pem
生成的文件默认是一个压缩包，解压后里面包含key和crt文件。生成文件目录在/usr/share/elasticsearch/,解压文件，并将ca目录全部复制到目录/etc/elasticsearch/下，并修改对应权限，确保elasticsearch可以访问。
3、在elasticsearch.yml中添加如下配置项：
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.key: /etc/elasticsearch/ca/ca.key
xpack.security.transport.ssl.certificate: /etc/elasticsearch/ca/ca.crt
xpack.security.transport.ssl.certificate_authorities: [ /etc/elasticsearch/ca/ca.crt ]
xpack.security.enabled: true
4、将第一个节点生成的/ca/ca.crt和ca.key文件拷贝到其他节点对应位置，并设置上面3步骤中的参数。
5、将解压的ca.crt和ca.key复制到其他节点，并配置上面一步的配置，重启ES服务(非常重要，否则无法添加内置用户)。
6、第一次修改内置用户密码命令如下，注意，设置elastic用户密码后，引导密码将不再有效，也不能通过elasticsearch-setup-passwords命令再次改密码，需要使用Kibana中的‘管理à用户’页面或‘更改密码API’来设置内置用户的初始密码。并且需要提供elastic用户及其引导密码才能登陆Kibana或运行API。但必须在启动elasticsearch之前设置bootstrap.password。
elasticsearch-keystore add "bootstrap.password"
修改内置用户密码命令如下：
elasticsearch-setup-passwords interactive
7、在每个节点上将保护密码注入keystone，如果没有设置就直接回车。（注意：如果用pem方式不用执行以下命令）
elasticsearch-keystore add xpack.security.transport.ssl.keystore.secure_password
elasticsearch-keystore add xpack.security.transport.ssl.truststore.secure_password
8、启用http的ssl参考官网链接：
https://www.elastic.co/guide/en/elasticsearch/reference/7.9/configuring-tls.html
1）将文件解压出来的http.p12文件，复制到/etc/elasticsearch/目录。
2）在配置文件elasticsearch.yml中添加如下行：
xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.keystore.path: /etc/elasticsearch/http.p12
3）添加保护密码，生成证书时设置的：
elasticsearch-keystore add xpack.security.http.ssl.keystore.secure_password
4）在其他节点做同样的操作。
5）Kibana端配置https连接，注意Kibana只支持.pem格式，将
9、logstash输入端配置，参见：
https://www.elastic.co/guide/en/logstash/7.8/ls-security.html
https://blog.csdn.net/LSY929981117/article/details/107714001
1）将ES上/usr/share/elasticsearch/kibana/elasticsearch-ca.pem复制到/etc/logstash/目录下
2）配置logstash.yml文件如下：
xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.username: logstash_system
xpack.monitoring.elasticsearch.password: Xxxx#LOG
xpack.monitoring.elasticsearch.hosts: ["https://17.21.30.1:9200", "https://17.21.30.2:9200", "https://17.21.30.3:9200", "https://17.21.30.4:9200", "https://17.21.30.5:9200", "https://17.21.30.6:9200", "https://17.21.30.7:9200", "https://17.21.30.8:9200"]
xpack.monitoring.elasticsearch.ssl.certificate_authority: /etc/logstash/elasticsearch-ca.pem
xpack.monitoring.elasticsearch.ssl.verification_mode: certificate
xpack.monitoring.elasticsearch.sniffing: false
3）配置logstash配置输出文件：如下面步骤中的配置所示。
10、故障排除：
1）遇到如下报错，因ES集群中的system_log角色设置问题：
[2020-08-09T17:59:32,860][ERROR][logstash.outputs.elasticsearch][main] Encountered a retryable error. Will Retry with exponential backoff {:code=>403, :url=>"https://12.168.13.214:9200/_bulk"}
暂时解决方法：在logstash输出配置文件output中使用elastic账号
elasticsearch {
hosts => ["12.168.13.207:9200","12.168.13.208:9200","12.168.13.209:9200","12.168.13.211:9200","12.168.13.212:9200","12.168.13.213:9200","12.168.13.214:9200","12.168.13.215:9200","12.168.13.216:9200","12.168.13.217:9200"]
#index => "vc-20200722-test"
index => "vc-%{+YYYY.MM.dd}"
manage_template => false
template => "/etc/logstash/my_template/esxi_all.json"
template_name => "vc_all"
template_overwrite => true
user => "elastic"
password => "Xxxx#ES"
ssl => true
cacert => '/etc/logstash/elasticsearch-ca.pem'
}
2）如果提示CA证书密码错误，删除证书后从其他节点复制过来或者重新制作。另外注意文件权限设置问题，必须要elasticsearch用户可以访问。
11、Kibana端配置ssl连接到ES，修改配置文件/etc/kibana/kibana.yml：
elasticsearch.hosts:
- https://12.168.13.207:9200
- https://12.168.13.208:9200
- https://12.168.13.209:9200
- https://12.168.13.211:9200
- https://12.168.13.212:9200
- https://12.168.13.213:9200
- https://12.168.13.214:9200
- https://12.168.13.215:9200
- https://12.168.13.216:9200
- https://12.168.13.217:9200
elasticsearch.username: "kibana"
elasticsearch.password: "Xxxx#KIB"
elasticsearch.ssl.certificateAuthorities: [ "/etc/kibana/elasticsearch-ca.pem" ]
12、配置到Kibana的ssl访问方式：先生成证书，再将公钥和私钥将证书放在：
/etc/kibana/elk.adminset.local.crt
/etc/kibana/elk.adminset.local.key
在Kibana.yml中配置：
server.ssl.enabled: true
server.ssl.certificate: /etc/kibana/elk.adminset.local.crt
server.ssl.key: /etc/kibana/elk.adminset.local.key
13、开启Kibana对ES集群的监控，使用以下API语句：
PUT _cluster/settings
{
"persistent": {
"xpack.monitoring.collection.enabled": true
}
}
14、
六、ESXi主机日志接入
（一）、因ESXi日志只能通过loginsite方式采集日志（类似于操作系统的rsyslog），故用logstash1这台server作为前端日志采集器，ESXi主机接入到logstash1。具体步骤如下：
1、在logstash1上编辑/etc/logstash/conf.d/esxi-to-kafka.conf
内容如下：
input {
tcp {
port => 514
type => syslog
}
udp {
port => 514
type => syslog
}
}
filter {
grok {
match => {
"message" => "%{SYSLOG5424PRI}%{SYSLOGLINE}"
}
}
}
output {
kafka {
bootstrap_servers => "12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092"
# index => "esxi-%{+YYYY}"
#index => "esxi-%{+YYYY.MM.dd}"
topic_id => "esxi"
}
}
步骤：
1 esxi上检测远程日志服务器端口是否通，tcp、udp
2 esxi上配置远程日志服务器地址
3 远程日志服务器上抓包，查看是否收到包
4 使用rsyslog和logstash收集日志
5 使用logstash grok解析日志格式
6 使用kibana、grafana展示
# 检测esxi到日志服务器端口状态
# tcp
nc -z 12.168.13.204 514
# udp
nc -zu 12.168.13.204 514
# 如果不通，检查日志服务器服务器是否正确启动，esxi防火墙是否允许。
# esxi上开启防火墙允许
esxcli network firewall ruleset set --ruleset-id=syslog --enabled=true
esxcli network firewall refresh
# 查看syslog配置
:~] esxcli system syslog config get
Check Certificate Revocation: false
Default Network Retry Timeout: 180
Dropped Log File Rotation Size: 100
Dropped Log File Rotations: 10
Enforce SSLCertificates: true
Local Log Output: /scratch/log
Local Log Output Is Configured: false
Local Log Output Is Persistent: true
Local Logging Default Rotation Size: 1024
Local Logging Default Rotations: 8
Log To Unique Subdirectory: false
Message Queue Drop Mark: 90
Remote Host: tcp://12.168.13.204:514
Strict X509Compliance: false # 配置远程日志服务器
esxcli system syslog config get
esxcli system syslog config set --loghost='tcp://12.168.13.204:514'
esxcli system syslog reload
esxcli system syslog config get
如果需要重置为空这使用：
esxcli system syslog config set --reset=loghost
esxcli system syslog reload
VC日志接入：
1、打开VC地址对应的5480端口，以root账号登录。
2、在syslog配置页面填入对应的信息即可。
3、/etc/init.d/目录下的VC服务：
vmware-sts-idmd | ignore_priorities | network | vami-lighttp |
vmafdd | vmdird | vmware-stsd | vmware-vcha |
netconsole | vaos | vmcad | vmdnsd |
ESXi服务有：
DCUI | iofilterd-spm | rabbitmqproxy | vmfstraced |
ESXShell | iofilterd-vmwarevmcrypt | rhttpproxy | vmsyslogd |
SSH | iofiltervpd | sdrsInjector | vmtoolsd |
cdp | lacp | sensord | vmware-fdm |
clomd | lbtd | sfcbd-watchdog | vobd |
cmmdsTimeMachine | lwsmd | slpd | vpxa |
cmmdsd | memscrubd | smartd | vsanObserver |
dcbd | nfcd | snmpd | vsandevicemonitord |
ddecomd | nfsgssd | storageRM | vsandpd |
epd | nicmgmtd | swapobjd | vsanmgmtd |
esxui | nscd | upitd | vsantraced |
hbr-agent | ntpd | usbarbitrator | vvold |
hostd | osfsd | vit_loader | wsman |
hostdCgiServer | pcscd | vitd | xorg |
ESXi常见日志类型如下：
日志路径 | 服务描述 | 服务关键字 |
/var/log/vpxa.log | vCenter 代理日志 | Vpxa： |
/var/log/vobd.log | VMware 监测守护进程日志 | 无 |
/var/log/vmkwarning.log | VMkernel 警告日志 | )WARNING: |
/var/log/vmkeventd.log | VMkernel 事件守护进程日志 | vmkeventd[ |
/var/log/vmkernel.log | VMkernel 子系统中的信息 | 无 |
/var/log/vmkdevmgr.log | VMkernel 设备管理器日志 | vmkdevmgr[ |
/var/log/vmauthd.log | vMotion 身份验证守护进程日志 | vmauthd[ |
/var/log/syslog.log | 常规系统日志 | 无 |
/var/log/sysboot.log | 系统引导日志 | sysboot: |
/var/log/shell.log | ESXi Shell 活动日志 | shell[ 和 ESXShell: 和SSH: |
/var/log/hostd.log | 主机代理日志 | hostd[ |
/var/log/fdm.log | Fault Tolerance 管理代理日志 | fdm[ 或 –[ |
/var/log/esxupdate.log | ESX 更新日志文件 | esxupdate: |
/var/log/dhclient.log | DHCP 客户端日志 | dhclient-uw[ |
/var/log/auth.log | 身份验证子系统日志 | sshd[ |
/var/log/rhttpproxy.log | HTML5反向代理服务 | Rhttpproxy: |
ESXi主机上所有后台运行服务及日志信息，绿色为不重要或未启用相关功能的：
服务 | 日志文件名 | 进程关键字 | 功能描述 |
DCUI | 直接控制台UI | ||
ESXShell | /var/log/shell.log | ESXi DCUI Shell | |
SSH | /var/log/shell.log | 远程SSH | |
cdp | 网卡物理适配器（CDP）Cisco Discovery Protocol | ||
clomd | vSAN 群集级别对象管理器 （CLOM）Cluster Level Object Manager Daemon 负责创建新对象， | ||
cmmdsTimeMachine | Cluster Monitoring, Membership, and Directory Service | ||
cmmdsd | vSAN CMMDS（ClusterMonitoring, Membership, and Directory Service，群集监控、成员和目录服务） | ||
dcbd | 网卡物理适配器数据中心桥接功能（DCB）Data Center Bridging | ||
ddecomd | 未查到 | ||
epd | /var/log/epd | 条目持久性守护进程 (EPD) | |
esxui | ESXi单独访问的网页UI | ||
hbr-agent | /var/log/hbragent.log | vSphere Replication（VR备份）给ESXi安装的复制数据加密代理软件 | |
hostd | /var/log/hostd.log | ESXi管理代理 | |
hostdCgiServer | /var/log/hostdCgiServer.log | 一个独立的CGI服务，可处理对URL / cgi-bin的所有HTTP请求。它已从托管进程中移出，并移入了它自己的守护进程，以便在托管死锁时，我们有一种收集vm-support捆绑包的方法。 | |
iofilterd-spm | /var/log/iofiltervpd.log | I/O筛选器(spm) VMware Storage I/O Control | |
iofilterd-vmwarevmcrypt | /var/log/iofiltervpd.log | I/O筛选器( vmwarevmcrypt) VMcrypt IO Filter | |
iofiltervpd | /var/log/iofiltervpd.log | I/O筛选器服务 | |
lacp | /var/log/lacp.log | 分布式交换机链路聚合组 | |
lbtd | 基于负载的绑定守护进程 | ||
lwsmd | Active Directory服务 | ||
memscrubd | 未查到 | ||
nfcd | /var/log/nfcd.log | TCP/IP堆栈日志 | |
nfsgssd | NFS服务 | ||
nicmgmtd | 未查到 | ||
nscd | |||
ntpd | NTP守护进程 | ||
osfsd | /var/log/osfd.log | 对象存储文件系统守护程序负责将对象存储在本地文件系统上，并通过网络为vSAN提供对它们的访问 | |
pcscd | PC/SC 智能卡守护进程 | ||
rabbitmqproxy | /var/log/rabbitmqproxy.log | 在 Esxi 主机上运行的代理。此代理允许在虚拟机内运行的应用程序与在 vCenter 网络域中运行的 AMQP 代理进行通信。虚拟机不必在vCenter网络上，也就是说，不需要 NIC。确保传出连接 IP 地址至少包括正在使用或将来使用的代理。 | |
rhttpproxy | /var/log/rhttpproxy.log | Rhttpproxy: | 在其他ESXi主机上的HTML5反向代理服务 |
sdrsInjector | /var/log/sdrslnjector.log | vSphere Storage DRS设备注入程序日志 | |
sensord | 传感器服务 | ||
sfcbd-watchdog | CIM 服务 | ||
slpd | 未查到 | ||
smartd | 未查到 | ||
snmpd | SNMP 服务 | ||
storageRM | /var/log/storagerm.log | storageRM | SIOC日志，串行输入输出控制器 |
swapobjd | /var/log/swapobjd.log | 交换VVol对象守护程序 | |
upitd | /var/log/upitd.log | upit守护进程日志，主要用于快照操作期间的虚拟磁盘管理 | |
usbarbitrator | 未查到 | ||
vit_loader | vSAN iSCSI负载器 | ||
vitd | /var/log/vitd.log | vSAN iSCSI目标守护程序。保留vSAN iSCSI目标服务的事件。 | |
vmfstraced | 未查到 | ||
vmsyslogd | /var/log/syslog | Syslog 服务 | |
vmtoolsd | vmtools管理服务 | ||
vmware-fdm | /var/log/fdm.log | Fdm: | vSphere High Availability Agent |
vobd | /var/log/vobd.log | VMkernel观察事件 | |
vpxa | /var/log/vpxa.log | Vpxa: | VC代理（ESXi端）负责收集VC信息，将信息传递给esxi内核。 vpxd是VC代理服务（VC端） |
vsanObserver | Ruby vSphere Console（RVC）实验性图形用户界面实用程序，可从VSAN客户端角度显示与VSAN相 | ||
vsandevicemonitord | 未查到 | ||
vsandpd | /var/log/vsanvpd.log | vSAN存储提供程序守护程序 | |
vsanmgmtd | /var/log/vsanmgmt.log | VSANMGMTSVC | vSAN管理服务 |
vsantraced | /var/log/vsantraces/* | ||
vvold | /var/log/vvold.log | 虚拟机卷日志 | |
wsman | WS-Management 用于Powershell和管理的Web管理服务 | ||
xorg | /var/log/Xorg.log | Xorg服务生成的日志。Xorg服务处理VMware虚拟图形（使用GPU虚拟化VSGA模式下才启动这个服务） |
VSAN日志类型如下：
VSANMGMTSVC: vsan管理服务
vsantraceUrgent: 集群或对象/组件发生的问题相关的冗长日志记录服务，组件的状态从“活动”更改为“不存在”：（“'oldCompState：'active，'newCompState'：'absent'”例如：旧组件状态为'active'组件现在标记为'absent'，这种类型的事件可能与集群中的其他问题（例如主机）重合何时/是否看到相同UUID的相关日志记录，可以确定是否失去与集群其余部分的连接，控制器重置，磁盘/磁盘组发生故障，此事件/状态是否是暂时的，例如： oldCompState：“不存在”，“ newCompState”：“活动”）。
vsansystem:
关键字段类型如下：
VSANMGMTSVC:
cmmdsTimeMachineDump:
vsantraceUrgent:
vsansystem:
Rhttpproxy:
Vpxa:
esxupdate:
hostd-probe:
epd:
crond[
Hostd:
Fdm:
watchdog-cdp:
cmmdsd
切割日志采用的通用字段：
主机IP：host
时间：timestamp 或 timestamp8601
日志幻数：log_pri
主机名：hostname
服务名称：service
日志级别：level
进程ID：pid
消息：message
满足以下条件的日志类型，自定义为VPXA_SYSLOG_FORMAT（常见型），定义模板为VPXA_SYSLOG_FORMAT6和VPXA_SYSLOG_FORMAT5符合的日志有：
%{IP} %{pri} %{timestamp}%{hostname}%{program}%{pid}%{level}%{message}
示例：
1、Vpxa
2、Hostd
3、Rhttpproxy
4、hostd-probe
5、vsansystem
6、VSANMGMTSVC
7、Fdm
满足以下条件的日志类型，自定义为ESXUPDATE（特殊型），定义模板为ESXUPDATE符合的日志有：
示例esxupdate 
1、esxupdate
2、
满足以下条件的日志类型，自定义为COMMON（普通型），定义模板为COMMON符合的日志有：
示例：watchdog-cdp 
1、hostd
2、watchdog-cdp
3、edp
4、
满足以下条件的日志类型，自定义为CRON_STORAGERM（特殊型），定义模板为CRON_STORAGERM符合的日志有：
示例：crond 
1、cron
2、storageRM
3、cmmdsd
无法匹配的：
1、Hostd 有部分不能匹配，原因未知，让其匹配COMMON
2、vsantraceUrgent
3、cmmdsTimeMachineDump 日志时间格式不一致，此日志为记录备份了多少组件，意义不大可以忽略
ESXi 6及以后版本，日志告警级别有：
0、none ？turn off？
1、emergency
2、alert
3、error
4、warning
5、notice
6、info
7、verbose
ESXi 5系列版本，日志告警级别定义：
1、emergency
2、alert
3、error
4、warning
5、notice?
6、verbose
vSAN 日志告警级别定义：
1、None？
2、
3、Error
4、Warning
5、Verbose
6、INFO
vCenter 日志告警级别官方KB查询的定义：
1、None
2、Error
3、Warning
4、Info
5、Verbose
6、Trivia（debug）
但自己从VC配置中查看到的定义：
1、emergency
2、alert
3、crit
4、error
5、warn
6、notice
7、info
8、debug
NSX 日志告警级别定义：
1、OFF
2、FATAL
3、ERROR
4、WARN
5、INFO
6、DEBUG
7、TRACE
日志字段标准化面临的难点：
1、VMware产品为商业产品，很多文档属于内部机密，无法获取。
2、很多字段都是字母缩写，无法查找到相关资料。
3、不同版本的ESXi日志级别定义不一致。
4、相同主版本，不同小版本或不同硬件厂商的ESXi服务器，同一日志，可能具有不同日志字段。
5、超1KB单条日志，被syslog切割为多条日志发送出来；在大规模主机环境下，需要先通过kafka进行缓存消息，后续logstash消费kafka消息，会出现乱序的情况，严重影可读性。需要进行日志消息聚合后，再进行ES存储。
6、正则表达式规则太多，严重影响性能，需要对此进行持续性优化。
通过日志预警故障难点：
1、应用场景模型不明确，需要按照运维人员的经验去摸索和建立模型。
2、日志为事后发生记录，并不适合预警；但是对事后审计，故障定位非常有用。
3、预警场景特点，只有当某一故障由“多因子”引起或者具有“严重程度具有逐级加深”特点时且具有一定时间差才具有预警的实现可行性和实现意义。即：当一个严重故障产是由另一个或多个影响级别较低故障引起时，才具有预警可能性和实现意义。
七、logstash自定义模板
（一）模板使用方式：
（1）使用默认自带的索引模板 ，大部分的字段都会分词，适合开发和时候快速验证使用
（2）在logstash收集端自定义配置模板，因为分散在收集机器上，维护比较麻烦
（3）在elasticsearc服务端自定义配置模板，由elasticsearch负责加载模板，可动态更改，全局生效，维护比较容易
以上几种方式：
使用第一种，最简单，无须任何配置
使用第二种，适合小规模集群的日志收集，需要在logstash的output插件中使用template指定本机器上的一个模板json路径，如 template => "/tmp/logstash.json"
使用第三种，适合大规模集群的日志收集，主要配置logstash的output插件中两个参数：
1.manage_template => false//关闭logstash自动管理模板功能
2. template_name => "crawl"//映射模板的名字
如果使用了，第三种需要在elasticsearch的集群中的config/templates路径下配置模板json
（二）模板类型：静态模板和动态模板
静态模板：适合索引字段数据固定的场景，一旦配置完成，不能向里面加入多余的字段，否则会报错
优点：scheama已知，业务场景明确，不容易出现因字段随便映射从而造成元数据撑爆es内存，从而导致es集群全部宕机
缺点：字段数多的情况下配置稍繁琐
示例：
1.{
2. "crawl" : {
3. "template": "crawl-*",
4. "settings": {
5. "index.number_of_shards": 3,
6. "number_of_replicas": 0
7. },
8. "mappings" : {
9. "logs" : {
10. "properties" : {
11. "@timestamp" : {
12. "type" : "date",
13. "format" : "dateOptionalTime",
14. "doc_values" : true
15. },
16. "@version" : {
17. "type" : "string",
18. "index" : "not_analyzed",
19. "doc_values" : true
20. },
21. "cid" : {
22. "type" : "string",
23. "index" : "not_analyzed"
24. },
25. "crow" : {
26. "type" : "string",
27. "index" : "not_analyzed"
28. },
29. "erow" : {
30. "type" : "string",
31. "index" : "not_analyzed"
32. },
33. "host" : {
34. "type" : "string",
35. "index" : "not_analyzed"
36. },
37. "httpcode" : {
38. "type" : "string",
39. "index" : "not_analyzed"
40. },
41. "message" : {
42. "type" : "string"
43. },
44. "path" : {
45. "type" : "string"
46. },
47. "pcode" : {
48. "type" : "string",
49. "index" : "not_analyzed"
50. },
51. "pro" : {
52. "type" : "string",
53. "index" : "not_analyzed"
54. },
55. "ptype" : {
56. "type" : "string",
57. "index" : "not_analyzed"
58. },
59. "save" : {
60. "type" : "string",
61. "index" : "not_analyzed"
62. },
63. "t1" : {
64. "type" : "string",
65. "index" : "not_analyzed"
66. },
67. "t2" : {
68. "type" : "string",
69. "index" : "not_analyzed"
70. },
71. "t3" : {
72. "type" : "string",
73. "index" : "not_analyzed"
74. },
75. "url" : {
76. "type" : "string"
77. }
78. }
79. }
80. }
81. }
82.}
动态模板：适合字段数不明确，大量字段的配置类型相同的场景，多加字段不会报错
优点：可动态添加任意字段，无须改动scheaml，
缺点：如果添加的字段非常多，有可能造成es集群宕机
如下的一个logstash的动态索引模板，只设置message字段分词，其他的字段默认都不分词 ：
1.{
2. "template" : "crawl-*",
3. "settings" : {
4. "index.number_of_shards": 5,
5. "number_of_replicas": 0
6.
7.},
8. "mappings" : {
9. "_default_" : {
10. "_all" : {"enabled" : true, "omit_norms" : true},
11. "dynamic_templates" : [ {
12. "message_field" : {
13. "match" : "message",
14. "match_mapping_type" : "string",
15. "mapping" : {
16. "type" : "string", "index" : "analyzed", "omit_norms" : true,
17. "fielddata" : { "format" : "disabled" }
18. }
19. }
20. }, {
21. "string_fields" : {
22. "match" : "*",
23. "match_mapping_type" : "string",
24. "mapping" : {
25. "type" : "string", "index" : "not_analyzed", "doc_values" : true
26. }
27. }
28. } ],
29. "properties" : {
30. "@timestamp": { "type": "date" },
31. "@version": { "type": "string", "index": "not_analyzed" },
32. "geoip" : {
33. "dynamic": true,
34. "properties" : {
35. "ip": { "type": "ip" },
36. "location" : { "type" : "geo_point" },
37. "latitude" : { "type" : "float" },
38. "longitude" : { "type" : "float" }
39. }
40. }
41. }
42. }
43. }
44.}
（三）总结：
定制索引模板，是搜索业务中一项比较重要的步骤，需要注意的地方有很多，比如：
（1）字段数固定吗
（2）字段类型是什么
（3）分不分词
（4）索引不索引
（5）存储不存储
（6）排不排序
（7）是否加权
除了这些还有其他的一些因素，比如，词库的维护改动，搜索架构的变化等等。
如果前提没有充分的规划好，后期改变的话，改动其中任何一项，都需要重建索引，这个代价是非常大和耗时的，尤其是在一些数据量大的场景中。
提高正则查询效率，参见博文https://www.elastic.co/blog/do-you-grok-grok：
（1）、避免失败匹配，匹配失败会浪费大量资源和时间，注意正则表达式的锚定，使用^$匹配行起始和终止，可以提升匹配失败速度，性能提升7倍左右。匹配失败的耗时大多发生在每行起始和结尾处。
（2）、如果使用多行匹配且不使用锚定时，可以使用多级匹配方式，可以提升性能3-4倍。如果使用了锚定，再使用多级匹配性能反而会有所下降。
八、创建索引规则规划
一、按照时间（天）进行索引自动创建，最初接入方式
二、按照服务类型（即日志类型）区分建立索引
1、在日志传入时，通过logstash按照不同日志类型，过滤和标准化日志字段，再存入kafka中。（kafka）
2、在存入ES之前，通过logstash对日志的存储采用动态模板方式，按照“日期+日志类型”建立索引，存入ES。
九、对接zabbix监控系统
对接zabbix监控系统注意事项（坑）：
1、因分布式部署logstash-kafka-logstash，每存入一次logstash，都会自动增加一个当前时间戳。前期将最前端的logstash配置文件中，添加了删除@timestamp选项。
2、日志从最前端的logstash接收到存入水土ES集群，期间链路太长，经过了logstash-kafka-logstash-kafka-logstash-es。如果再通过ES中查询日志后告警方式，链路更长logstash-kafka-logstash-kafka-logstash-es-logstash-zabbix。传输中间环节出现拥堵后，告警延时非常大，失去日志告警和预警的意义。
3、从ES定时查询日志告警推送到zabbix系统时，因ES按照每天进行索引日志，每条告警会出现多次，每查询一次都会出现一次。zabbix无法做到对已告警日志进行判断，例如：一台ESXi主机在8:00发生了重启，以后的每一分钟每一次查询，结果中都会以后这条记录，而zabbix当前版本无法做到对易发生的告警，进行忽略。
解决方案：
1、将告警推送点迁移至每个机房的最前端logstash（即接收syslog端），同时也将所有日志按原来路径进行存储一份到ES，供后续查询。解决了日志告警滞后问题。
2、将告警日志推送到每个机房对应的zabbix代理服务器，分布式推送，降低了统一集中查询时的负载压力。同时缩小了故障影响范围，各机房互不影响。
此方案缺点：
1、增加了前置logstash脚本编写工作量和后期的维护管理成本。
2、增加了一个时间字段，降低了正则匹配效率，处理延时从原来的1.1ms增加到了16.1ms。ES的前置logstash压力增大。还需解决。
3、需修改ES前置的logstash脚本的grok匹配规则。
具体实现：
一、在zabbix系统中创建ESXi监控主机对象，按照ESXi的主机IP地址进行创建。使用python脚本实现（主机列表后期维护需要完善，例如：主机新增、删除、变动）。
二、在前置logstash上面编写脚本，实现告警日志过滤。
示例：
input {
tcp {
port => 514
type => syslog
}
}
filter {
#ESXi主机正在关机、正在重启、已经重启
if "Host has booted" in [message] or "Host is powering off" in [message] or "Host is rebooting" in [message] and "bootstop" in [message] {
mutate {
add_field => [ "[zkey]","bootstop" ]
}
}
#ESXi主机紫屏，发现了coredump文件
else if "file core dump found" in [message] {
mutate {
add_field => [ "[zkey]","bootstop" ]
}
}
#网卡状态改变
else if "linkstate" in [message] and "vmnic" in [message] and "shell" not in [message] {
mutate {
add_field => [ "[zkey]","vobd" ]
}
}
#hostd服务挂掉
else if "hostd detected to be non-responsive" in [message] {
mutate {
add_field => [ "[zkey]","hostd" ]
}
}
#ESXi主机从VC脱管
else if "is disconnected from vCenter Server" in [message] {
mutate {
add_field => [ "[zkey]","vpxd" ]
}
}
#SSD出现拥堵
else if "Virtual SAN node" in [message] and "congestion reached" in [message] {
mutate {
add_field => [ "[zkey]","vmkernel" ]
}
}
#HBA卡出现死循环
else if "LOOP DEAD detected" in [message] {
mutate {
add_field => [ "[zkey]","vmkernel" ]
}
}
#LUN分区表丢失
else if "Could not open device" in [message] and "No such partition on target" in [message] {
mutate {
add_field => [ "[zkey]","vmkernel" ]
}
}
#iSCSI预留冲突
else if "I/O failed due to too many reservation conflicts" in [message] {
mutate {
add_field => [ "[zkey]","vmkernel" ]
}
}
#VMFS心跳超时，VSAN的某些文件夹失去访问权限
else if "esx.problem.vmfs.heartbeat.timedout" in [message] {
mutate {
add_field => [ "[zkey]","vobd" ]
}
}
#VSAN磁盘出现永久性错误
else if "vob.vsan.lsom.diskerror" in [message] {
mutate {
add_field => [ "[zkey]","vobd" ]
}
}
#硬盘故障灯亮起
else if "LED for disk" in [message] and "is lit up" in [message] {
mutate {
add_field => [ "[zkey]","clomd" ]
}
}
#虚拟机因HA保护将重启
else if "Virtual machine" in [message] and "is vSphere HA Protected and HA will attempt to restart it after a failure" in [message] {
mutate {
add_field => [ "[zkey]","clomd" ]
}
}
#虚拟机磁盘无法写入
else if "NVRAM: write failed" in [message] {
mutate {
add_field => [ "[zkey]","vpxd" ]
}
}
#ESXi存储卷访问权限丢失
else if "access to volume" in [message] and "DataStore" in [message] {
mutate {
add_field => [ "[zkey]","vpxd" ]
}
}
#HHD硬盘预测性故障，每90秒会报一次，需修改zabbix告警周期
else if "Predictive Failure state assert for" in [message] {
mutate {
add_field => [ "[zkey]","hostd" ]
}
}
}
output {
kafka {
bootstrap_servers => "17.29.53.111:9092,17.29.53.112:9092,17.29.53.113:9092"
topic_id => "esxi-log"
}
if [zkey] == "bootstop"{
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
else if [zkey] == "vobd" {
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
else if [zkey] == "hostd" {
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
else if [zkey] == "vpxd" {
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
else if [zkey] == "vmkernel" {
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
else if [zkey] == "clomd" {
zabbix {
zabbix_host => "host"
zabbix_key => "[zkey]"
zabbix_value => "message"
zabbix_server_host => "17.29.53.240"
zabbix_server_port => "10051"
}
}
}
十、参考信息汇总
官方中文网站：https://www.elastic.co/cn/
演示：https://demo.elastic.co
产品文档：https://elastic.co/guide/index.html
性能测试工具：https://esrally.readthedocs.io/en/stable/
ESS：https://cloud.elastic.co/
系统支持矩阵：https://www.elastic.co/support/matrix
EOL：https://www.elastic.co/support/eol
中文社区：https://elasticsearch.cn/
最后附上常用命令：
1、查看消息堆积数量：
./kafka-consumer-groups.sh --bootstrap-server 12.168.13.224:9092 --describe --group esxi-consumer-group
2、创建topic
./kafka-topics.sh –zookeeper 12.168.13.201:2181,12.168.13.202:2181,12.168.13.203:2181 --create --topic esxi --replication-factor 3 --partitions 3
3、查看分区和leader
./kafka-topics.sh --describe --zookeeper 12.168.13.201:2181 --topic esxi-log
4、手动增加分区数和副本数，先创建一个test.json的配置文件，在执行如下命令。
./kafka-reassign-partitions.sh --zookeeper 12.168.13.224:2181 --reassignment-json-file ../config/test.json –execute
配置文件内容如下：
[root@kafka1 config]# cat test.json
{
"version": 1,
"partitions": [
{
"topic": "esxi-log",
"partition": 0,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 1,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 2,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 3,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 4,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 5,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 6,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 7,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 8,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 9,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 10,
"replicas": [
1,
2,
3,
4
]
},
{
"topic": "esxi-log",
"partition": 11,
"replicas": [
1,
2,
3,
4
]
}
]
}
5、查看分区情况：
./kafka-topics.sh --describe --zookeeper 12.168.13.201:2181 --topic esxi-log
Kafka命令集合：
1、定义消费组：kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=cg_name_1 --topic kafka_test_topic --from-beginnin
2、查看获取数据：kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka_test_topic --consumer-property group.id=cg_name_3 --from-beginning
3、查看消费组：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
4、查看元数据消费情况：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group cg_name_1 --describe --topic test
5、重置当前偏移量，并未实际生效，仅仅显示而已，实际执行需要加--execute：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group cg_name_1 --reset-offsets --to-earliest --topic kafka_test_topic
6、将当前偏移量设置为最后开始：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group cg_name_1 --reset-offsets --to-latest --topic kafka_test_topic --execute
7、删除消费组：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group cg_name_1
8、删除多个消费组：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group cg_name_1 --group cg_name_3
9、没有客户端处于活动状态：kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group-test --members
Output:Consumer group 'my-group-test' has no active members.10、修改分区数：bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic--partitions < 新分区数 >
11、修改主题级别参数：bin/kafka-configs.sh --zookeeper zookeeper_host:port --entity-type topics --entity-name--alter --add-config max.message.bytes=10485760
12、变更副本数：bin/kafka-configs.sh --zookeeper zookeeper_host:port --alter --add-config 'leader.replication.throttled.rate=104857600,follower.replication.throttled.rate=104857600' --entity-type brokers --entity-name 0
13、均衡leader或者增加分区数：
1）创建一个.json文件：
{"version":1, "partitions":[ |
{"topic":"__consumer_offsets","partition":0,"replicas":[0,1,2]}, |
{"topic":"__consumer_offsets","partition":1,"replicas":[0,2,1]}, |
{"topic":"__consumer_offsets","partition":2,"replicas":[1,0,2]}, |
{"topic":"__consumer_offsets","partition":3,"replicas":[1,2,0]}, |
... |
{"topic":"__consumer_offsets","partition":49,"replicas":[0,1,2]} |
]}` |
2）执行分区脚本：bin/kafka-reassign-partitions.sh --zookeeper 12.168.13.224:2181,12.168.13.201:2181,12.168.13.202:2181,12.168.13.203:2181 --reassignment-json-file test.json --execute14、按首选方式调整leader：bin/kafka-leader-election.sh --all-topic-partitions --bootstrap-server 12.168.13.201:9092,12.168.13.202:9092,12.168.13.203:9092,12.168.13.224:9092 --election-type PREFERRED
15、修改ES集群每个node默认分片最大数量：
查看GET /_cluster/settings?pretty永久修改：
PUT /_cluster/settings{ "persistent": { "cluster": { "max_shards_per_node": "10000" } }}临时修改：
PUT /_cluster/settings{" transient": {
"cluster": { "max_shards_per_node": "10000" } }}
性能优化点：
1、logstash的filter中grok正则匹配，语句修改，减少字段等。
2、logstash的output中条件判断输出语句优化。
3、增加logstash服务器的计算资源。
4、做好kafka的消费者集群logstash负载均衡。
5、调整kafka的分区数、leader负载分布。
