ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）-Toy模板网

这篇具有很好参考价值的文章主要介绍了ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Filebeat

Filebeat：轻量级的开源日志文件数据搜集器。通常在需要采集数据的客户端安装 Filebeat，并指定目录与日志格式，Filebeat 就能快速收集数据，并发送给 logstash 进行解析，或是直接发给 Elasticsearch 存储，性能上相比运行于 JVM 上的 logstash 优势明显，是对它的替代。常应用于 EFLK 架构当中。

Filebeat 结合 logstash 带来好处：

1）通过 Logstash 具有基于磁盘的自适应缓冲系统，该系统将吸收传入的吞吐量，从而减轻 Elasticsearch 持续写入数据的压力
2）从其他数据源（例如数据库，S3对象存储或消息传递队列）中提取
3）将数据发送到多个目的地，例如S3，HDFS（Hadoop分布式文件系统）或写入文件
4）使用条件数据流逻辑组成更复杂的处理管道

Fluentd

Fluentd：是一个流行的开源数据收集器。由于 logstash 太重量级的缺点，Logstash 性能低、资源消耗比较多等问题，随后就有 Fluentd 的出现。相比较 logstash，Fluentd 更易用、资源消耗更少、性能更高，在数据处理上更高效可靠，受到企业欢迎，成为 logstash 的一种替代方案，常应用于 EFK 架构当中。在 Kubernetes 集群中也常使用 EFK 作为日志数据收集的方案。
在 Kubernetes 集群中一般是通过 DaemonSet 来运行 Fluentd，以便它在每个 Kubernetes 工作节点上都可以运行一个 Pod。它通过获取容器日志文件、过滤和转换日志数据，然后将数据传递到 Elasticsearch 集群，在该集群中对其进行索引和存储。

Filebeat 和 Fluentd 都可以代替 logstash 来收集日志数据

ELK + Filebeat 部署

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk
服务器结构
node1节点：192.168.60.5 部署 Elasticsearch 。
node2节点：192.168.60.8 部署 Elasticsearch 。
Nginx节点：192.168.60.12 部署 kiabana、Logstash。
Filebeat节点：192.168.60.11 部署 Filebeat、httpd

//在 Filebeat 节点上操作

1．安装 Filebeat & Httpd

#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录
tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz
mv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat

yum -y install httpd

systemctl start httpd.service 
systemctl enable httpd.service

2．设置 filebeat 的主配置文件

cd /usr/local/filebeat

