filebeat+kafka+logstash+elasticsearch+kibena环境部署
·
服务器资源规划
| ip地址 | 软件 |
|---|---|
| 192.168.198.33 | filebeat、logstash、kafka、Master node、data node |
| 192.168.198.35 | kafka、Master node、data node |
| 192.168.198.46 | kafka data node、data node |
工作流程：
filebeat收集数据，缓冲到 -> kafka -> logstash从kafka中获取数据，通过gork或其他工具处理日志数据 -> elasticsearch -> kibena
1、安装filebeat
# 关闭selinux
setenforce 0
sed -i -e 's/^SELINUX=.*$/SELINUX=disabled/g' /etc/selinux/config
# 1、创建elkf用户
sudo useradd -r -m -d /home/elkf -s /bin/bash elkf
# 2. 将 elkf 用户加入 adm 组
sudo usermod -aG adm elkf
# 关闭防火墙
systemctl disable firewalld --now
# 版本：V9.3.0
# 下载地址：
https://www.elastic.co/downloads/beats/filebeat
# 解压到/usr/local/目录
tar -zxvf filebeat-9.3.0-linux-x86_64.tar.gz -C /usr/local/filebeat
[root@node1 filebeat]# cat /usr/local/filebeat/filebeat.yml |grep -v '^\s*#'|grep -v '^$'
filebeat.inputs:
- type: filestream
id: oslog # id标识：收集多个日志，对每一种日志可以进行标识
enabled: true # 是否开启使用手动配置，默认使用模块化配置
paths:
- /var/log/secure # 手动配置日志收集来源
fields:
log_topic: oslogs # 自定义字段，供后面输出使用；
# 例如：kafka，可以引用这个topic
# 模块化配置，不够灵活
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
# elasticsearch模板
setup.template.settings:
index.number_of_shards: 1
# 指定日志产生所在的主机名
name: 192.168.198.33
setup.kibana:
# 将数据推送到kafka消息队列中
output.kafka:
enabled: true
hosts: ["node1:9092", "node2:9092", "node3:9092"]
topic: '%{[fields.log_topic]}'
codec.format.string: '%{[message]}'
partition.round_robin:
reachable_only: true
worker: 2
required_acks: 1
compression: gzip
# 单条日志限制
max_message_bytes: 10000000
logging.level: debug
# 启动命令：
cd /usr/local/filebeat && nohup ${PWD}/filebeat -e -c filebeat.yml &
2、logstash分析日志
# 分析日志工具：V9.3.0
下载地址：https://www.elastic.co/downloads/logstash
tar -zxvf logstash-9.3.0.tar.gz -C /usr/local/
mv /usr/local/logstash-9.3.0 /usr/local/logstash
# jvm参数设置
/usr/local/logstash/config/jvm.options
[root@node1 config]# cat jvm.options |grep -v '^#'|grep -v '^$'
-Xms1g # 按照业务量进行定义
-Xmx1g # 按照业务量进行定义
-Djava.awt.headless=true
-Dfile.encoding=UTF-8
-XX:+HeapDumpOnOutOfMemoryError
-Djava.security.egd=file:/dev/urandom
# 使用Logstash事件配置文件启动Logstash
# logstash不允许root用户登录
sudo chown -R elkf:elkf /usr/local/logstash
sudo chown -R elkf:elkf /var/log/logstash
sudo chown -R elkf:elkf /var/lib/logstash
# 调试使用
sudo -uelkf ./logstash -e 'input{stdin{}} output{stdout{codec=>rubydebug}}'
2.1 事件配置文件编写
input {输入插件}
filter {过滤匹配插件} # 可选
output {输出插件}
Logstash从文件读取数据
# beginning：从日志的头开始读
input {
file {
path => ["/var/log/secure"]
type => "system"
start_position => "beginning"
}
}
output {
stdout{
codec=>rubydebug
}
}
# 启动logstash服务，加载事件配置文件；
cd /usr/local/logstash && sudo -uelkf bin/logstash -f logstash-file.conf
# 1、给 /var/log 目录设置执行权限，让用户可以进入
sudo chmod +x /var/log
# 2、修改文件所属组：将 /var/log/secure 文件的所属组改为 adm。
sudo chgrp adm /var/log/secure
# 3、修改文件权限：给文件所属组（现在是 adm 组）添加读取权限。这样，属于 adm 组的用户就能读了。
sudo chmod 640 /var/log/secure
# 4、将 elkf 用户加入 adm 组：最后，把运行 Logstash 的 elkf 用户添加到 adm 组中 。
sudo usermod -aG adm elkf
# 5、重新启动服务
cd /usr/local/logstash && sudo -uelkf bin/logstash -f logstash-file.conf
######################################################
# type和tags是logstash的两个特殊字段,
# type一般会放在input中标记事件类型,
# tags主要用于在事件中增加标签，以便在后续的处理流程中使用，
# 主要用于filter或output阶段。
# logstash-tag.conf
input{
stdin{
add_field=>{"key"=>"ok"}
tags=>["add field"] # tag标签
type=>"mytype" # type标签
}
}
output {
stdout{
codec=>rubydebug
}
}
# 启动服务
cd /usr/local/logstash && sudo -uelkf bin/logstash -f logstash-file.conf
###################################################
# 空解析器；
# 输入什么，输出就是什么内容，不做任何解析
input{
stdin {
}
}
output{
stdout {
codec => "plain"
}
}
###################################################
# json解析
input {
stdin {
}
}
output {
stdout {
codec => json # 输出内容将按照json格式
}
}
######################################################
Grok 正则捕获
filter {
grok {
match => ["message", "%{HTTPDATE:timestamp}"]
}
date {
match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
}
}
3、kafka
# 安装包下载路径：版本：kafka_2.13-4.0.0
# 需要提前安装jdk
https://kafka.apache.org/downloads
tar -zxvf kafka_2.12-3.1.0.tgz -C /usr/local
mv /usr/local/kafka_2.12-3.1.0 /usr/local/kafka
# jmx 调优 修改KAFKA_HEAP_OPTS这段
cat /usr/local/kafka/bin/kafka-server-start.sh
if [ $# -lt 1 ];
then
echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
exit 1
fi
base_dir=$(dirname $0)
if [ -z "$KAFKA_LOG4J_OPTS" ]; then
export KAFKA_LOG4J_OPTS="-Dlog4j2.configurationFile=$base_dir/../config/log4j2.yaml"
fi
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx1g -Xms1g -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80 -XX:+ExplicitGCInvokesConcurrent"
fi
EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}
COMMAND=$1
case $COMMAND in
-daemon)
EXTRA_ARGS="-daemon "$EXTRA_ARGS
shift
;;
*)
;;
esac
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
# JMX监控Agent配置
# 为了使用Prometheus监控Kafka指标，需要配置JMX收集Agent：
export KAFKA_OPTS="-javaagent:/usr/local/kafka/jmx/jmx_javaagent-1.2.0.jar=7071:/usr/local/kafka/jmx/kafka_kraft.yml"
############################################################################################
# 修改配置文件
/usr/local/kafka/config/server.properties
[root@node1 config]# cat server.properties |grep -v "^#" |grep -v "^$"
process.roles=broker,controller
node.id=1 # 每台服务器都不一样
controller.quorum.bootstrap.servers=localhost:9093
listeners=PLAINTEXT://node1:9092 # 每台服务器都不一样
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/kafka/kafka-logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
share.coordinator.state.topic.replication.factor=1
share.coordinator.state.topic.min.isr=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093
###########################################################################
# 修改文件所属组：将/usr/local/kafka的所属组改为 adm。
sudo chgrp adm /usr/local/kafka
# 获取集群id
# 这个只执行一次
/usr/local/kafka/bin/kafka-storage.sh random-uuid
5UxGj6CNRk2fBN-3wVTTHQ
# 每个服务器都要执行格式化
/usr/local/kafka/bin/kafka-storage.sh format -t U7pi7TQST_6DGGCbBAeS7A -c /usr/local/kafka/config/server.properties
# 先启动所有Controller节点（控制器之间启动顺序无关紧要）：
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
# 显示topic列表
./kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --list
# 查看某个topic的状态
./kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --describe --topic testtopic
# 创建一个topic，并指定topic属性
./kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --create --replication-factor 3 --partitions 3 --topic testtopic
# 生产消息
./kafka-console-producer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic testtopic
# 消费消息
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic oslogs
# 删除topic
./kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --delete --topic testtopic
4、elasticsearch
操作系统调优
对于操作系统，需要调整几个内核参数，将下面内容添加到/etc/sysctl.conf文件中： fs.file-max=655360 vm.max_map_count = 262144 另外，还需要调整进程最大打开文件描述符（nofile）、最大用户进程数（nproc）和最大锁定内存地址空间（memlock），添加如下内容到/etc/security/limits.conf文件中： * soft nproc 20480 * hard nproc 20480 * soft nofile 65536 * hard nofile 65536 * soft memlock unlimited * hard memlock unlimited 最后，还需要修改/etc/security/limits.d/20-nproc.conf文件（centos7.x系统），将： * soft nproc 4096 修改为： * soft nproc 20480 或者直接删除/etc/security/limits.d/20-nproc.conf文件也行。 # 使配置文件生效 sysctl -p
安装步骤
版本：9.3.0 下载地址： https://www.elastic.co/cn/downloads/elasticsearch # 使用elkf用户 chown -R elkf:elkf /usr/local/elasticsearch ########################################################## 配置文件 Jvm调优 /usr/local/elasticsearch/config/jvm.options ########################################################### /usr/local/elasticsearch/config/elasticsearch.yml [root@node2 config]# cat elasticsearch.yml|grep -v "^#"|grep -v "^$" cluster.name: esdata node.name: server2 path.data: /usr/local/elasticsearch/data path.logs: /usr/local/elasticsearch/logs bootstrap.memory_lock: true network.host: 0.0.0.0 http.port: 9200 discovery.seed_hosts: ["node1", "node2","node3"] cluster.initial_master_nodes: ["node1", "node2"] xpack.security.enabled: false http.cors.enabled: true http.cors.allow-origin: "*" [root@node2 config]# # 启动服务 sudo -u elkf /usr/local/elasticsearch/bin/elasticsearch -d
安装elasticsearch的Head插件
下载地址：
https://github.com/mobz/elasticsearch-head
yum install -y nodejs
[root@localhost elasticsearch-head]# npm install
############################################################
首先，克隆下来的head插件程序目录名为elasticsearch-head，进入此目录，修改配置文件 /usr/local/elasticsearch-head/_site/app.js ,找到如下内容：
this.base_uri = this.config.base_uri || this.prefs.get("app-base_uri") || "http://localhost:9200";
将其中的
http://localhost:9200
修改为elasticsearch集群中任意一台主机的IP地址，这里修改为:
<http://172.16.213.225:9200>
接着，修改elasticsearch配置文件，允许head插件跨域访问elasticsearch，在elasticsearch.yml文件最后添加如下内容：
http.cors.enabled: true
http.cors.allow-origin: "*"
最后，所有配置完成之后，就可以启动插件服务了，执行如下操作：
[root@localhost ~]# cd /usr/local/elasticsearch-head
[root@localhost elasticsearch-head]# npm run start
# 启动服务
npm run start
5、logstash从kafka获取日志
编写事件配置文件，读取kafka中的nginx日志
input {
kafka {
bootstrap_servers => "node1:9092,node2:9092,node3:9092"
topics => ["oslogs"]
codec => "plain"
consumer_threads => 1
group_id => "logstash-oslogs-group"
auto_offset_reset => "latest"
}
}
filter {
# 解析访问日志
grok {
match => { "message" => "%{IP:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:bytes_sent} \"%{DATA:referrer}\" \"%{DATA:user_agent}\" \"%{DATA:forwarded_for}\"" }
remove_tag => ["_grokparsefailure"] # 如果成功解析，移除失败标签
}
# 如果没有匹配上，记录原始日志以便调试
if "_grokparsefailure" in [tags] {
mutate {
add_field => { "raw_message" => "%{message}" }
}
}
# 转换字段类型
mutate {
convert => { "response_code" => "integer" }
convert => { "bytes_sent" => "integer" }
convert => { "http_version" => "float" }
}
# 将 timestamp 转换为 @timestamp
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
target => "@timestamp"
remove_field => ["timestamp"] # 移除原始时间戳字段
}
# 添加地理信息（可选）
geoip {
source => "client_ip"
target => "geo"
}
# 移除不需要的字段
mutate {
remove_field => ["message", "host", "@version", "path"]
}
}
output {
elasticsearch {
hosts => ["192.168.198.33:9200"]
index => "access-logs-%{+YYYY.MM.dd}"
# batch_size => 5000
# batch_delay => 5
}
stdout {
codec => rubydebug
}
}
# 重启logstash
cd /usr/local/logstash && sudo -uelkf nohup bin/logstash -f kafka_os_into_es.conf &
6、kibana
# 版本号：9.3.0 # 下载地址： https://www.elastic.co/cn/downloads/kibana tar -zxvf kibana-9.3.0-linux-x86_64.tar.gz -C /usr/local mv /usr/local/kibana-9.3.0-linux-x86_64 /usr/local/kibana ########################################################### # 修改配置文件 /usr/local/kibana/config/kibana.yml # 常用的配置项 server.port: 5601 server.host: "172.16.213.229" elasticsearch.hosts: ["http://172.16.213.225:9200"] 其中，每个配置项的含义介绍如下： server.port，Kibana 绑定的监听端口，默认是 5601； server.host，Kibana 绑定的 IP 地址，如果内网访问，设置为内网地址即可； elasticsearch.hosts，Kibana 访问 Elasticsearch 的地址，如果是 Elasticsearch 集群，添加任一集群节点 IP 即可，官方推荐设置为 Elasticsearch 集群中 client node 角色的节点 IP； kibana.index，用于存储 Kibana 数据信息的索引，这个可以在 Kibana Web 界面中看到。 注意，这里我配置了 Kibana 从 172.16.213.225节点上获取 Elasticsearch 数据，因此需要在此节点的 Elasticsearch 配置文件中添加如下内容： http.cors.enabled: true http.cors.allow-origin: "*" ############################################### 启动服务：cd /usr/local/kibana && nohup /usr/local/kibana/bin/kibana --allow-root & 浏览器访问：http://192.168.198.33:5601/
7、启动服务
# filebeat收集日志，将日志缓冲到kafka中
cd /usr/local/filebeat && nohup ${PWD}/filebeat -e -c filebeat.yml &
# kafka缓存日志
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
# logstash收集kafka中的日志，并做gork处理，给elasticsearch
cd /usr/local/logstash && sudo -uelkf nohup bin/logstash -f kafka_os_into_es.conf &
# elasticsearch 数据存储层
sudo -u elkf /usr/local/elasticsearch/bin/elasticsearch -d
# elasticsearch-head展示数据
cd /usr/local/elasticsearch-head && npm run start
# kibana可视化大屏，可查看日志
cd /usr/local/kibana && nohup /usr/local/kibana/bin/kibana --allow-root &
浏览器访问：http://192.168.198.33:5601/
腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。
更多推荐
12
0- 0
已为社区贡献1条内容
所有评论(0)