关键词：

Logstash 可以轻松解析 CSV 和 JSON 文件，因为这些格式的数据组织得很好，可以进行 Elasticsearch 分析。 但是，有时我们需要处理非结构化数据，例如纯文本日志。 在这些情况下，我们需要使用 Logstash Grok 或其他第三方服务解析数据以使其成为结构化数据。 本教程将通过使用 Logstash Grok 进行解析，帮助你利用 Elasticsearch 的分析和查询功能。

因此，让我们深入了解如何使用 Logstash Grok 过滤器处理非结构化数据。

理论

在 CSV 文件中，每条记录都有相同的字段列表。 这些字段的顺序可预测地重复，以便任何程序以结构化方式读取。 自然，这对于 Elasticsearch 来说是一个理想的情况。

在其他情况下，无法以这种理想方式组织信息。 考虑一个典型的 Linux 系统日志。 多个程序将它们的状态或错误消息写入同一个日志文件。 在日志的一行中，最后一个字段可能是错误代码。 但在下一行，最后一个字段可能是 IP 地址。 这意味着解析器现在需要有一种方法来 “检测” 每个字段所代表的内容。 让我们看看 Grok 过滤器是如何做到这一点的。

理解 Grok 模式和语法：

当我们看到 john@example.com 之类的内容时，我们会立即想到 电子邮件地址，因为我们已经学会了识别这种模式。 Grok 可以通过分析每段文本并检查它是否与你指示它查找的模式匹配来做同样的事情。

Grok 在幕后使用正则表达式，或简称 regex，这对于还不熟悉它们的人来说可能看起来有点奇怪。 例如，匹配电子邮件的正则表达式如下所示：

^([a-zA-Z0-9_\\-\\.]+)@([a-zA-Z0-9_\\-\\.]+)\\.([a-zA-Z]2,3)$

幸运的是，在 Logstash 的 Grok 过滤器中已经预定义了很多常用表达式，我们可以使用它们的模式名称来代替自己编写那些复杂的字符串。你可以在地址查看：Grok filter plugin | Logstash Reference [8.4] | Elastic。

我们可以在 Grok 过滤器插件的文档页面上找到这些预定义模式名称的列表。Grok 模式的基本语法为：%SYNTAX:SEMANTIC。

简而言之，我们告诉它要寻找什么模式以及如何标记与这些模式匹配的字符串。

 回到我们之前的示例，这是定义和标记电子邮件地址的方法：

%EMAILADDRESS:client_email

此 Grok 模式将查找所有电子邮件地址并将每个地址标识为 client_email。 你可以自由选择您想要的任何标识符名称。

在我们的例子中，标识符将用作匹配值的字段名称，这些匹配值将被导入到索引中。 我们将在接下来的动手练习中看到它是如何工作的。

示例 Grok 过滤器：

让我们分析一下我们将如何使用 Grok。 考虑日志文件中的以下行：

2020-07-16T19:20:30.45+01:00 DEBUG This is a sample log

我们可以看到这里有三个逻辑组件：时间戳，然后是日志记录级别，最后是错误或状态消息。时间戳采用所谓的 ISO 格式，这很有用，因为我们已经有一个可用的 Grok 模式。那么，我们如何定义一个 Grok 过滤器来从这段日志文本中提取三个组件？

 好吧，这里是：

%TIMESTAMP_ISO8601:time %LOGLEVEL:logLevel %GREEDYDATA:logMessage

TIMESTAMP_ISO8601 和 LOGLEVEL 从我们的示例日志文本中提取前两部分。 GREEDYDATA 可能听起来很神秘，但它所做的只是匹配所分析字符串中的所有其他字符，因此它会提取文本的其余部分。

自然地，我们也想测试这些 Grok 过滤器，看看它们是否按预期工作，在我们的特定数据上。我们可以使用 Kibana 自带的 Grok Debugger，或者你可以使用网上的 Grok Debugger tool。我们现在以 Grok Debugger 为例来进行展示。

 

现在我们已经建立了一些 Grok 基础知识，让我们使用各种示例来探索这些概念。

 

动手练习

在今天的练习中，我们以最新的 Elastic Stack 8.4.3 为例进行展示。如果你还没有安装好自己的 Elastic Stack，请参考如下的文章：

• 如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch

• Kibana：如何在 Linux，MacOS 及 Windows 上安装 Elastic 栈中的 Kibana

• 如何安装 Elastic 栈中的 Logstash