vim filebeat.yml
filebeat.inputs:
- type: log         #指定 log 类型，从日志文件中读取消息
  enabled: true
  paths:
    - /var/log/messages       #指定监控的日志文件
    - /var/log/*.log
  tags: ["sys"]		#设置索引标签
  fields:           #可以使用 fields 配置选项设置一些参数字段添加到 output 中
    service_name: filebeat
    log_type: syslog
    from: 192.168.80.13

--------------Elasticsearch output-------------------
(全部注释掉)

----------------Logstash output---------------------
output.logstash:
  hosts: ["192.168.60.12:5044"]      #指定 logstash 的 IP 和端口

#启动 filebeat
nohup ./filebeat -e -c filebeat.yml > filebeat.out &
#-e：输出到标准输出，禁用syslog/文件输出
#-c：指定配置文件
#nohup：在系统后台不挂断地运行命令，退出终端不会影响程序的运行

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk
修改配置文件

4．在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d

vim filebeat.conf
input {
    beats {
        port => "5044"
    }
}

output {
    elasticsearch {
        hosts => ["192.168.80.10:9200","192.168.80.11:9200"]
        index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"
    }
    stdout {
        codec => rubydebu=
    }
}

#启动 logstash
logstash -f filebeat.conf

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk

5．浏览器访问 http://192.168.60.12:5601 登录 Kibana

单击“Create Index Pattern”按钮添加索引“httpd-*”，单击 “create” 按钮创建，单击 “Discover” 按钮可查看图表信息及日志信息。

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk

logstash 的四大插件（grok、date、mutate、multiline）

grok ：将大文本字段进行分片成多个小字段 %{内置正则:字段名} (?<字段名>自定义正则)
date ：将logstash收集的日志事件时间与实际记录的日志时间进行格式统一，需要配合grok模块使用
mutate ：对logstash收集的日志事件字段进行再处理，可以重命名，删除，替换和修改事件中的字段
multiline ：将多行日志内容合并成一整行

链接: 内置正则大全

grok 正则捕获插件

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.60.12 GET /index.html 15824 0.043
  可以使用如下grok pattern来匹配这种记录
  %{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk
结合logstash

input {
    beats {
        port => "5044"
    }
}

filter {
  grok {
    match => ["message", "%{IP:remote_addr} - - \[%{HTTPDATE:log_time}\] \"%{WORD:http_method} %{URIPATH:request_url} (?<http_ver>.+)\" %{NUMBER:response_code} .+ \".+\" \"(?<user_angent>.+)\""]
  }
}
#用match进行匹配 ["匹配字段", "表达式匹配并将其添加到指定字段"]

output {
    elasticsearch {
        hosts => ["192.168.60.5:9200","192.168.60.8:9200"]
        index => "%{[fields][service_name]}-%{+yyyy.MM.dd}"
    }
    stdout {
        codec => rubydebug
    }
}

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk

ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）,elk

自定义表达式
自定义表达式格式：（?<自定义名称>正则表达式）

mutate 数据修改插件

它提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段。

//Mutate 过滤器常用的配置选项
- add_field ：向事件添加新字段，也可以添加多个字段
- remove_field ：从事件中删除任意字段（他只能删除logstash自己的字段或者日志的字段无法删除通过filedeat生成的字段）
- add_tag ：向事件添加任意标签，在tag字段中添加一段自定义的内容，当tag字段中超过一个内容的时候会变成数组
- remove_tag ：从事件中删除标签（如果存在）
- convert ：将字段值转换为另一种数据类型
- id ：向现场事件添加唯一的ID
- lowercase ：将字符串字段转换为其小写形式
- replace ：用新值替换字段
- strip ：删除开头和结尾的空格
- uppercase ：将字符串字段转换为等效的大写字母
- update ：用新值更新现有字段
- rename ：重命名事件中的字段
- gsub ：通过正则表达式替换字段中匹配到的值
- merge ：合并数组或 hash 事件
  split ：通过指定的分隔符分割字段中的字符串为数组

//示例：

●将字段old_field重命名为new_field
input 
。。。。
filter {
    mutate {
	    #写法1，使用中括号括起来
        rename => ["old_field" => "new_field"]

        #写法2，使用大括号{}括起来
	    rename => { "old_field" => "new_field" }		
    }
}
output
。。。

●添加字段
input 
。。。。
filter {
    mutate {
        add_field => {
        	"f1" => "field1"
        	"f2" => "field2"
        	#添加f1字段 以及他的值field1
        }
    }
}
output
。。。

●将字段删除
input 
。。。。
filter {
    mutate {
        remove_field  =>  ["message", "@version", "tags"]
    }
}
output
。。。

●将filedName1字段数据类型转换成string类型，filedName2字段数据类型转换成float类型
input 
。。。。
filter {
    mutate {
        #写法1，使用中括号括起来
        convert  =>  ["filedName1", "string"]
		
        #写法2，使用大括号{}括起来
		convert => { "filedName2" => "float" }
    }
}
output
。。。

●将filedName字段中所有"/“字符替换为”_"
input 
。。。。
filter {
    mutate {
        gsub => ["filedName", "/" , "_"]
    }
}
output
。。。

●将filedName字段中所有",“字符后面添加空格
input 
。。。。
filter {
    mutate {
        gsub => ["filedName", "," , ", "]
    }
}
output
。。。

●将filedName字段以"|"为分割符拆分数据成为数组
input 
。。。。
filter {
    mutate {
        split => ["filedName", "|"]
    }
}
output
。。。

●合并 “filedName1” 和 “ filedName2” 两个字段
input 
。。。。
filter {
    merge  { "filedName2" => "filedName1" }
}
output
。。。

●用新值替换filedName字段的值
input 
。。。。
filter {
    mutate {
        replace => { "filedName" => "new_value" }
    }
}
output
。。。

●添加字段first，值为message数组的第一个元素的值
input 
。。。。
filter {
    mutate {
        split => ["message", "|"]
        add_field => {
            "first" => "%{[message][0]}"　　　　
        }　
    }
}
output
。。。

●有条件的添加标签
input 
。。。。
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"]
        }
    }
}
output
。。。

multiline 多行合并插件

java错误日志一般都是一条日志很多行的，会把堆栈信息打印出来，当经过 logstash 解析后，每一行都会当做一条记录存放到 ES，那这种情况肯定是需要处理的。这里就需要使用 multiline 插件，对属于同一个条日志的记录进行拼接。

例：

2022-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

//安装 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

bin/logstash-plugin install file:///usr/share/logstash/logstash-filter-multiline.zip

检查下插件是否安装成功，可以执行以下命令查看插件列表
bin/logstash-plugin list

//使用 multiline 插件

第一步：每一条日志的第一行开头都是一个时间，可以用时间的正则表达式匹配到第一行。
第二步：然后将后面每一行的日志与第一行合并。
第三步：当遇到某一行的开头是可以匹配正则表达式的时间的，就停止第一条日志的合并，开始合并第二条日志。
第四步：重复第二步和第三步。

filter {
  multiline {
    pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}"
    negate => true
    what => "previous"
  }
}

●pattern：用来匹配文本的表达式，也可以是grok表达式

●what：如果pattern匹配成功的话，那么匹配行是归属于上一个事件，还是归属于下一个事件。previous: 归属于上一个事件，向上合并。next: 归属于下一个事件，向下合并

●negate：是否对pattern的结果取反。false：不取反，是默认值。true：取反。将多行事件扫描过程中的行匹配逻辑取反（如果pattern匹配失败，则认为当前行是多行事件的组成部分）

date 时间处理插件

用于分析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。

在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

//案例：
192.168.80.10 - - [07/Feb/2022:16:24:19 +0800] “GET /HTTP/1.1” 403 5039

现在我们想转换时间，那就要写出"dd/MMM/yyyy:HH:mm:ss Z"
你发现中间有三个M，你要是写出两个就不行了，因为两个大写的M表示两位数字的月份，可是我们要解析的文本中，月份则是使用简写的英文，所以只能去找三个M。还有最后为什么要加上个大写字母Z，因为要解析的文本中含有“+0800”时区偏移，因此我们要加上去，否则filter就不能正确解析文本数据，从而转换时间戳失败。

filter{
      grok{
           match => {"message" => ".* -\ -\ \[%{HTTPDATE:timestamp}\]"}
      }
      date{
           match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
    }
}

运行结果：
{
          "host" => "localhost",
     "timestamp" => "07/Feb/2022:16:24:19 +0800",
    "@timestamp" => 2022-02-07T08:24:19.000Z,
       "message" => "192.168.80.10 - - [07/Feb/2022:16:24:19 +0800] \"GET /HTTP/1.1\" 403 5039",
      "@version" => "1"
}

在上面那段rubydebug编码格式的输出中，@timestamp字段虽然已经获取了timestamp字段的时间，但是仍然比北京时间晚了8个小时，这是因为在Elasticsearch内部，对时间类型字段都是统一采用UTC时间，而日志统一采用UTC时间存储，是国际安全、运维界的一个共识。其实这并不影响什么，因为ELK已经给出了解决方案，那就是在Kibana平台上，程序会自动读取浏览器的当前时区，然后在web页面自动将UTC时间转换为当前时区的时间。文章来源地址https://www.toymoban.com/news/detail-570557.html

到了这里，关于ELK + Filebeat 部署及 logstash 的四大插件（grok、date、mutate、multiline）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！