EFLK与logstash过滤

一、Filebeat工作原理：

二、为什么要使用Filebeat：

三、Filebeat和Logstash的区别：

四、logstash 的过滤插件：

五、Filebeat+ELK 部署：

1. 安装filebeat：

2. 设置 filebeat 的主配置文件：

3. 启动 filebeat:

4. 新建一个 Logstash 配置文件:

六、grok 使用：

1.内置正则表达式调用：

2. 自定义表达式调用：

2.1 格式：

七、multiline 多行合并插件：

1. 安装 multiline 插件：

2. 使用 multiline 插件：

八、mutate 数据修改插件：

1.Mutate 过滤器常用的配置选项：

2.示例：

2.1 将字段old_field重命名为new_field：

2.2 添加字段：

2.3 将字段删除：

2.4 转换字段类型：

2.5 替换字符：

九、date 时间处理插件：

1.时间戳详解：

2 .案例：

一、Filebeat工作原理：

Filebeat可以保持每个文件的状态，并且频繁地把文件状态从注册表里更新到磁盘。这里所说的文件状态是用来记录上一次Harvster读取文件时读取到的位置，以保证能把全部的日志数据都读取出来，然后发送给output。如果在某一时刻，作为output的ElasticSearch或者Logstash变成了不可用，Filebeat将会把最后的文件读取位置保存下来，直到output重新可用的时候，快速地恢复文件数据的读取。在Filebaet运行过程中，每个Prospector的状态
信息都会保存在内存里。如果Filebeat出行了重启，完成重启之后，会从注册表文件里恢复重启之前的状态信息，让FIlebeat继续从之前已知的位置开始进行数据读取。

二、为什么要使用Filebeat：

因为logstash是jvm跑的，资源消耗比较大，启动一个logstash就需要消耗500M左右的内存（这就是为什么logstash启动特别慢的原因），而filebeat只需要10来M内存资源。常用的ELK日志采集方案中，大部分的做法就是将所有节点的日志内容通过filebeat发送到logstash，logstash根据配置文件进行过滤。然后将过滤之后的文件输送到elasticsearch中，通过kibana去展示。

三、Filebeat和Logstash的区别：

内存	大	小
CPU	大	小
插件	多	多
功能	从多种输入端采集并实时解析和转换数据并输出到多种输出端	传输
过滤能力	强大的过滤能力	有过滤能了但是弱
轻重	相对较重	轻量级二进制文件
进程	一台服务器只允许一个logstash进程，挂掉之后需要手动启动
集群	单节点	单节点

filebeat收集日志统一交由logstash过滤，然后存储在es上，最后通过kibana web图形化展示

四、logstash 的过滤插件：

grok 正则捕获插件
multiline 多行合并插件
mutate 数据修改插件
date 时间处理插件

五、Filebeat+ELK 部署：

node1节点（2C/4G）：node1/192.168.88.101 Elasticsearch
node2节点（2C/4G）：node2/192.168.80.103 Elasticsearch
Apache节点：apache/192.168.88.100 Logstash Kibana Apache
Filebeat节点：filebeat/192.168.88.104 Filebeat

1. 安装filebeat：

#上传软件包 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

2. 设置 filebeat 的主配置文件：

cd /usr/local/filebeat

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

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

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

3. 启动 filebeat:

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

4. 新建一个 Logstash 配置文件:

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

output {
    elasticsearch {
        hosts => ["192.168.88.101:9200","192.168.88.103:9200"]
        index => "nginx_access.log-%{+YYYY.MM.dd}"
    }
}

logstash -f filebeat.conf  ##启动

浏览器访问 http://192.168.88.100:5601 登录 Kibana，单击“Create Index Pattern”按钮添加索引“nginx_access*”，单击 “create” 按钮创建，单击 “Discover” 按钮可查看图表信息及日志信息。

六、grok 使用：

使用文本片段切分的方式来切分日志事件

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.conf
filter {
  grok {
    match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
  }
}

以下是filter结果：

client_id_address: 192.168.80.10
method: GET
request: /index.html
bytes: 15824
http_response_time: 0.043

logstash 官方也给了一些常用的常量来表达那些正则表达式，可以到这个 Github 地址查看有哪些常用的常量：
https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/grok-patterns

2. 自定义表达式调用：

2.1 格式：

(?<field_name>pattern)

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\.]+)

filter {
  grok {
    match => { "message" => "(?<remote_addr>%{IP}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)"}
  }
}

七、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

1. 安装 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

2. 使用 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匹配失败，则认为当前行是多行事件的组成部分）

八、mutate 数据修改插件：

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

1.Mutate 过滤器常用的配置选项：

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

2.示例：

2.1 将字段old_field重命名为new_field：

filter {
    mutate {
	    #写法1，使用中括号括起来
        rename => ["old_field" => "new_field"]

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

2.2 添加字段：

filter {
    mutate {
        add_field => {
        	"f1" => "field1"
        	"f2" => "field2"
        }
    }
}

2.3 将字段删除：

filter {
    mutate {
        remove_field  =>  ["message", "@version", "tags"]
    }
}

2.4 转换字段类型：

将filedName1字段数据类型转换成string类型，filedName2字段数据类型转换成float类型

filter {
    mutate {
        #写法1，使用中括号括起来
        convert  =>  ["filedName1", "string"]
		
        #写法2，使用大括号{}括起来
		convert => { "filedName2" => "float" }
    }
}

2.5 替换字符：

filter {
    mutate {
        gsub => ["filedName", "/" , "_"]
    }
}

九、date 时间处理插件：

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

在Logstash产生了一个Event对象的时候，会给该Event设置一个时间，字段为“@timestamp”，同时，我们的日志内容一般也会有时间，但是这两个时间是不一样的，因为日志内容的时间是该日志打印出来的时间，而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间，所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点，因为Logstash监控数据变化，数据输入，创建Event导致的时间延迟。这两个时间都可以使用，具体要根据自己的需求来定。

配合gork使用

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参数。

1.时间戳详解：

●年
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

2 .案例：

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时间转换为当前时区的时间。