在安装时，请选择 Elastic Stack 8.x 的安装。在默认的情况下，Elasticsearch 的访问是带有 HTTPS 的。

使用 Grok 解析一个日志文件

让我们应用这些新获得的知识，看看如何在示例日志文件中使用 Logstash Grok 过滤器插件。

首先，让我们创建一个目录来存储我们的示例数据：

mkdir -p $HOME/data/logstash

你需要根据自己的 HOME 目录修改上面的路径。接下来，让我们创建我们将解析的文件：

$ pwd
/Users/liuxg/data/logstash
$ vi sample.log

现在让我们将以下文本复制并粘贴到 vi 编辑器中：

2020-10-11T09:49:35Z INFO variable server value is tomcat
2020-03-14T22:50:34Z ERROR cannot find the requested resource
2020-01-02T14:58:40Z INFO initializing the bootup
2020-06-04T06:56:04Z DEBUG initializing checksum
2020-05-07T03:07:11Z INFO variable server value is tomcat

我们保存好 sample.log 文件。我们可以使用如下的命令来进行查看：

$ pwd
/Users/liuxg/data/logstash
$ cat sample.log 
2020-10-11T09:49:35Z INFO variable server value is tomcat
2020-03-14T22:50:34Z ERROR cannot find the requested resource
2020-01-02T14:58:40Z INFO initializing the bootup
2020-06-04T06:56:04Z DEBUG initializing checksum
2020-05-07T03:07:11Z INFO variable server value is tomcat

接下来，我们来创建一个 Logstash 的配置文件 grok-example.conf。它的内容如下：

grok-example.conf

input 
  file 
   path => "/Users/liuxg/data/logstash/sample.log"
   start_position => "beginning"
   sincedb_path => "/dev/null"
  


filter 
   grok 
    match =>  "message" => ['%TIMESTAMP_ISO8601:time %LOGLEVEL:logLevel %GREEDYDATA:logMessage'] 
  


output 
    stdout 

正如一个典型的 Logstash pipeline，它由一个 input，filter 及 output 组成：

我们可以看到过滤器部分下的配置非常简单。 我们指示 Logstash 使用 grok 过滤器插件，并在我们使用之前探索的相同模式和标识符的地方添加匹配指令。

"message" => ['%TIMESTAMP_ISO8601:time %LOGLEVEL:logLevel %GREEDYDATA:logMessage']

我们使用如下的命令来运行这个 pipeline：

./bin/logstash -f /Users/liuxg/data/logstash/grok-example.conf

 

从上面的输入中，我们可以清楚地看到非结构化的文档，现在已经变成了结构化的文档。我们可以充分利用 Elasticsearch 的数据分析能力对数据进行分析和制作仪表盘。

如何把数据写入到 Elasticsearch

为了把数据写入到 Elasticsearch，我们需要做一些额外的工作，这是因为 Elasticsearch 的访问是 HTTPS 的。我们参考文章 “Logstash：如何连接到带有 HTTPS 访问的集群”。我们进入到 Elasticsearch 的安装目录中，并打入如下的命令：

keytool -import -file http_ca.crt -keystore truststore.p12 -storepass password -noprompt -storetype pkcs12

$ pwd
/Users/liuxg/test/elasticsearch-8.4.3/config/certs
$ ls
http.p12      http_ca.crt   transport.p12
$ keytool -import -file http_ca.crt -keystore truststore.p12 -storepass password -noprompt -storetype pkcs12
Certificate was added to keystore
$ ls
http.p12       http_ca.crt    transport.p12  truststore.p12

在上面的命令中，我们生成了一个叫做 truststore.p12 的文件，而访问它的密码是 password。

我们进一步修改 grok-example.conf 文件如下：

grok-example.conf

input 
  file 
   path => "/Users/liuxg/data/logstash/sample.log"
   start_position => "beginning"
   sincedb_path => "/dev/null"
  


filter 
   grok 
    match =>  "message" => ['%TIMESTAMP_ISO8601:time %LOGLEVEL:logLevel %GREEDYDATA:logMessage'] 
  


output 
    stdout 

    elasticsearch 
    	hosts => ["https://localhost:9200"]
    	index => "data-%+YYYY.MM.dd"
    	user => "elastic"
    	password => "6bTlJp388KkgJKWi+hQr"
		ssl_certificate_verification => true
		truststore => "/Users/liuxg/test/elasticsearch-8.4.3/config/certs/truststore.p12"
		truststore_password => "password"
  	

