Logstash - 简要使用

2020/01/31 共 1847 字 5 分钟阅读
DOCKER • ELK • OS • PROGRAMMING elastic

Table of Content

AI 摘要: 本文主要介绍Docker运行和Logstash的筛选器插件，以及Logstash的输出功能。Docker是一个容器化平台，可以快速部署和运行应用程序。筛选器插件可以解析日志数据并提取补充信息，如使用geoip插件获取地理位置信息。Logstash可以通过lumberjack协议将事件发送到Logstash，并进行进一步的处理和路由。

1. Docker 运行

1
2
3
4
5
6
// 安装
$ docker pull logstash:7.5.2
7.5.2: Pulling from library/logstash

// 运行()
$ docker run --rm -u root --privileged --name tmp-logstash -dit logstash:7.5.2

2. 基本概念

Logstash 管道具有两个必需元素（输入和输出）和一个可选元素（过滤器）。输入插件使用源中的数据，过滤器插件根据的指定修改数据，输出插件将数据写入目标。

1
2
cd logstash-7.5.2
bin/logstash -e 'input { stdin { } } output { stdout {} }'

3. Filebeat 作为输入源

Logstash pipeline: 它通常具有一个或多个输入，过滤器和输出插件

在创建 Logstash 管道之前，需要配置 Filebeat 以将日志行发送到 Logstash。 Filebeat 客户端是一种轻量级，资源友好的工具，可从服务器上的文件收集日志并将这些日志转发到 Logstash 实例进行处理。 Filebeat 专为可靠性和低延迟而设计。 Filebeat 在主机上的资源占用很少，Beats 输入插件使 Logstash 实例上的资源需求最小化。

3.1. filebeat.yml

安装Filebeat之后，需要对其进行配置。打开位于 Filebeat 安装目录中的 filebeat.yml 文件，并将内容替换为以下几行。确保路径指向先前下载的示例 Apache 日志文件 logstash-tutorial.log：

Filebeat 将尝试在端口5044上进行连接，filebeat.yml 配置：

1
2
3
4
5
6
filebeat.inputs:
- type: log
  paths:
    - /path/to/file/logstash-tutorial.log
output.logstash:
  hosts: ["localhost:5044"]

At the data source machine, run Filebeat with the following command:

1
sudo ./filebeat -e -c filebeat.yml -d "publish"

3.2. filebeat.yml

filebeat 只能定义一个输出源

4. first-pipeline.conf - Logstash 的管道配置支持 Beats 输入插件

接下来，创建一个使用 Beats 输入插件从 Beats 接收事件的 Logstash 配置管道。first-pipeline.conf配置管道的框架input{} filter{} output{}：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
    # The # character at the beginning of a line indicates a comment. Use
    # comments to describe your configuration.
    input {
        beats {
            port => "5044"
        }
    }
    # The filter part of this file is commented out to indicate that it is
    # optional.
    # filter {
    #
    # }
    output {
        stdout { codec => rubydebug }
    }

验证配置：

1
2
3
4
5
// --config.test_and_exit选项解析的配置文件并报告任何错误。
bin/logstash -f first-pipeline.conf --config.test_and_exit

// --config.reload.automatic选项启用了自动重新加载配置，因此不必在每次修改配置文件时都停止并重新启动Logstash。
bin/logstash -f first-pipeline.conf --config.reload.automatic

调试结果：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
{
    "@timestamp" => 2017-11-09T01:44:20.071Z,
        "offset" => 325,
      "@version" => "1",
          "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
          "host" => "My-MacBook-Pro.local",
    "prospector" => {
        "type" => "log"
    },
    "input" => {
        "type" => "log"
    },
        "source" => "/path/to/file/logstash-tutorial.log",
       "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ]
}
...

4.1. pipelines.yml

当 Logstash 启动时，可能会看到有关 Logstash 的一个或多个警告消息，而忽略了pipelines.yml文件，可以放心地忽略此警告。pipelines.yml文件用于在单个 Logstash 实例中运行多个管道。

5. Grok 作为过滤插件

