ElasticSerach学习(五)-基于ES+filebeat+logstash实现日志收集系统-阿里云开发者社区

ElasticSerach学习(五)-基于ES+filebeat+logstash实现日志收集系统

2021-09-05 759

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

检索分析服务 Elasticsearch 版，2核4GB开发者规格 1个月

简介： ElasticSerach学习(五)-基于ES+filebeat+logstash实现日志收集系统

记录一下搭建elk+filebeat收集日志的全过程

环境:

es 7.7.1 单节点集群

logstash 7.7.1

filebeat 7.7.1

kibana 7.7.1

搭建环境均为linux,centos7

搭建目标:

使用filebeat读取log4j日志文件发送到logstash
logstash进行过滤解析,发送到es

先直接发送到es
尝试过滤解析日志字段,拆分后发送到es

使用kibana进行查询

1. ES环境搭建

此处搭建es单节点集群

1.1 下载解压

下载地址: https://www.elastic.co/cn/downloads/past-releases/elasticsearch-7-7-1 当前最新版本为7.12 也可以下载

下载后上传到虚拟机,解压

1.2 新建用户

因为es不允许使用root用户,所以要新建用户

useradd es
passwd es
#输入密码 es

修改解压目录的所有者为es,防止权限问题

chown-R es:es #解压目录

1.3 修改虚拟机配置

修改虚拟机配置,文件数

vim /etc/security/limits.conf

在文件末尾添加一下内容

*               soft    nofile          65536*               hard    nofile          65536*               soft    nproc           4096*               hard    nproc           4096

修改虚拟机配置

vim /etc/sysctl.conf

文件末尾增加配置(如有配置,需修改)

vm.max_map_count =262144

注意:上述配置都需要使用root账号进行修改

1.4 修改es启动配置

接下来切换到es用户

su es

进入到解压目录,找到config目录,修改jvm的内存(默认为1g,需调小一点,不然虚拟机扛不住),如果虚拟机内存够,也可以不用调

cd config
vim jvm.options
#修改如下两行 (默认为1g)-Xms256m-Xmx256m

1.5 修改es配置

可以通过修改config目录下elasticsearch.yml修改es的相关配置,比如节点名称,访问ip,集群配置等.

由于我们这里是单节点集群,所以使用默认配置即可

1.6 启动es

cd bin
./elasticsearch -d

-d 使es后台运行,第一次运行可以不加-d,在控制台查看启动日志是否有异常.

1.7 关闭es

# 查看pid

nestat -nptl# 默认端口号为9200 9300对应的pidkill <pid>

2. logstash

2.1 下载解压

下载地址 https://www.elastic.co/cn/downloads/past-releases/logstash-7-7-1

解压到指定目录(使用es用户即可)

2.2 修改配置文件

2.2.1 修改jvm.options

进入config目录下,修改jvm内存

cd config
vim jvm.options
# 修改-Xms -Xmx两行-Xms256m-Xmx256m

2.2.2 修改logstash-sample.config文件

备份后修改此文件

input {
  beats {
    port => 5044
  }
}
output {
  elasticsearch {
    hosts => ["http://192.168.164.132:9200"]
    index => "yxjcjk_log-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

input 为数据来源

从beats输入,也就是filebeat访问的端口为5044

output 为数据输出

输出到es,配置es的访问地址,如果为集群,请配置多个,注意: es的访问地址是es的配置文件中配置的,默认为localhost,我这里是修改了该配置文件,换成了ip地址.

index是输入到es的index名称,可以指定年月日.

2.3 启动logstash

./bin/logstash -f ./config/logstash-sample.conf
# 使用nohup 后台启动nohup ./bin/logstah -f ./config/logstash-sample.conf > ./log/nohup.out 2>&1 &

2.4 停止logstash

# 同es 停止方法netstat -nptl# -9是发现kill杀不掉,使用-9强制停止kill-9 <pid>

3. filebeat

3.1 下载解压

下载地址 https://www.elastic.co/cn/downloads/past-releases/filebeat-7-7-1

解压到指定目录

3.2 修改配置

进入解压目录下 filebeat.reference.yml文件时filebeat的所有配置

我们备份下filebeat.yml,然后修改filebeat.yml文件

filebeat.inputs:
# Each - is an input. Most options can be set at the input level, so# you can use different inputs for various configurations.# Below are the input specific configurations.- type: log
# Change to true to enable this input configuration.  enabled: true# Paths that should be crawled and fetched. Glob based paths.  paths:
- /home/elastic/test_logs/*.log  
#不以[开头的行都合并到上一行的末尾  multiline.pattern: ^\[
  multiline.negate: true  multiline.match: after
output.logstash:
# The Logstash hosts  hosts: ["localhost:5044"]

主要配置input,output

input 从日志文件读取,需要配置文件的地址,和多行合并,

output 输出到logstash,配置logstash的端口号即可

3.2 启动filebeat

# 直接启动,控制台输出日志./filebeat -e-c filebeat.yml
# 后台启动nohup ./filebeat -e-c filebeat.yml > nohup.out 2>&1 &

-e 启动

-c 指定配置文件

3.2 停止filebeat

同logstash/es

netstat -nptlkill <pid>

4. kibana

kibana主要用于可视化

4.1 下载解压

下载地址: https://www.elastic.co/cn/downloads/past-releases/kibana-7-7-1

解压到指定地址

4.2 启动

kibana无需修改什么配置,默认直接连接localhost:9200 的es,如果es的地址有修改的话改动下此配置即可,其他配置无需修改

nohup /home/elastic/kibata/kibana-7.7.1-linux-x86_64/bin/kibana > /home/elastic/kibata/kibana-7.7.1-linux-x86_64/log/nohup.out  2>&1 &

4.3 停止

netstat -nptl# 找到5601端口对应的pidkill <pid>

5. 读取日志发送到es

根据上述配置,逐个启动 es,kibana,logstash,filebeat,就会默认读取日志文件并发送到logstash,过滤后再发送到es

但是此时发送到es的日志还是一个消息整体,就是一个字符串,我们需要对每一条日志记录进行分析,拆分为字段后写入es.

5.1 logstash对日志进行分析过滤,拆分字段

使用的是Grok插件

Grok插件将记录拆分为一个个字段

5.1.1 Grok使用

Grok的语句是%{模式名:字段名}

会把对应的数据转换为字段

常用的模式记录如下:

WORD:配置字符串

TIMESTAMP_ISO8601:配置时间

LOGLEVEL:匹配日志级别

DATA:匹配长数据

INT:匹配数字

GREEDYDATA:匹配剩余数据

使用示例:

日志记录如下:

[yxjcjk][2021-05-13 16:22:33,773][ERROR][org.gocom.components.coframe.auth.LoginManager.mobile_login_biz:540] [Name=org.gocom.components.coframe.auth.LoginManager.mobile_login.biz][activity name=存入登录用户信息日志][activity id=invokePojo2] throw an exception:java.lang.Exception,exception :

对应开发的Grok表达式为:

\[%{WORD:sys}\]\[%{TIMESTAMP_ISO8601:time}\]\[%{LOGLEVEL:level}\]\[%{DATA:uri}:%{INT:line_num}\] %{GREEDYDATA:data}

此表达式提取了系统名称(sys),时间(time),日志级别(level),访问路径(uri),路径行数(line_num),剩余具体日志(data),最终输入到es