在上面，我们添加了 elasticsearch 输出。特别需要注意的是，我们需要根据自己的配置修改上面的 password 部分。为了说明问题的方便，我使用了超级用户 elastic 来做展示。在实际的使用中，这个是一个不好的建议。你需要根据自己的需求创建适合数据采集的用户及密码。我们重新运行 Logstash：

 ./bin/logstash -f /Users/liuxg/data/logstash/grok-example.conf

从上面的输出中，我能可以看出来数据解析是成功的。我们回到 Kibana 的界面，并打入如下的命令：

GET _cat/indices

 

我们可以看到一个以 data 开始的索引已经被生成了。我们使用如下的命令来查看它的文档数：

GET data-2022.10.10/_count

  "count": 5,
  "_shards": 
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  

上面显示有 5 个文档。我们可以对它进行搜索：

很显然，我们的文档被成功地摄入到 Elasticsearch 中了。

假如我们的文档中有不完全匹配 Grok 模式的，那么会发生什么情况呢？我们首先在 Kibana 中使用如下的命令来删除之前已经创建的索引：

DELETE data-2022.10.10

我们在之前的 sample.log 文件的最后添加如下的一个新的行：

55.12.32.134 GET /user/id/properties

$ pwd
/Users/liuxg/data/logstash
$ cat sample.log 
2020-10-11T09:49:35Z INFO variable server value is tomcat
2020-03-14T22:50:34Z ERROR cannot find the requested resource
2020-01-02T14:58:40Z INFO initializing the bootup
2020-06-04T06:56:04Z DEBUG initializing checksum
2020-05-07T03:07:11Z INFO variable server value is tomcat
55.12.32.134 GET /user/id/properties

 很显然，它的格式和我们之前的行不一样。它的第一个部分不是时间戳。我们重新运行 Logstash：

./bin/logstash -f /Users/liuxg/data/logstash/grok-example.conf

我们可以看到该文档缺少 time、logLevel 和 logMessage 字段。 message 字段显示了我们的示例日志中生成此文档的行。 我们可以看到这是没有任何字段与我们的 Grok 过滤器模式匹配的行。 添加了一个名为 _grokparsefailure 的标记，以表明解析器在文件中的这一行出现问题。

logstash：日志解析的grok模式示例(代码片段)

...和分析工具中解析日志数据。在这里查看我的Grok教程“Logstash：Grokfilter入门”。但是用Grok解析日志可能会很棘手。本博客将研究一些Grok模式示例，这些示例可以帮助你了解如何解析日志数据。什么是grok?最初的术语实际... 查看详情

logstash：grok模式示例(代码片段)

Logstash可以轻松解析CSV和JSON文件，因为这些格式的数据组织得很好，可以进行Elasticsearch分析。但是，有时我们需要处理非结构化数据，例如纯文本日志。在这些情况下，我们需要使用LogstashGrok或其他第三方服... 查看详情

logstash / grok 模式文件

】logstash/grok模式文件【英文标题】：logstash/grokpatternfile【发布时间】：2015-09-2902:24:38【问题描述】：我正在解析IIS日志，当所有模式都在配置文件中时，我一切正常。我想取出所有模式并将它们放在一个模式文件中，但似乎无... 查看详情

未找到 logstash grok 过滤器模式

】未找到logstashgrok过滤器模式【英文标题】：logstashgrokfilterpatternnotfound【发布时间】：2015-04-1419:46:12【问题描述】：我一直在尝试为logstash创建一些自定义的grok模式。他们中的大多数工作正常，但有一个让我难过。模式是：WINUSE... 查看详情

logstash：使用自定义正则表达式模式(代码片段)

有时LogstashGrok没有我们需要的模式。幸运的是我们有正则表达式库：Oniguruma。在很多时候，如果Logstash所提供的正则表达不能满足我们的需求，我们选用定制自己的表达式。定义Logstash是一种服务器端数据处理管道ÿ... 查看详情

logstash之grok(代码片段)

nginx匹配示例nginx日志格式\'$remote_user[$time_local]$http_x_Forwarded_for$remote_addr$request$status$upstream_status\'\'$http_x_forwarded_for\'\'$upstream_addr\'\'ups_resp_time:$upstream_response_time\'\'request_time:$request_time\';nginx日志示例-[09/May/2023:15:01:31+0800]11.20.1.3038.... 查看详情