工作管道从Filebeat中读取日志行，但是日志消息的格式并不理想，想解析日志消息以从日志中创建特定的命名字段。为此，将使用grok过滤器插件，以将非结构化日志数据解析为结构化和可查询的内容。

grok 过滤器插件是 Logstash 默认提供的几个插件之一，由于 grok 过滤器插件会在传入的日志数据中查找模式，因此配置插件需要做出有关如何识别用例感兴趣的模式的决策。

Web 服务器日志示例中的代表行如下所示，该行开头的 IP 地址很容易识别，括号中的时间戳也很容易识别。

要解析数据，可以使用％{COMBINEDAPACHELOG} grok 模式，该模式使用以下模式从 Apache 日志中构造行：：

1
2
3
83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

Apache log using the following schema:

Information	Field Name
IP Address	clientip
User ID	ident
User Authentication	auth
timestamp	timestamp
HTTP Verb	verb
Request body	request
HTTP Version	httpversion
HTTP Status Code	response
Bytes served	bytes
Referrer URL	referrer
User agent	agent

5.1. Grok Debugger 调试 ¹

Grok 是一种模式匹配语法，可用于解析任意文本并对其进行结构化。 Grok 非常适合解析 syslog，apache 和其他 Web 服务器日志，mysql 日志，以及通常用于人类使用的任何日志格式。

Elastic Stack 预制了 120 多个可重复使用的 grok 模式，有关模式的完整列表，请参见摄取节点 grok 模式和Logstash grok模式。

先在 Kibana Grok 调试器(该工具在 Kibana 中自动启用)中构建和调试 grok 模式，然后再在数据处理管道中使用它们：

DevTools > Grok Debugger.
Sample Data 输入要解析的消息，比如55.3.244.1 GET /index.html 15824 0.043
Grok Pattern 输入grok pattern，比如%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
Simulate点击模仿，检测输出的结构化数据

5.2. 自定义 Grok 模式

Sample Data ，比如Jan 1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>
Grok Pattern模式，比如%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{MSG:syslog_message}

Custom Pattern模式，比如：

1
2
POSTFIX_QUEUEID [0-9A-F]{10,11}
MSG message-id=<%{GREEDYDATA}>

Simulate点击模仿，检测输出的结构化数据。

如果发生错误，可以继续遍历自定义模式，直到输出与期望的事件匹配为止

5.3. first-pipeline.conf - 支持 Grok 模式

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
input {
    beats {
        port => "5044"
    }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    stdout { codec => rubydebug }
}

由于启用了自动重新加载配置，因此无需重启 Logstash 即可获取更改。

若确实需要强制 Filebeat从头读取日志文件，按Ctrl + C关闭 Filebeat，然后删除 Filebeat 注册表文件并重启 Filebeat：

1
2
3
4
5
// 删除Filebeat注册表文件，强制Filebeat从头读取其收集的所有文件
sudo rm data/registry

// 重启
sudo ./filebeat -e -c filebeat.yml -d "publish"

Logstash 应用 grok 模式后，事件将具有以下 JSON 表示形式（事件包括原始消息，但日志消息也细分为特定字段）：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
{
    "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
         "offset" => 325,
           "auth" => "-",
          "ident" => "-",
           "verb" => "GET",
     "prospector" => {
        "type" => "log"
    },
     "input" => {
        "type" => "log"
    },
         "source" => "/path/to/file/logstash-tutorial.log",
        "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
     "@timestamp" => 2017-11-09T02:51:12.416Z,
       "response" => "200",
          "bytes" => "203023",
       "clientip" => "83.149.9.216",
       "@version" => "1",
           "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
           "host" => "My-MacBook-Pro.local",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000"
}

6. 筛选器插件

// todo

除解析日志数据以进行更好的搜索外，筛选器插件还可以从现有数据中获取补充信息。例如，geoip 插件查找 IP 地址，从地址中获取地理位置信息，然后将该位置信息添加到日志中。

7. Logstash

Logstash 输出通过使用通过 TCP 运行的 lumberjack 协议将事件直接发送到 Logstash。 Logstash 允许对生成的事件进行其他处理和路由。

https://www.elastic.co/guide/en/kibana/7.5/xpack-grokdebugger.html ↩︎