一、grok
1. grok 使用文本片段切分的方式来切分日志事件
filter { grok { match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" } } }
2. 内置正则表达式调用
%{SYNTAX:SEMANTIC}
SYNTAX：代表匹配值的类型，例如，0.11可以NUMBER类型所匹配，10.222.22.25可以使用IP匹配。常用表达式：
============================ 常用的常量 =================================== USERNAME [a-zA-Z0-9._-]+ USER %{USERNAME} EMAILLOCALPART [a-zA-Z][a-zA-Z0-9_.+-=:]+ EMAILADDRESS %{EMAILLOCALPART}@%{HOSTNAME} INT (?:[+-]?(?:[0-9]+)) BASE10NUM (?<![0-9.+-])(?>[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+))) NUMBER (?:%{BASE10NUM}) BASE16NUM (?<![0-9A-Fa-f])(?:[+-]?(?:0x)?(?:[0-9A-Fa-f]+)) BASE16FLOAT \b(?<![0-9A-Fa-f.])(?:[+-]?(?:0x)?(?:(?:[0-9A-Fa-f]+(?:\.[0-9A-Fa-f]*)?)|(?:\.[0-9A-Fa-f]+)))\b POSINT \b(?:[1-9][0-9]*)\b NONNEGINT \b(?:[0-9]+)\b WORD \b\w+\b NOTSPACE \S+ SPACE \s* DATA .*? GREEDYDATA .* QUOTEDSTRING (?>(?<!\\)(?>"(?>\\.|[^\\"]+)+"|""|(?>'(?>\\.|[^\\']+)+')|''|(?>(?>\\.|[^\\]+)+)|)) UUID [A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12} # URN, allowing use of RFC 2141 section 2.3 reserved characters URN urn:[0-9A-Za-z][0-9A-Za-z-]{0,31}:(?:%[0-9a-fA-F]{2}|[0-9A-Za-z()+,.:=@;$_!*'/?#-])+ ============================ Networking =================================== MAC (?:%{CISCOMAC}|%{WINDOWSMAC}|%{COMMONMAC}) CISCOMAC (?:(?:[A-Fa-f0-9]{4}\.){2}[A-Fa-f0-9]{4}) WINDOWSMAC (?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2}) COMMONMAC (?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2}) IPV6 ((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))(%.+)? (img-SbCluW8t-1689073158082)(?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5])][.](?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5])[.](?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5]))(?![0-9]) IP (?:%{IPV6}|%{IPV4}) HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b) IPORHOST (?:%{IP}|%{HOSTNAME}) HOSTPORT %{IPORHOST}:%{POSINT} ============================ paths =================================== PATH (?:%{UNIXPATH}|%{WINPATH}) UNIXPATH (/([\w_%!$@:.,+~-]+|\\.)*)+ TTY (?:/dev/(pts|tty([pq])?)(\w+)?/?(?:[0-9]+)) WINPATH (?>[A-Za-z]+:|\\)(?:\\[^\\?*]*)+ URIPROTO [A-Za-z]([A-Za-z0-9+\-.]+)+ URIHOST %{IPORHOST}(?::%{POSINT:port})? # uripath comes loosely from RFC1738, but mostly from what Firefox # doesn't turn into %XX URIPATH (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%&_\-]*)+ #URIPARAM \?(?:[A-Za-z0-9]+(?:=(?:[^&]*))?(?:&(?:[A-Za-z0-9]+(?:=(?:[^&]*))?)?)*)? URIPARAM \?[A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]<>]* URIPATHPARAM %{URIPATH}(?:%{URIPARAM})? URI %{URIPROTO}://(?:%{USER}(?::[^@]*)?@)?(?:%{URIHOST})?(?:%{URIPATHPARAM})? ============================ Months =================================== MONTH \b(?:[Jj]an(?:uary|uar)?|[Ff]eb(?:ruary|ruar)?|[Mm](?:a|ä)?r(?:ch|z)?|[Aa]pr(?:il)?|[Mm]a(?:y|i)?|[Jj]un(?:e|i)?|[Jj]ul(?:y)?|[Aa]ug(?:ust)?|[Ss]ep(?:tember)?|[Oo](?:c|k)?t(?:ober)?|[Nn]ov(?:ember)?|[Dd]e(?:c|z)(?:ember)?)\b MONTHNUM (?:0?[1-9]|1[0-2]) MONTHNUM2 (?:0[1-9]|1[0-2]) MONTHDAY (?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9]) ============================ Days =================================== DAY (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?) ============================ Years =================================== YEAR (?>\d\d){1,2} HOUR (?:2[0123]|[01]?[0-9]) MINUTE (?:[0-5][0-9]) # '60' is a leap second in most time standards and thus is valid. SECOND (?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?) TIME (?!<[0-9])%{HOUR}:%{MINUTE}(?::%{SECOND})(?![0-9]) # datestamp is YYYY/MM/DD-HH:MM:SS.UUUU (or something like it) DATE_US %{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR} DATE_EU %{MONTHDAY}[./-]%{MONTHNUM}[./-]%{YEAR} ISO8601_TIMEZONE (?:Z|[+-]%{HOUR}(?::?%{MINUTE})) ISO8601_SECOND (?:%{SECOND}|60) TIMESTAMP_ISO8601 %{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:?%{MINUTE}(?::?%{SECOND})?%{ISO8601_TIMEZONE}? DATE %{DATE_US}|%{DATE_EU} DATESTAMP %{DATE}[- ]%{TIME} TZ (?:[APMCE][SD]T|UTC) DATESTAMP_RFC822 %{DAY} %{MONTH} %{MONTHDAY} %{YEAR} %{TIME} %{TZ} DATESTAMP_RFC2822 %{DAY}, %{MONTHDAY} %{MONTH} %{YEAR} %{TIME} %{ISO8601_TIMEZONE} DATESTAMP_OTHER %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{TZ} %{YEAR} DATESTAMP_EVENTLOG %{YEAR}%{MONTHNUM2}%{MONTHDAY}%{HOUR}%{MINUTE}%{SECOND} ============================ Syslog Dates =================================== SYSLOGTIMESTAMP %{MONTH} +%{MONTHDAY} %{TIME} PROG [\x21-\x5a\x5c\x5e-\x7e]+ SYSLOGPROG %{PROG:program}(?:\[%{POSINT:pid}\])? SYSLOGHOST %{IPORHOST} SYSLOGFACILITY <%{NONNEGINT:facility}.%{NONNEGINT:priority}> HTTPDATE %{MONTHDAY}/%{MONTH}/%{YEAR}:%{TIME} %{INT} ============================ Shortcuts =================================== QS %{QUOTEDSTRING} ============================ Log formats =================================== SYSLOGBASE %{SYSLOGTIMESTAMP:timestamp} (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}: ============================ Log Levels =================================== ```dart LOGLEVEL ([Aa]lert|ALERT|[Tt]race|TRACE|[Dd]ebug|DEBUG|[Nn]otice|NOTICE|[Ii]nfo|INFO|[Ww]arn?(?:ing)?|WARN?(?:ING)?|[Ee]rr?(?:or)?|ERR?(?:OR)?|[Cc]rit?(?:ical)?|CRIT?(?:ICAL)?|[Ff]atal|FATAL|[Ss]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)
SEMANTIC：表示存储该值的一个变量声明，它会存储在elasticsearch当中方便kibana做字段搜索和统计
3. 自定义表达式调用
(?<field_name>pattern)
举例
目标信息：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\.]+) 或者 (?<remote_addr>%{IPV4}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)
二、date
1. 概念
用于分析字段中的日期，然后使用该日期或时间戳作为事件的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参数。
2、时间戳详解
# 年 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# 年 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
3. 案例
192.168.80.10 - - [07/Feb/2022:16:24:19 +0800] “GET /HTTP/1.1” 403 5039 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" }
1. 它提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段。
2. Mutate 过滤器常用的配置选项
3. 示例
1）rename：对于已经存在的字段，重命名其字段名称。
filter { mutate { rename => ["syslog_host", "host"] } }
2）update：更新字段内容，如果字段不存在，不会新建
filter { mutate { update => { "sample" => "My new message" } } }
3）replace：与 update 功能相同，区别在于如果字段不存在则会新建字段
filter { mutate { replace => { "message" => "%{source_host}: My new message" } } }
4）convert：数据类型转换。
filter { mutate { convert => ["request_time", "float"] } }
5）gsub：提供了通过正则表达式实现文本替换的功能
filter { mutate { gsub => [ "fieldname", "/", "_", # replace backslashes, question marks, hashes, and minuses # with a dot "." "fieldname2", "[\\?#-]", "." ] } }
6）uppercase / lowercase：大小写转换
filter { mutate { uppercase => [ "fieldname" ] } }
7）split：将提取到的某个字段按照某个字符分割
filter { mutate { split => ["message", "|"] } }
针对字符串 "123|321|adfd|dfjld*=123"，可以看到输出结果：
{ "message" => [ [0] "123", [1] "321", [2] "adfd", [3] "dfjld*=123" ], "@version" => "1", "@timestamp" => "2014-08-20T15:58:23.120Z", "host" => "raochenlindeMacBook-Air.local" }
8）strip：类似 trim，只去除首尾的空白字符
filter { mutate { strip => ["field1", "field2"] } }
9）remove_field：删除字段：
filter { mutate { remove_field => [ "foo_%{somefield}" ] } }
10）join：将类型为 array 的字段中的 array 元素使用指定字符为分隔符聚合成一个字符串。如我们可以将 split 分割的结果再重新聚合起来：
filter { mutate { split => ["message", "|"] } mutate { join => ["message", ","] } }
输出结果：
{ "message" => "123,321,adfd,dfjld*=123", "@version" => "1", "@timestamp" => "2014-08-20T16:01:33.972Z", "host" => "raochenlindeMacBook-Air.local" }
11）merge：对于几个类型为 array 或 hash 或 string 的字段，我们可以使用 merge 合并
filter { mutate { merge => [ "dest_field", "added_field" ] } }
1、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
2、安装 multiline 插件
1）在线安装插件
cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline
2）离线安装插件
先在有网的机器上在线安装插件，然后打包，拷贝到服务器，执行安装命令 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
3）检查下插件是否安装成功，可以执行以下命令查看插件列表
bin/logstash-plugin list
3、使用 multiline 插件
1）每一条日志的第一行开头都是一个时间，可以用时间的正则表达式匹配到第一行。
2）然后将后面每一行的日志与第一行合并。
3）当遇到某一行的开头是可以匹配正则表达式的时间的，就停止第一条日志的合并，开始合并第二条日志。
4）重复第二步和第三步。
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匹配失败，则认为当前行是多行事件的组成部分）