文章目录
- EFLK（ELK+filebeat）
- 部署filebeat
- 修改配置文件
- logstash配置
- logstash的filter过滤
- grok(正则捕获插件)
- 内置正则表达式调用
- 自定义表达式
- mutate(数据修改插件)
- 重命名字段
- 添加字段
- 删除字段
- 转换数据类型
- 替换字段内容
- 以"|"为分割符拆分数据成为数组
- 合并字段
- 用新值替换字段的值
- 添加字段first，值为message数组的第一个元素的值
- 有条件的添加标签
- multiline(多行合并插件)
- 安装multiline插件
- date（时间插件）
- 案例
EFLK（ELK+filebeat）
elsaticsearch:
node1：192.168.242.66
node2：192.168.242.67nginx+logstash+kibana：
192.168.242.68filebeat：192.168.242.69
部署filebeat
systemctl stop firewalld
systemctl disable firewalld
setenforce 0vim /etc/selinux/configSELINUX=disabled
##安装 Filebeat#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录tar xf filebeat-6.7.2-linux-x86_64.tar.gzmv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat
##安装 apache 服务，用来提供访问日志yum -y install httpdsystemctl start httpd
systemctl enable httpd
修改配置文件
###设置 filebeat 的主配置文件cd /usr/local/filebeatcp filebeat.yml{,.bak}
vim filebeat.ymlfilebeat.inputs:- type: log
#指定 log 类型，从日志文件中读取消息enabled: truepaths:- /var/log/httpd/access_log #指定监控的日志文件tags: ["filebeat"] #设置索引标签#可以使用 fields 配置选项设置一些参数字段添加到 output fields: server_name: httpdlog_type: accessfrom: 192.168.242.69--------------Elasticsearch output-------------------
(全部注释掉)----------------Logstash output---------------------
output.logstash:hosts: ["192.168.242.68:5044"] #指定 logstash 的 IP 和端口
###启动 filebeatcd /usr/local/filebeatnohup
./filebeat -e -c filebeat.yml #-e：输出到标准输出，禁用syslog/文件输出
#-c：指定配置文件
#nohup：在系统后台不挂断地运行命令，退出终端不会影响程序的运行
logstash配置
###在 Logstash 组件所在节点上新建一个 Logstash 配置文件cd /etc/logstash/conf.dvim filebeat.conf
input {beats {port => "5044"}
}output {elasticsearch {hosts => ["192.168.242.66:9200","192.168.242.67:9200"]index => "%{[fields][server_name]}-%{+YYYY.MM.dd}"}}
###启动 logstashlogstash -f filebeat.conf###如果配置失败，清空缓存文件rm -rf /var/lib/logstash/.lock
###浏览器访问
--》http://192.168.242.68:5601 登录 Kibana
--》单击“Create Index Pattern”按钮添加索引
--》“filebeat-*”
--》单击 “create” 按钮创建
--》单击 “Discover” 按钮可查看图表信息及日志信息。
logstash的filter过滤
grok(正则捕获插件)
- grok 使用文本片段切分的方式来切分日志事件
- logstash 官方也给了一些常用的常量来表达那些正则表达式，可以到这个 Github 地址查看有哪些常用的常量：
内置的grok正则表达式
内置正则表达式调用
%{SYNTAX:SEMANTIC}##SYNTAX代表匹配值的类型，例如，0.11可以NUMBER类型所匹配，10.222.22.25可以使用IP匹配。##SEMANTIC表示存储该值的一个变量声明，它会存储在elasticsearch当中方便kibana做字段搜索和统计，你可以将一个IP定义为客户端IP地址client_ip_address，如%{IP:client_ip_address}，所匹配到的值就会存储到client_ip_address这个字段里边，类似数据库的列名，也可以把 event log 中的数字当成数字类型存储在一个指定的变量当中，比如响应时间http_response_time，假设event log record如下:message: 192.168.80.10 GET /index.html 15824 0.043可以使用如下grok pattern来匹配这种记录
%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}
###在logstash.conf 配置文件中使用filter过滤规则filter {grok {match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }}
}
192.168.80.1 - - [11/Jul/2023:09:48:24 +0800] "GET / HTTP/1.1" 403 4897 "-" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.5359.125 Safari/537.36"192.168.80.1 - - [11/Jul/2023:10:24:52 +0800] "GET /test.html HTTP/1.1" 200 32 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36"192.168.80.20 - - [11/Jul/2023:10:24:36 +0800] "GET /test.html HTTP/1.1" 200 32 "-" "curl/7.29.0"###通过grok过滤规则过滤上面的三段数据%{IP:remote_addr} - - \[%{HTTPDATE:log_time}\] \"%{WORD:http_method} %{URIPATH:request_url} (?<http_var>.+)\" %{NUMBER:response_code} .+ \".+\" \"(?<user_agent>.+)\"
###在配置文件中添加filtervim /etc/logstash/conf.d/filebeat.conf##添加filter {grok {match => { "message" => "%{IP:remote_addr} - - \[%{HTTPDATE:log_time}\] \"%{WORD:http_method} %{URIPATH:request_url} (?<http_var>.+)\" %{NUMBER:response_code} .+ \".+\" \"(?<user_agent>.+)\"" }}
}
自定义表达式
语法：(?<field_name>pattern)举例：捕获10或11和长度的十六进制数的queue_id可以使用表达式(?<queue_id>[0-9A-F]{10,11})
message: 192.168.80.10 GET /index.html 15824 0.043(?<remote_addr>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)
如果表达式匹配失败，会生成一个tags字段，字段值为 _grokparsefailure，需要重新检查上边的match配置解析是否正确。
ipv4 的正则表达式0-9 [0-9]
10-99 [1-9][0-9]
100-199 1[0-9][0-9]
200-249 2[0-4][0-9]
250-255 25[0-5]
mutate(数据修改插件)
- 它提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段
###Mutate 过滤器常用的配置选项add_field 向事件添加新字段，也可以添加多个字段
remove_field 从事件中删除任意字段
add_tag 向事件添加任意标签，在tag字段中添加一段自定义的内容，当tag字段中超过一个内容的时候会变成数组
remove_tag 从事件中删除标签（如果存在）
convert 将字段值转换为另一种数据类型
id 向现场事件添加唯一的ID
lowercase 将字符串字段转换为其小写形式
replace 用新值替换字段
strip 删除开头和结尾的空格
uppercase 将字符串字段转换为等效的大写字母
update 用新值更新现有字段
rename 重命名事件中的字段
gsub 通过正则表达式替换字段中匹配到的值
merge 合并数组或 hash 事件
split 通过指定的分隔符分割字段中的字符串为数组重命名字段
filter {mutate {#写法1，使用中括号括起来rename => ["old_field" => "new_field"]#写法2，使用大括号{}括起来rename => { "old_field" => "new_field" } }
}
添加字段
filter {mutate {add_field => {"f1" => "field1""f2" => "field2"}}
}
删除字段
将字段删除
filter {mutate {remove_field => ["message", "@version", "tags"]}
}转换数据类型
###将filedName1字段数据类型转换成string类型，filedName2字段数据类型转换成float类型filter {mutate {#写法1，使用中括号括起来convert => ["filedName1", "string"]#写法2，使用大括号{}括起来convert => { "filedName2" => "float" }}
}替换字段内容
filter {mutate {gsub => ["filedName", "/" , "_"]}
}
###将filedName字段中所有",“字符后面添加空格filter {mutate {gsub => ["filedName", "," , ", "]}
}
以"|"为分割符拆分数据成为数组
filter {mutate {split => ["filedName", "|"]}
}
合并字段
filter {merge { "filedName2" => "filedName1" }
}
用新值替换字段的值
filter {mutate {replace => { "filedName" => "new_value" }}
}
添加字段first，值为message数组的第一个元素的值
filter {mutate {split => ["message", "|"]add_field => {"first" => "%{[message][0]}" } }
}
有条件的添加标签
filter {#在日志文件路径包含 access 的条件下添加标签内容if [path] =~ "access" {mutate {add_tag => ["Nginx Access Log"]}}#在日志文件路径是 /var/log/nginx/error.log 的条件下添加标签内容if [path] == "/var/log/nginx/error.log" {mutate {add_tag => ["Nginx Error Log"]}}
}
multiline(多行合并插件)
- java错误日志一般都是一条日志很多行的，会把堆栈信息打印出来，当经过 logstash 解析后，每一行都会当做一条记录存放到 ES， 那这种情况肯定是需要处理的。
- 这里就需要使用 multiline 插件，对属于同一个条日志的记录进行拼接。
安装multiline插件
###在线安装插件cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline
###离线安装插件先在有网的机器上在线安装插件，然后打包，拷贝到服务器，执行安装命令bin/logstash-plugin prepare-offline-pack --overwrite --output logstash-filter-multiline.zip logstash-filter-multiline###上传压缩包文件 logstash-offline-plugins-6.7.2.zip ./bin/logstash-plugin install file:///usr/share/logstash/logstash-offline-plugins-6.7.2.zip###检查下插件是否安装成功，可以执行以下命令查看插件列表bin/logstash-plugin list
- 使用 multiline 插件
- 第一步：每一条日志的第一行开头都是一个时间，可以用时间的正则表达式匹配到第一行。
- 第二步：然后将后面每一行的日志与第一行合并。
- 第三步：当遇到某一行的开头是可以匹配正则表达式的时间的，就停止第一条日志的合并，开始合并第二条日志。
- 第四步：重复第二步和第三步。
###合并下面的java日志vim /var/log/java.log2022-11-11 17:09:19.774[XNIo-1 task-1]ERROR com.passjava.controlle .NembercController-查询用户 活动数据失败,异常信息为:com.passjava.exception.MemberException: 当前没有配置活动规则at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)at com.passjava.service.impl.dailyserviceImpL$$FastcLass
2022-11-11 17:10:56.256][KxNIo-1 task-1] ERROR com.passjava.controlle .NemberControl1er-查询员工 饭活动数据失败,异常信息为:com.passjava.exception.MemberException: 当前没有配置活动规则at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)at com.passjava.service.impL.daiLyserviceImpL$$FastcLass##过滤规则
##创建配置文件vim /etc/logstash/conf.d/java.confinput {file {path => "/var/log/java.log"type => "java"start_position => "beginning"}}filter {multiline {pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}"negate => truewhat => "previous"}
}output {elasticsearch {hosts => ["192.168.242.66:9200","192.168.242.67:9200"]index => "java-%{+YYYY.MM.dd}"}}
pattern：用来匹配文本的表达式，也可以是grok表达式what：如果pattern匹配成功的话，那么匹配行是归属于上一个事件，还是归属于下一个事件。previous: 归属于上一个事件，向上合并。ext: 归属于下一个事件，向下合并negate：是否对pattern的结果取反。false：不取反，是默认值。true：取反。将多行事件扫描过程中的行匹配逻辑取反（如果pattern匹配失败，则认为当前行是多行事件的组成部分）
date（时间插件）
- 在Logstash产生了一个Event对象的时候，会给该Event设置一个时间，字段为“@timestamp”，同时，我们的日志内容一般也会有时间，但是这两个时间是不一样的，因为日志内容的时间是该日志打印出来的时间，而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间，所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点，因为Logstash监控数据变化，数据输入，创建Event导致的时间延迟。这两个时间都可以使用，具体要根据自己的需求来定。
filter {date {match => ["access_time", "dd/MMM/YYYY:HH:mm:ss Z", "UNIX", "yyyy-MM-dd HH:mm:ss", "dd-MMM-yyyy HH:mm:ss"]target => "@timestamp"timezone => "Asia/Shanghai"}
}
match：用于配置具体的匹配内容规则，前半部分内容表示匹配实际日志当中的时间戳的名称，后半部分则用于匹配实际日志当中的时间戳格式，这个地方是整条配置的核心内容，如果此处规则匹配是无效的，则生成后的日志时间戳将会被input插件读取的时间替代。如果时间格式匹配失败，会生成一个tags字段，字段值为 _dateparsefailure，需要重新检查上边的match配置解析是否正确。target：将匹配的时间戳存储到给定的目标字段中。如果未提供，则默认更新事件的@timestamp字段。timezone：当需要配置的date里面没有时区信息，而且不是UTC时间，需要设置timezone参数。
###时间戳详解：年
yyyy #全年号码。 例如：2015。
yy #两位数年份。 例如：2015年的15。月
M #最小数字月份。 例如：1 for January and 12 for December.。
MM #两位数月份。 如果需要，填充零。 例如：01 for January and 12 for Decembe
MMM #缩短的月份文本。 例如： Jan for January。 注意：使用的语言取决于您的语言环境。 请 参阅区域设置以了解如何更改语言。
MMMM #全月文本，例如：January。 注意：使用的语言取决于您的语言环境。日
d #最少数字的一天。 例如：1月份的第一天1。
dd #两位数的日子，如果需要的话可以填零.例如：01 for the 1st of the month。时
H #最小数字小时。 例如:0表示午夜。
HH #两位数小时，如果需要填零。 例如：午夜00。分
m #最小的数字分钟。 例如：0。
mm #两位数分钟，如果需要填零。 例如：00。秒
s #最小数字秒。 例如：0。
ss #两位数字，如果需要填零。 例如：00。毫秒（ 秒的小数部分最大精度是毫秒（SSS）。除此之外，零附加。）
S #十分之一秒。例如：0为亚秒值012
SS #百分之一秒 例如：01为亚秒值01
SSS #千分之一秒 例如：012为亚秒值012时区偏移或身份
Z #时区偏移，结构为HHmm（Zulu/UTC的小时和分钟偏移量）。例如：-0700。
ZZ #时区偏移结构为HH:mm（小时偏移和分钟偏移之间的冒号）。 例如：-07:00。
ZZZ #时区身份。例如：America/Los_Angeles。 注意：有效的ID在列表中列出http://joda-time.sourceforge.net/timezones.html
案例
过滤message数据中，只获得时间参数vim /etc/logstash/conf.d/filebeat.confinput {beats {port => "5044"}
}filter{grok{match => {"message" => ".* -\ -\ \[%{HTTPDATE:timestamp}\]"}}date{match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]target => "@timestamp"timezone => "Asia/Shanghai"}
}output {elasticsearch {hosts => ["192.168.242.66:9200","192.168.242.67:9200"]index => "%{[fields][server_name]}-%{+YYYY.MM.dd}"}}###在web界面查看两个时间是否一致
