Logstash- 数据流引擎

Logstash-数据流引擎

作者 | WenasWei

一 Logstash

Logstash 是具有实时流水线功能的开源数据收集引擎。Logstash 可以动态统一来自不同来源的数据，并将数据标准化到您选择的目标位置。清除所有数据并使其民主化，以用于各种高级下游分析和可视化用例。

1.1 Logstash 简介

Logstash 是一个数据流引擎：

• 它是用于数据物流的开源流式 ETL（Extract-Transform-Load）引擎

• 在几分钟内建立数据流管道

• 具有水平可扩展及韧性且具有自适应缓冲

• 不可知的数据源

• 具有 200 多个集成和处理器的插件生态系统

• 使用 Elastic Stack 监视和管理部署

官方介绍：Logstash is an open source data collection engine with real-time pipelining capabilities。简单来说 logstash 就是一根具备实时数据传输能力的管道，负责将数据信息从管道的输入端传输到管道的输出端；与此同时这根管道还可以让你根据自己的需求在中间加上滤网，Logstash 提供里很多功能强大的滤网以满足你的各种应用场景。

1.2 数据处理

Logstash 是一个功能强大的工具，可与各种部署集成。 它提供了大量插件，可帮助你解析，丰富，转换和缓冲来自各种来源的数据。 如果你的数据需要 Beats 中没有的其他处理，则需要将 Logstash 添加到部署中。

当下最为流行的数据源:

Logstash 可以摄入日志，文件，指标或者网路真实数据。经过 Logstash 的处理，变为可以使用的 WebApps 可以消耗的数据，也可以存储于数据中心，或变为其它的流式数据:

• Logstash 可以很方便地和 Beats 一起合作，这也是被推荐的方法

• Logstash 也可以和那些著名的云厂商的服务一起合作处理它们的数据

• 它也可以和最为同样的信息消息队列，比如 redis 或 kafka 一起协作

• Logstash 也可以使用 JDBC 来访问 RDMS 数据

• 它也可以和 IoT 设备一起处理它们的数据

• Logstash 不仅仅可以把数据传送到 Elasticsearch，而且它还可以把数据发送至很多其它的目的地，并作为它们的输入源做进一步的处理

二 Logstash 系统架构

Logstash 包含 3 个主要部分： 输入（inputs），过滤器（filters）和输出（outputs）

Logstash 的事件（logstash 将数据流中等每一条数据称之为一个 event）处理流水线有三个主要角色完成：inputs –> filters –> outputs：

• inpust：必须，负责产生事件（Inputs generate events），常用：File、syslog、redis、kakfa、beats（如：Filebeats）

• filters：可选，负责数据处理与转换（filters modify them），常用：grok、mutate、drop、clone、geoip

• outpus：必须，负责数据输出（outputs ship them elsewhere），常用：elasticsearch、file、graphite、kakfa、statsd

三 Logstash 安装

3.1 环境清单

• 操作系统：Linux #56-Ubuntu SMP Tue Jun 4 22:49:08 UTC 2019 x86_64

• Logstash 版本：logstash-6.2.4

• Jdk 版本：1.8.0_152

3.2 Linux 安装 JDK

3.2.1 解压缩并移动到指定目录（约定的目录:/usr/local）

(1)解压缩

tar -zxvf jdk-8u152-linux-x64.tar.gz

(2)创建目录

mkdir -p /usr/local/java

(3)移动安装包

mv jdk1.8.0_152/ /usr/local/java/

(4)设置所有者

chown -R root:root /usr/local/java/

3.2.2 配置环境变量

(1)配置系统环境变量

vi /etc/environment

(2)添加如下语句

PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games"export JAVA_HOME=/usr/local/java/jdk1.8.0_152export JRE_HOME=/usr/local/java/jdk1.8.0_152/jreexport CLASSPATH=$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib

(3)配置用户环境变量

nano /etc/profile

(4)添加如下语句(一定要放中间)

