一、Filebeat+ELK 部署

在 Node1 节点上操作

1.安装 Filebeat
#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录tar zxvf filebeat-6.7.2-linux-x86_64.tar.gzmv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat
2. 设置 filebeat 的主配置文件
cd /usr/local/filebeatvim filebeat.ymlfilebeat.inputs:- type: log #指定 log 类型，从日志文件中读取消息enabled: truepaths:- /var/log/messages #指定监控的日志文件- /var/log/*.logtags: ["sys"]#设置索引标签fields: #可以使用 fields 配置选项设置一些参数字段添加到 output 中service_name: filebeatlog_type: syslogfrom: 192.168.80.13--------------Elasticsearch output-------------------(全部注释掉)----------------Logstash output---------------------output.logstash:hosts: ["192.168.80.12:5044"]#指定 logstash 的 IP 和端口

3. 启动 filebeat
nohup ./filebeat -e -c filebeat.yml > filebeat.out ">作用：grok 使用文本片段切分的方式来切分日志事件
 
2.1内置正则表达式调用
 
格式：%{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.d文件夹下面创建filter conf文件，内容如下# /etc/logstash/conf.d/01-filter.conffilter {grok {match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }}}以下是filter结果client_id_address: 192.168.80.10method: GETrequest: /index.htmlbytes: 15824http_response_time: 0.043
logstash 官方也给了一些常用的常量来表达那些正则表达式，可以到这个 Github 地址查看有哪些常用的常量：
 https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/grok-patterns
2.2自定义表达式调用
语法：(" /> "(?%{IP}) (?[A-Z]+) (?/.*) (?[0-9]+) (?[0-9\.]+)"}}}如果表达式匹配失败，会生成一个tags字段，字段值为 _grokparsefailure，需要重新检查上边的match配置解析是否正确。
三、multiline 多行合并插件
java错误日志一般都是一条日志很多行的，会把堆栈信息打印出来，当经过 logstash 解析后，每一行都会当做一条记录存放到 ES， 那这种情况肯定是需要处理的。 这里就需要使用 multiline 插件，对属于同一个条日志的记录进行拼接。
3.1安装 multiline 插件
在线安装插件cd /usr/share/logstashbin/logstash-plugin install logstash-filter-multiline离线安装插件先在有网的机器上在线安装插件，然后打包，拷贝到服务器，执行安装命令bin/logstash-plugin prepare-offline-pack --overwrite --output logstash-filter-multiline.zip logstash-filter-multilinebin/logstash-plugin install file:///usr/share/logstash/logstash-filter-multiline.zip检查下插件是否安装成功，可以执行以下命令查看插件列表bin/logstash-plugin list
3.2使用 multiline 插件
第一步：每一条日志的第一行开头都是一个时间，可以用时间的正则表达式匹配到第一行。
 第二步：然后将后面每一行的日志与第一行合并。
 第三步：当遇到某一行的开头是可以匹配正则表达式的时间的，就停止第一条日志的合并，开始合并第二条日志。
 第四步：重复第二步和第三步。
input {file { path => "/opt/java.log" start_position => "beginning" sincedb_path => "/etc/logstash/sincedb_path/log_progress"}}filter { multiline { pattern => "^\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{1,2}:\d{1,2}.\d{1,3}" negate => true what => "previous" }}output { elasticsearch {hosts => ["192.168.136.180:9200","192.168.136.190:9200","192.168.136.195:9200"]index => "java-error-%{+yyy.MM.dd}" }}●pattern：用来匹配文本的表达式，也可以是grok表达式●what：如果pattern匹配成功的话，那么匹配行是归属于上一个事件，还是归属于下一个事件。previous: 归属于上一个事件，向上合并。next: 归属于下一个事件，向下合并●negate：是否对pattern的结果取反。false：不取反，是默认值。true：取反。将多行事件扫描过程中的行匹配逻辑取反（如果pattern匹配失败，则认为当前行是多行事件的组成部分）
四、mutate 数据修改插件
它提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段。
4.1Mutate 过滤器常用的配置选项
选项作用
add_field向事件添加新字段，也可以添加多个字段
remove_field从事件中删除任意字段
add_tag向事件添加任意标签，在tag字段中添加一段自定义的内容，当tag字段中超过一个内容的时候会变成数组
remove_tag从事件中删除标签（如果存在）
convert将字段值转换为另一种数据类型
id向现场事件添加唯一的ID
lowercase将字符串字段转换为其小写形式
replace用新值替换字段
strip删除开头和结尾的空格
uppercase将字符串字段转换为等效的大写字母
update用新值更新现有字段
rename重命名事件中的字段
gsub通过正则表达式替换字段中匹配到的值
merge合并数组或 hash 事件
split通过指定的分隔符分割字段中的字符串为数组
4.2将字段path重命名为new_path
filter {mutate {#写法1，使用中括号括起来rename => ["path" => "new_path"]#写法2，使用大括号{}括起来rename => { "path" => "new_path" }}}
4.3添加字段
filter {mutate {add_field => {"f1" => "field1""f2" => "field2"}}}
4.4将字段删除
filter {mutate {remove_field=>["message", "@version", "tags"]}}
五、date 时间处理插件
用于分析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。
在Logstash产生了一个Event对象的时候，会给该Event设置一个时间，字段为“@timestamp”，同时，我们的日志内容一般也会有时间，但是这两个时间是不一样的，因为日志内容的时间是该日志打印出来的时间，而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间，所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点，因为Logstash监控数据变化，数据输入，创建Event导致的时间延迟。这两个时间都可以使用，具体要根据自己的需求来定。
filter { grok{ match => {"message" => ".+\[(" /> ["log_time","dd/MMM/yyyy:HH:mm:ss Z"] 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