grok自定义模式(代码片段)

我的logstash中有简单的消息：2018-09-3020:25:07.708INFO8013---[nio-8443-exec-3]c.e.demo.controller.UsuarioController:INICIOCHAMADA|311我想从此消息中删除以下字段。"311"答案你快到了。你只需要用反斜杠逃避管道|字符（否则它匹配801之后的INFO），如下... 查看详情

logstash grok 模式来监控 logstash 本身

】logstashgrok模式来监控logstash本身【英文标题】：logstashgrokpatterntomonitorlogstashitself【发布时间】：2016-05-0316:07:48【问题描述】：我想将logstash.log日志添加到我的ELK堆栈中，但我总是遇到grokparsefailure。我的模式在http://grokconstructor.a... 查看详情

Logstash grok 模式过滤自定义日志消息

】Logstashgrok模式过滤自定义日志消息【英文标题】：LogstashgrokpatterntofiltercustomLogmessage【发布时间】：2015-03-1608:40:25【问题描述】：我是logstash的新手，我想从日志消息中过滤文件。这是日志消息：[2015-03-1613:12:05,130]INFO-LogMediatorSe... 查看详情

Grok 用于日志文件 Logstash

】Grok用于日志文件Logstash【英文标题】：GrokforlogfilesLogstash【发布时间】：2020-04-2212:03:00【问题描述】：我需要编写一个grok模式来检索“****”中的内容-----Startofscriptforserversitboap1at**FriApr1714:24:19**HKT2020---------**user11**8775110Apr16?00:00... 查看详情

logstash / grok 自定义字段

】logstash/grok自定义字段【英文标题】：logstash/grokcustomfileds【发布时间】：2017-07-1810:37:32【问题描述】：我是ELK堆栈的全新用户。我在从日志中过滤掉特定部分时遇到了一点问题。示例日志：[2017-05-3013:58:09,336]INFO[com.qwerty.test.core... 查看详情

Logstash Grok 解析器

】LogstashGrok解析器【英文标题】：LogstashGrokparser【发布时间】：2015-12-0217:10:00【问题描述】：我是log-stash和grok的新手，我需要解析非常自定义的日志文件。我在任何地方都找不到一个很好的教程来完成这项工作。尝试了syslog示... 查看详情

grok内置的默认类型有很多种，查看所有默认类型(代码片段)

源地址：（https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns）USERNAME[a-zA-Z0-9._-]+USER%USERNAMEEMAILLOCALPART[a-zA-Z][a-zA-Z0-9_.+-=:]+EMAILADDRESS%EMAILLOCA 查看详情

记录logstash的filter使用(代码片段)

概述logstash 之所以强大和流行，与其丰富的过滤器插件是分不开的过滤器提供的并不单单是过滤的功能，还可以对进入过滤器的原始数据进行复杂的逻辑处理，甚至添加独特的新事件到后续流程中 强大的文本解析工具&nbs... 查看详情

logstash：使用logshark来调试logstash及filebeatpipelines(代码片段)

我们知道Logstash及Filebeat在Elasticsearch数据摄入及清理中起到非常大的作用。它们是常用的工具用来对数据进行处理。我们可以运用Logstash丰富的过滤器来处理数据，我们也可以使用Filebeat的processors来处理数据。使用这些工具&#x... 查看详情

logstash：使用logshark来调试logstash及filebeatpipelines(代码片段)

我们知道Logstash及Filebeat在Elasticsearch数据摄入及清理中起到非常大的作用。它们是常用的工具用来对数据进行处理。我们可以运用Logstash丰富的过滤器来处理数据，我们也可以使用Filebeat的processors来处理数据。使用这些工具&#x... 查看详情

logstash：wso api manager 日志到弹性

】logstash：wsoapimanager日志到弹性【英文标题】：logstash:wsoapimanagerlogstoelastic【发布时间】：2022-01-1520:35:00【问题描述】：我的日志如下所示，日志中的每个参数都是静态的。我尝试通过logstash发送弹性并使用Grok模式。但我看不到... 查看详情

使用logstashfiltergrok过滤日志文件(代码片段)

Logstash提供了一系列filter过滤plugin来处理收集到的logevent，根据logevent的特征去切分所需要的字段，方便kibana做visualize和dashboard的dataanalysis。所有logstash支持的event切分插件查看这里。下面我们主要讲grok切分。Grok基本介绍Grok使用... 查看详情