介绍

logstash拥有丰富的filter插件,它们扩展了进入过滤器的原始数据，进行复杂的逻辑处理，甚至可以无中生有的添加新的 logstash 事件到后续的流程中去！Grok 是 Logstash 最重要的插件之一。也是迄今为止使蹩脚的、无结构的日志结构化和可查询的最好方式。Grok在解析 syslog logs、apache and other webserver logs、mysql logs等任意格式的文件上表现完美。

官网地址：https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html#_getting_help_116

使用grok前注意

grok 模式是正则表达式，因此这个插件的性能受到正则表达式引擎严重影响。尽管知道 grok 模式与日志条目可以多快匹配非常重要，但是了解它在什么时候匹配失败也很重要。匹配成功和匹配失败的性能可能会差异很大。

原文地址：https://www.elastic.co/blog/do-you-grok-grok

翻译地址：https://segmentfault.com/a/1190000013051254

grok基础

grok模式的语法如下：

%{SYNTAX:SEMANTIC}

SYNTAX：代表匹配值的类型,例如3.44可以用NUMBER类型所匹配,127.0.0.1可以使用IP类型匹配。

SEMANTIC：代表存储该值的一个变量名称,例如 3.44 可能是一个事件的持续时间,127.0.0.1可能是请求的client地址。所以这两个值可以用 %{NUMBER:duration} %{IP:client} 来匹配。

你也可以选择将数据类型转换添加到Grok模式。默认情况下，所有语义都保存为字符串。如果您希望转换语义的数据类型，例如将字符串更改为整数，则将其后缀为目标数据类型。例如%{NUMBER:num:int}将num语义从一个字符串转换为一个整数。目前唯一支持的转换是int和float。

例子： 通过这种语法和语义的思想，我们可以从一个示例日志中抽出有用的字段，就像这个虚构的http请求日志：

55.3.244.1 GET /index.html 15824 0.043

可以使用如下grok pattern来匹配这种记录：

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

我们在logstash.conf中添加过滤器配置：

