Elastic：在 Grok 中运用 custom pattern 来定义 pattern

在我之前的文章 “Elastic可观测性 - 数据结构化及处理”，讲述了如果把一个非结构化的数据变为一个结构化的数据结构。其中Grok processor 是非常重要的一个。在今天的文章中，我们来更加深入地对它进行描述。今天的这个 Grok 的实践也适用于 Logstash 的 Grok filer。关于 Logstash Grok 的教程，你可以参阅文章 “Logstash：Grok filter 入门”。

我们先来看一下如下的一个日志：

157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing

在上面的日志中，我们可以看到一个日期信息：2019 09 29 00:39:02.912。它是被空格字符串所分开，如果没有正确的 Grok pattern 来帮我们提取的话，我们将会很难提取到一个完整的日期。我们的日志信息符合如下的一个数据结构：

ip timestamp server Process process_id action

首先，我们打开 Kibana:

我们可以先提取 IP:

之后的，就是年，月，日，及时间。我们可以通过如下的方式来进行提取：

运用 YEAY 来提取年份
运用 MONTHNUM 来提取月份
运用 MONTHDAY 来提取日期
运用 TIME 来提取时间
运用 WORD 来提取一个单词
运用 NUMBER 来提取一个数值
对于 Process 来说，我们就不提取了，忽略它

这样，我们可以使用如下的 Grok pattern:

%{IP:ip} %{YEAR:year} %{MONTHNUM:month} %{MONTHDAY:day} %{TIME:time} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}

显然，它正确地解析了我们的日志，但是美中不足的是我们最终需要的是一个真正的日期，而不是用 year, month, day, time 来表示的一个时间。我们可以点击上面的 custerm pattern，并输入一下的句子：

EVENTDATE %{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME}

在上面，我们定义了 EVENDATE 为 YEAR， MONTHNUM, MONTHDAY 及 TIME 的组合。那么我们该如和应用上面的 custom patttern呢？

我们必须修改上面的 Grok pattern 为：

%{IP:ip} %{EVENTDATE:@timestamp} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}

从上面，我们可以看出来，我们的 EVENTDATE 起作用了。它正确地解析了我们的时间。

那么在我们实际的使用中，我们该如何地应用呢？

我们可以创建如下的一个命令：

POST /_ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "grok": {
          "field": "message",
          "patterns": [
            "%{IP:ip} %{EVENTDATE:@timestamp} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}"
          ],
          "pattern_definitions": {
            "EVENTDATE": "%{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME}"
          }
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "message": "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
      }
    }
  ]
}

运行上面的命令：

{
  "docs" : [
    {
      "doc" : {
        "_index" : "_index",
        "_type" : "_doc",
        "_id" : "_id",
        "_source" : {
          "server" : "myserver",
          "process_id" : "107673",
          "@timestamp" : "2019 09 29 00:39:02.912",
          "ip" : "157.97.192.70",
          "action" : "Initializing",
          "message" : "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
        },
        "_ingest" : {
          "timestamp" : "2020-06-15T08:33:01.28191Z"
        }
      }
    }
  ]
}

上面显示我们的日志被正确地解析并结构化。

另外一种方法是通过 set processor 来把上面的日期相关的字段来组成我们需要的 @timestamp 字段。

POST /_ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "grok": {
          "field": "message",
          "patterns": [
            "%{IP:ip} %{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}"
          ]
        }
      },
      {
        "set": {
          "field": "@timestamp",
          "value": "{{year}} {{month}} {{day}} {{time}}"
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "message": "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
      }
    }
  ]
}

在上面，我们通过：

      {
        "set": {
          "field": "@timestamp",
          "value": "{{year}} {{month}} {{day}} {{time}}"
        }
      }

来把 @timestamp 进行定义，它组合了 year, month, day 及 time 的值。

Elastic：在 Grok 中运用 custom pattern 来定义 pattern

猜你喜欢