logstash使用filter删除不需要的日志

Posted 2023-03-16

参考技术A 日志的流转路径：
logstash收集log4j的日志，并对日志进行过滤，输出给elasticsearch，kibana从es的索引中查询数据进行展示。

有一部分日志没有多大的意义，但是占据了很大的磁盘空间，因此想通过filter将其过滤掉，再将有用的日志输出给es。

filter的过滤插件很多，用得最多的就是grok，感兴趣的可以学一下。本次需求实现用不到grok，就不赘述了。以下是logstash的配置文件。

过滤掉tenant为“unmarked”的日志，以及message以“应变数据已经被采集”开头的日志并删除。

修改配置文件后，启动logstash出现了一个报错如下：

从中可以看出：配置文件的第10行、第13列有错误。错误为符号错误；
因为我之前写的是：

不能写=，要写==才行。

Logstash的filter插件-Grok

一、介绍

Grok是一种将非结构化日志数据解析为结构化和可查询数据的好方法。

这个工具非常适合于syslog日志、apache和其他web服务器日志、mysql日志，以及通常为人类而不是计算机使用编写的任何日志格式。

二、选择

Grok or Dissect？还是两者兼有？

dissect filter插件是使用分隔符将非结构化事件数据提取到字段中的另一种方法。

Dissect与Grok的不同之处在于它不使用正则表达式，而且速度更快。当数据可靠地重复时，Dissect工作得很好。当文本的结构因行而异时，Grok是一个更好的选择。

当行的一部分可靠地重复，但整行没有重复时，您可以对混合用例使用Dissect和Grok。分离过滤器可以解构重复的线段。Grok过滤器可以处理剩余的字段值，具有更大的regex可预测性。

三、Grok 基础

Grok的工作原理是将文本模式组合成与日志匹配的内容。

grok模式的语法: %{SYNTAX:SEMANTIC}

SYNTAX 是与文本匹配的模式的名称。例如，3.44将由 NUMBER 模式匹配，55.3.244.1将由 IP 模式匹配。

SEMANTIC （语义）是您为匹配的文本段指定的标识符（可理解为字段名称）。例如，3.44可以是一个事件的持续时间，因此可以简单地称之为 duration。此外，字符串55.3.244.1可能标识发出请求的 client。

对于上面的示例，grok过滤器如下所示：

%{NUMBER:duration} %{IP:client}

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

使用语义和语法，我们可以从一个示例日志中提取有用的字段，如下面这个虚构的http请求日志：

55.3.244.1 GET /index.html 15824 0.043

其模式可能是：

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

更现实的例子是，让我们从文件中读取以下日志：

    input {
      file {
        path => "/var/log/http.log"
      }
    }
    filter {
      grok {
        match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
      }
    }

在grok过滤器之后，事件中会有几个额外的字段：

• client: 55.3.244.1
• method: GET
• request: /index.html
• bytes: 15824
• duration: 0.043

四、正则表达式

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

五、自定义模式

有时候logstash没有你需要的模式。为此，你有几个选择。

1 首先，可以使用Oniguruma语法对捕获到的内容命名，这样可以匹配一段文本并将其保存为字段：

(?<field_name>the pattern here)

例如，日志的后缀 队列 id 的值为10或11个字符的十六进制。我可以这样轻松地捕捉到：

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

一个实际的示例如下

input { stdin {}}

filter {
   grok {
     match => {"message" => "(?<queue_id>[0-9A-F]{10,11})" }
   }
}

output { stdout {}}

在终端输入 BEF25A72965

会看到如下输出

{
      "@version" => "1",
          "host" => "es03",
      "queue_id" => "BEF25A72965",
       "message" => "BEF25A72965",
    "@timestamp" => 2021-06-07T13:54:19.661Z
}`

2 或者，您可以创建一个自定义模式文件。

创建一个名为patterns的目录，其中包含一个名为 postfix 的文件（文件名无关紧要，但要为自己命名）

在该文件中，编写所需的模式，作为模式名称、空格，然后是该模式的regexp。

例如，如上所述执行postfix queue id示例：

    # ./patterns/postfix 的内容:
    POSTFIX_QUEUEID [0-9A-F]{10,11}

然后使用这个插件中的 patterns_dir 设置告诉logstash你的自定义patterns目录在哪里。下面是一个完整的示例和一个示例日志:

    Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<[e-mail]>

    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=<[e-mail]>

六、关于匹配的细节

[root@es03 logstash]# cat testgrok-dir.conf
input { stdin {}}

filter {
   grok {
     patterns_dir => ["./patterns"]
     match => {"message" => "%{POSTFIX_QUEUEID:queue_id}:" }
   }
}

output { stdout {}}