if [ "$PS1" ]; then  if [ "$BASH" ] && [ "$BASH" != "/bin/sh" ]; then    # The file bash.bashrc already sets the default PS1.    # PS1='\h:\w\$ '    if [ -f /etc/bash.bashrc ]; then      . /etc/bash.bashrc    fi  else    if [ "`id -u`" -eq 0 ]; then      PS1='# '    else      PS1='$ '    fi  fifi
export JAVA_HOME=/usr/local/java/jdk1.8.0_152export JRE_HOME=/usr/local/java/jdk1.8.0_152/jreexport CLASSPATH=$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/libexport PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATH:$HOME/bin
if [ -d /etc/profile.d ]; then  for i in /etc/profile.d/*.sh; do    if [ -r $i ]; then      . $i    fi  done  unset ifi

(5)使用户环境变量生效

source /etc/profile

(6)测试是否安装成功

$ java -versionjava version "1.8.0_152"Java(TM) SE Runtime Environment (build 1.8.0_152-b16)Java HotSpot(TM) 64-Bit Server VM (build 25.152-b16, mixed mode)

3.3 安装 Logstash

3.3.1 创建安装目录

$ sudo mkdir /usr/local/logstash

3.3.2 下载 Logstash 安装文件

$ wget -P /usr/local/logstash https://artifacts.elastic.co/downloads/logstash/logstash-6.2.4.tar.gz

3.3.2 解压缩安装文件

$ cd /usr/local/logstash/$ sudo tar -zxvf logstash-6.2.4.tar.gz

3.3.3 测试安装是否成功

测试: 快速启动，标准输入输出作为 input 和 output，没有 filter

$ cd logstash-6.2.4/$ ./bin/logstash -e 'input { stdin {} } output { stdout {} }'
Sending Logstash's logs to /usr/local/logstash/logstash-6.2.4/logs which is now configured via log4j2.properties[2021-05-27T00:22:28,729][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/usr/local/logstash/logstash-6.2.4/modules/fb_apache/configuration"}[2021-05-27T00:22:28,804][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/usr/local/logstash/logstash-6.2.4/modules/netflow/configuration"}[2021-05-27T00:22:29,827][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified[2021-05-27T00:22:30,979][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.2.4"}[2021-05-27T00:22:31,821][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}[2021-05-27T00:22:36,463][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}[2021-05-27T00:22:36,690][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<Thread:0x55a5abea run>"}The stdin plugin is now waiting for input:[2021-05-27T00:22:36,853][INFO ][logstash.agent           ] Pipelines running {:count=>1, :pipelines=>["main"]}
## 此时命令窗口停留在等待输入状态，键盘键入任意字符 ##
hello world
## 下方是Logstash输出到效果 ##
{    "@timestamp" => 2021-05-26T16:22:52.527Z,          "host" => "*******",       "message" => "hello world",      "@version" => "1"}

四 Logstash 参数与配置

4.1 常用启动参数

4.2 配置文件结构及语法

(1)区段

Logstash 通过{}来定义区域，区域内可以定义插件，一个区域内可以定义多个插件，如下：

input {    stdin {    }    beats {        port => 5044    }}

(2)数据类型

Logstash 仅支持少量的数据类型：

• Boolean：ssl_enable => true

• Number：port => 33

• String：name => “Hello world”

• Commonts：# this is a comment

(3)字段引用

Logstash 数据流中的数据被称之为 Event 对象，Event 以 JSON 结构构成，Event 的属性被称之为字段，如果你像在配置文件中引用这些字段，只需要把字段的名字写在中括号[]里就行了，如[type]，对于嵌套字段每层字段名称都写在[]里就可以了，比如：[tags][type]；除此之外，对于 Logstash 的 arrag 类型支持下标与倒序下表，如：[tags][type][0],[tags][type][-1]。

(4)条件判断

Logstash 支持下面的操作符：

• equality：==, !=, <, >, <=, >=

• regexp：=~, !~

• inclusion：in, not in

• boolean：and, or, nand, xor

• unary：!

例如：

if EXPRESSION {  ...} else if EXPRESSION {  ...} else {  ...}

(5)环境变量引用

Logstash 支持引用系统环境变量，环境变量不存在时可以设置默认值，例如：

export TCP_PORT=12345
input {  tcp {    port => "${TCP_PORT:54321}"  }}

4.3 常用输入插件（Input plugin）

输入插件包含有以下多种，详情查看官网文档-常用输入插件:

• elasticsearch

• exec

• file

• github

• http

• jdbc

• jms

• jmx

• kafka

• log4j

• rabbitmq

• redis

• tcp

• udp

• unix

• websocket

4.3.1 File 读取插件

文件读取插件主要用来抓取文件的变化信息，将变化信息封装成 Event 进程处理或者传递。

• 配置事例

input  file {    path => ["/var/log/*.log", "/var/log/message"]    type => "system"    start_position => "beginning"  }}

• 常用参数

4.3.2 TCP 监听插件

TCP 插件有两种工作模式，“Client”和“Server”，分别用于发送网络数据和监听网络数据。

• 配置事例

tcp {    port => 41414}

• 常用参数（空 => 同上）

4.3.3 Redis 读取插件

用于读取 Redis 中缓存的数据信息。

• 配置事例

input {  redis {    host => "127.0.0.1"    port => 6379    data_type => "list"    key => "logstash-list"  }}

• 常用参数（空 => 同上）

注意:

data_type 需要注意的是“channel”和“pattern_channel”是广播类型，相同的数据会同时发送给订阅了该 channel 的 logstash，也就是说在 logstash 集群环境下会出现数据重复，集群中的每一个节点都将收到同样的数据，但是在单节点情况下，“pattern_channel”可以同时定于满足 pattern 的多个 key

4.3.4 Kafka 读取插件

用于读取 Kafka 中推送的主题数据信息。

• 配置事例

input {  kafka {    bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"    topics_pattern  => "elk-.*"    consumer_threads => 5    decorate_events => true    codec => "json"    auto_offset_reset => "latest"    group_id => "logstash1"##logstash 集群需相同  }}

• 常用参数:

注意:

• auto_offset_reset: earliest-将偏移量自动重置为最早的偏移量;latest-自动将偏移量重置为最新偏移量;none-如果未找到消费者组的先前偏移量，则向消费者抛出异常;anything else-向消费者抛出异常。

• decorate_events: none：未添加元数据，basic：添加了记录的属性，extended：记录的属性，添加标题，false：不建议使用的别名 none，true：不建议使用的别名 basic

4.4 常用过滤插件（Filter plugin）

丰富的过滤器插件的是 logstash 威力如此强大的重要因素，过滤器插件主要处理流经当前 Logstash 的事件信息，可以添加字段、移除字段、转换字段类型，通过正则表达式切分数据等，也可以根据条件判断来进行不同的数据处理方式,详情查看官网文档-常用过滤插件

4.4.1 grok 正则捕获

grok 是 Logstash 中将非结构化数据解析成结构化数据以便于查询的最好工具，非常适合解析 syslog logs，apache log， mysql log，以及一些其他的 web log

(1)预定义表达式调用:

• Logstash 提供 120 个常用正则表达式可供安装使用，安装之后你可以通过名称调用它们，语法如下：%{SYNTAX:SEMANTIC}

• SYNTAX：表示已经安装的正则表达式的名称

• SEMANTIC：表示从 Event 中匹配到的内容的名称

例如:Event 的内容为“[debug] 127.0.0.1 - test log content”，匹配 %{IP:client}将获得“client: 127.0.0.1”的结果，前提安装了 IP 表达式；如果你在捕获数据时想进行数据类型转换可以使用 %{NUMBER:num:int}这种语法，默认情况下，所有的返回结果都是 string 类型，当前 Logstash 所支持的转换类型仅有“int”和“float”；

一个稍微完整一点的事例：

• 日志文件 http.log 内容：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}"}  }}

• 输出结果：

client: 55.3.244.1method: GETrequest: /index.htmlbytes: 15824duration: 0.043

(2)自定义表达式调用

语法：(?<field_name>the pattern here)

举例：捕获 10 或 11 和长度的十六进制 queue_id 可以使用表达式(?<queue_id>[0-9A-F]{10,11})安装自定义表达式

与预定义表达式相同，你也可以将自定义的表达式配置到 Logstash 中，然后就可以像于定义的表达式一样使用；以下是操作步骤说明：

• 1、在 Logstash 根目录下创建文件夹“patterns”，在“patterns”文件夹中创建文件“extra”（文件名称无所谓，可自己选择有意义的文件名称）；

• 2、在文件“extra”中添加表达式，格式：patternName regexp，名称与表达式之间用空格隔开即可，如下：

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

• 3、使用自定义的表达式时需要指定“patterns_dir”变量，变量内容指向表达式文件所在的目录，举例如下：

<1>日志内容

Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>

<2>Logstash 配置

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

<3>运行结果

timestamp: Jan 1 06:25:43logsource: mailserver14program: postfix/cleanuppid: 21403queue_id: BEF25A72965

(3)grok 常用配置参数（空 => 同上）

4.4.2 date 时间处理插件

该插件用于时间字段的格式转换，比如将“Apr 17 09:32:01”（MMM dd HH:mm:ss）转换为“MM-dd HH:mm:ss”。而且通常情况下，Logstash 会为自动给 Event 打上时间戳，但是这个时间戳是 Event 的处理时间（主要是 input 接收数据的时间），和日志记录时间会存在偏差（主要原因是 buffer），我们可以使用此插件用日志发生时间替换掉默认是时间戳的值。

常用配置参数（空 => 同上）

注意:

match 的格式:时间字段匹配,可自定多种格式，直到匹配到或匹配结束，格式: [ field,formats…]，如：match=>[ “logdate”,“MMM dd yyyy HH:mm:ss”,“MMM d yyyy HH:mm:ss”,“ISO8601”]

4.4.3 mutate 数据修改插件

mutate 插件是 Logstash 另一个重要插件。它提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段。

常用配置参数（空 => 同上）

4.4.4 JSON 插件

JSON 插件用于解码 JSON 格式的字符串，一般是一堆日志信息中，部分是 JSON 格式，部分不是的情况下

(1)配置事例

json {    source => ...}

• 事例配置，message 是 JSON 格式的字符串："{\"uid\":3081609001,\"type\":\"signal\"}"
  

filter {    json {        source => "message"        target => "jsoncontent"    }}

• 输出结果:

{    "@version": "1",    "@timestamp": "2014-11-18T08:11:33.000Z",    "host": "web121.mweibo.tc.sinanode.com",    "message": "{\"uid\":3081609001,\"type\":\"signal\"}",    "jsoncontent": {        "uid": 3081609001,        "type": "signal"    }}

• 如果从事例配置中删除target，输出结果如下:

{    "@version": "1",    "@timestamp": "2014-11-18T08:11:33.000Z",    "host": "web121.mweibo.tc.sinanode.com",    "message": "{\"uid\":3081609001,\"type\":\"signal\"}",    "uid": 3081609001,    "type": "signal"}

(2)常用配置参数（空 => 同上）

4.4.5 elasticsearch 查询过滤插件

用于查询 Elasticsearch 中的事件，可将查询结果应用于当前事件中

常用配置参数（空 => 同上）

4.5 常用输出插件（Output plugin）

4.5.1 ElasticSearch 输出插件

用于将事件信息写入到 Elasticsearch 中，官方推荐插件，ELK 必备插件

(1)配置事例

output {    elasticsearch {        hosts => ["127.0.0.1:9200"]        index => "filebeat-%{type}-%{+yyyy.MM.dd}"        template_overwrite => true    }}

(2)常用配置参数（空 => 同上）

4.5.2 Redis 输出插件

用于将 Event 写入 Redis 中进行缓存，通常情况下 Logstash 的 Filter 处理比较吃系统资源，复杂的 Filter 处理会非常耗时，如果 Event 产生速度比较快，可以使用 Redis 作为 buffer 使用

(1)配置事例

output {    redis {        host => "127.0.0.1"        port => 6379        data_type => "list"        key => "logstash-list"    }}

(2)常用配置参数（空 => 同上）

4.5.3 File 输出插件

用于将 Event 输出到文件内

(1)配置事例

output {    file {        path => ...        codec => line { format => "custom format: %{message}"}    }}

(2)常用配置参数（空 => 同上）

4.5.4 Kafka 输出插件

用于将 Event 输出到 Kafka 指定的 Topic 中,官网Kafka详情配置

(1)配置事例

output {    kafka {        bootstrap_servers => "localhost:9092"        topic_id => "test"        compression_type => "gzip"    }}

(2)常用配置参数（空 => 同上）

4.6 常用编码插件（Codec plugin）

4.6.1 JSON 编码插件

直接输入预定义好的 JSON 数据，这样就可以省略掉 filter/grok 配置

• 配置事例

json {}

• 常用配置参数| 参数名称 | 类型 | 默认值 | 描述信息 || ------------- | ------ | ------- | -------- || charset | string | “UTF-8” | 字符集 || enable_metric | | | || id | | | |

五 Logstash 实例

5.1 接收 Filebeat 事件，输出到 Redis

input {    beats {        port => 5044    }}
output {    redis {        host => "127.0.0.1"        port => 6379        data_type => "list"        key => "logstash-list"    }}

5.2 读取 Redis 数据，根据“type”判断，分别处理，输出到 ES

input {    redis {        host => "127.0.0.1"        port => 6379        data_type => "list"        key => "logstash-list"    }}
filter {    if [type] == "application" {        grok {            match => ["message", "(?m)-(?<systemName>.+?):(?<logTime>(?>\d\d){1,2}-(?:0?[1-9]|1[0-2])-(?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9]) (?:2[0123]|[01]?[0-9]):(?:[0-5][0-9]):(?:(?:[0-5][0-9]|60)(?:[:.,][0-9]+)?)) \[(?<level>(\b\w+\b)) *\] (?<thread>(\b\w+\b)) \((?<point>.*?)\) - (?<content>.*)"]        }        date {            match => ["logTime", "yyyy-MM-dd HH:mm:ss,SSS"]        }        json {            source => "message"        }        date {            match => ["timestamp", "yyyy-MM-dd HH:mm:ss,SSS"]        }    }    if [type] == "application_bizz" {        json {            source => "message"        }        date {            match => ["timestamp", "yyyy-MM-dd HH:mm:ss,SSS"]        }    }    mutate {        remove_field => ["@version", "beat", "logTime"]    }}
output {    stdout{    }    elasticsearch {        hosts => ["127.0.0.1:9200"]        index => "filebeat-%{type}-%{+yyyy.MM.dd}"        document_type => "%{documentType}"        template_overwrite => true    }}

六 应用场景

6.1 以 logstash 作为日志搜索器

架构：logstash 采集、处理、转发到 elasticsearch 存储，在 kibana 进行展示

特点：这种结构因为需要在各个服务器上部署 Logstash，而它比较消耗 CPU 和内存资源，所以比较适合计算资源丰富的服务器，否则容易造成服务器性能下降，甚至可能导致无法正常工作。

6.2 消息模式

消息模式：Beats 还不支持输出到消息队列（新版本除外：5.0 版本及以上），所以在消息队列前后两端只能是 Logstash 实例。logstash 从各个数据源搜集数据，不经过任何处理转换仅转发出到消息队列（kafka、redis、rabbitMQ 等），后 logstash 从消息队列取数据进行转换分析过滤，输出到 elasticsearch，并在 kibana 进行图形化展示

架构（Logstash 进行日志解析所在服务器性能各方面必须要足够好）：

模式特点：这种架构适合于日志规模比较庞大的情况。但由于 Logstash 日志解析节点和 Elasticsearch 的负荷比较重，可将他们配置为集群模式，以分担负荷。引入消息队列，均衡了网络传输，从而降低了网络闭塞，尤其是丢失数据的可能性，但依然存在 Logstash 占用系统资源过多的问题

工作流程：Filebeat 采集—> logstash 转发到 kafka—> logstash 处理从 kafka 缓存的数据进行分析—> 输出到 es—> 显示在 kibana

6.3 logstash（非 filebeat）进行文件采集，输出到 kafka 缓存，读取 kafka 数据并处理输出到文件或 es

6.4 logstash 同步 mysql 数据库数据到 es（logstash5 版本以上已集成 jdbc 插件，无需下载安装，直接使用）

七 Logstash 和 Flume 对比

首先从结构对比，我们会惊人的发现，两者是多么的相似！Logstash 的 Shipper、Broker、Indexer 分别和 Flume 的 Source、Channel、Sink 各自对应！只不过是 Logstash 集成了，Broker 可以不需要，而 Flume 需要单独配置，且缺一不可，但这再一次说明了计算机的设计思想都是通用的！只是实现方式会不同而已。

从程序员的角度来说，上文也提到过了，Flume 是真的很繁琐，你需要分别作 source、channel、sink 的手工配置，而且涉及到复杂的数据采集环境，你可能还要做多个配置，这在上面提过了，反过来说 Logstash 的配置就非常简洁清晰，三个部分的属性都定义好了，程序员自己去选择就行，就算没有，也可以自行开发插件，非常方便。当然了，Flume 的插件也很多，但 Channel 就只有内存和文件这两种（其实现在不止了，但常用的也就两种）。读者可以看得出来，两者其实配置都是非常灵活的，只不过看场景取舍罢了。

其实从作者和历史背景来看，两者最初的设计目的就不太一样。Flume 本身最初设计的目的是为了把数据传入 HDFS 中（并不是为了采集日志而设计，这和 Logstash 有根本的区别），所以理所应当侧重于数据的传输，程序员要非常清楚整个数据的路由，并且比 Logstash 还多了一个可靠性策略，上文中的 channel 就是用于持久化目的，数据除非确认传输到下一位置了，否则不会删除，这一步是通过事务来控制的，这样的设计使得可靠性非常好。相反，Logstash 则明显侧重对数据的预处理，因为日志的字段需要大量的预处理，为解析做铺垫。

回过来看我当初为什么先讲 Logstash 然后讲 Flume？这里面有几个考虑

• 其一：Logstash 其实更有点像通用的模型，所以对新人来说理解起来更简单，而 Flume 这样轻量级的线程，可能有一定的计算机编程基础理解起来更好；

• 其二：目前大部分的情况下，Logstash 用的更加多，这个数据我自己没有统计过，但是根据经验判断，Logstash 可以和 ELK 其他组件配合使用，开发、应用都会简单很多，技术成熟，使用场景广泛。相反 Flume 组件就需要和其他很多工具配合使用，场景的针对性会比较强，更不用提 Flume 的配置过于繁琐复杂了。

最后总结下来，我们可以这么理解他们的区别：

Logstash 就像是买来的台式机，主板、电源、硬盘，机箱（Logstash）把里面的东西全部装好了，你可以直接用，当然也可以自己组装修改；

Flume 就像提供给你一套完整的主板，电源、硬盘，Flume 没有打包，只是像说明书一样指导你如何组装，才能运行的起来。

参考文档:

• [1] 迷途的攻城狮.CSDN: https://blog.csdn.net/chenleiking/article/details/73563930 , 2017-06-22.

• [2] Logstash 官网: https://www.elastic.co/cn/logstash