filter {grok {match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}

以下是filter结果：

● client: 55.3.244.1

● method: GET

● request: /index.html

● bytes: 15824

● duration: 0.043

Logstash附带约120个模式。你可以在这里找到它们https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

正则表达式

Grok位于正则表达式之上，所以任何正则表达式在grok中都是有效的。正则表达式库是Oniguruma，您可以在Oniguruma网站上看到完整支持的regexp语法。

中文学习正则表达式网站：http://www.runoob.com/regexp/regexp-tutorial.html

自定义类型

更多时候logstash grok没办法提供你所需要的匹配类型，这个时候我们可以使用自定义

第一种

直接使用Oniguruma语法来命名捕获，它可以让你匹配一段文本并保存为一个字段：

(?<field_name>the pattern here)

例如，日志有一个queue_id 为一个长度为10或11个字符的十六进制值。使用下列语法可以获取该片段，并把值赋予queue_id

(?<queue_id>[0-9A-F]{10,11})

第二种

创建自定义 patterns 文件。

①创建一个名为patterns其中创建一个文件postfix （文件名无关紧要,随便起）,在该文件中，将需要的模式写为模式名称，空格，然后是该模式的正则表达式。例如：

#contents of ./patterns/postfix:
POSTFIX_QUEUEID [0-9A-F]{10,11}

②然后使用这个插件中的patterns_dir设置告诉logstash目录是你的自定义模式。这是一个完整的示例的示例日志:

Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>
filter {grok {patterns_dir => ["./patterns"]match => { "message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" }}
}

匹配结果如下：

● timestamp: Jan 1 06:25:43

● logsource: mailserver14

● program: postfix/cleanup

● pid: 21403

● queue_id: BEF25A72965

● syslog_message: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>

Grok过滤器配置选项:

break_on_match

● 值类型是布尔值

● 默认是true

● 描述：match可以一次设定多组,预设会依照顺序设定处理，如果日志满足设定条件，则会终止向下处理。但有的时候我们会希望让Logstash跑完所有的设定，这时可以将break_on_match设为false。

keep_empty_captures

● 值类型是布尔值

● 默认值是 false

● 描述：如果为true，捕获失败的字段将设置为空值

match

● 值类型是数组

● 默认值是 {}

● 描述：字段⇒值匹配

例如：

filter {grok { match => { "message" => "Duration: %{NUMBER:duration}" } }
}

#如果你需要针对单个字段匹配多个模式，则该值可以是一组,例如：

filter {grok { match => { "message" => [ "Duration: %{NUMBER:duration}", "Speed: %{NUMBER:speed}" ] } }
}

named_captures_only

● 值类型是布尔值

● 默认值是 true

● 描述：If true, only store named captures from grok.（暂不清楚有什么用）

overwrite

● 值类型是 array

● 默认是[]

● 描述：覆盖字段内容

例如：

filter {grok {match => { "message" => "%{SYSLOGBASE} %{DATA:message}" }overwrite => [ "message" ]}
}

如果日志是May 29 16:37:11 sadness logger: hello world

经过match属性match => { “message” => “%{SYSLOGBASE} %{DATA:message}” }处理后，

message的值变成了hello world。这时如果使用了overwrite => [ “message” ]属性，那么原来的message的值将被覆盖成新值。

pattern_definitions

● 值类型是 数组

● 默认值是 {}

● 描述：模式名称和模式正则表达式，也是用于定义当前过滤器要使用的自定义模式。匹配现有名称的模式将覆盖预先存在的定义。可以将此视为仅适用于grok定义的内联模式，patterns_dir是将模式写在外部。

例如：

filter {grok {patterns_dir => "/usr/local/elk/logstash/patterns"pattern_definitions => {"MYSELFTIMESTAMP" => "20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND})"}match => {"message" => ["%{MYSELFTIMESTAMP:timestamp} %{JAVACLASS:message}","%{MYSELF:content}"]}}
}

patterns_dir

● 值类型是数组

● 默认值是 []

● 描述：一些复杂的正则表达式，不适合直接写到filter中，可以指定一个文件夹，用来专门保存正则表达式的文件，需要注意的是该文件夹中的所有文件中的正则表达式都会被依次加载，包括备份文件。

patterns_dir => ["/opt/logstash/patterns", "/opt/logstash/extra_patterns"]

正则文件以文本格式描述：

NAME PATTERN
#空格前是正则表达式的名称，空格后是具体的正则表达式

例如：这是一个数字的表达式

NUMBER \d+

patterns_file_glob

● 属性值的类型：string

● 默认值：“*”

● 描述：针对patterns_dir属性中指定的文件夹里哪些正则文件，可以在这个filter中生效，需要本属性来指定。默认值“*”是指所有正则文件都生效。

tag_on_failure

● 值类型是数组

● 默认值是 [“_grokparsefailure”]

●描述：没有成功匹配时，将值附加到字段

tag_on_timeout

● 值类型是字符串

● 默认值是 “_groktimeout”

● 描述：如果Grok正则表达式超时，则应用标记。

timeout_millis

● 值类型是数字

● 默认值是 30000

● 描述： 尝试在这段时间后终止正则表达式。如果应用了多个模式，则这适用于每个模式。这将永远不会提前超时，但超时可能需要一些时间。实际的超时时间是基于250ms量化的近似值。设置为0以禁用超时。

常用选项:

所有过滤器插件都支持以下配置选项：

add_field

● 值类型是散列

● 默认值是 {}

● 描述：在匹配日志中增加一个 field，可以通过%{field}动态命名field名或field的值。例如：

filter {grok {add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }}
}

# 你也可以一次添加多个字段

filter {grok {add_field => {"foo_%{somefield}" => "Hello world, from %{host}""new_field" => "new_static_value"}}
}

add_tag

● 值类型是数组

● 默认值是 []

● 描述：如果此过滤器成功，请向该事件添加任意标签。标签可以是动态的，并使用%{field} 语法包含事件的一部分。

例如：

filter {grok {add_tag => [ "foo_%{somefield}" ]}
}

# 你也可以一次添加多个标签

filter {grok {add_tag => [ "foo_%{somefield}", "taggedy_tag"]}
}

enable_metric

● 值类型是布尔值

● 默认值是 true

● 描述：禁用或启用度量标准

id

● 值类型是字符串

● 此设置没有默认值。

● 描述：向插件实例添加唯一ID，此ID用于跟踪插件特定配置的信息。

例如：

filter {grok {id => "ABC"}
}

periodic_flush

● 值类型是布尔值

● 默认值是 false

● 描述：如果设置为ture，会定时的调用filter的更新函数（flush method）

remove_field

● 值的类型：array

● 默认值：[]

● 描述：删除当前文档中的指定filted

filter {grok {remove_field => [ "foo_%{somefield}" ]}
}

# 你也可以一次移除多个字段:

filter {grok {remove_field => [ "foo_%{somefield}", "my_extraneous_field" ]}
}

remove_tag

● 值类型是数组

● 默认值是 []

● 描述：如果此过滤器成功，请从该事件中移除任意标签。标签可以是动态的，并使用%{field} 语法包括事件的一部分。

例如：

filter {grok {remove_tag => [ "foo_%{somefield}" ]}
}

# 你也可以一次删除多个标签

filter {grok {remove_tag => [ "foo_%{somefield}", "sad_unwanted_tag"]}
}