在今天的讲座里，我们来讲述一下如何把 Filebeat 里的数据传入到 Logstash 之中。在做这个练习之前，我想信大家已经安装我之前的文章 “如何安装Elastic栈中的Logstash” 把 Logstash 安装好。

之前，我们介绍了一这样的一幅图：

如上图所示，我们可以直接把 beats 里的数据直接传入到 Elasticsearch，也可以直接接入到 Logstash 之中。

数据采集流程：

• Beats 采集（Filebeat/Metricbeat）–>Elasticsearch–>Kibana
• Beats 采集（Filebeat/Metricbeat）–>Logstash–>Elasticsearch–>Kibana

既然 beats 可以直接把数据直接写入到 Elasticsearch 之中，为什么我们还需要 Logstash 呢？这主要的原因是因为 Logstash 有丰富的 filter 供我们使用，可以帮我们加工数据，并最终把我们的数据转为我们喜欢的格式。Beats 在 ELK 中是一个非常轻量级的应用。是用 go 语言写的。它不需要有很强大的运行环境来运行。它通常运行于客户端中。

建立样本文件

为了方便我们的说明，我们先创建一个简单的 sample.log 文件。这个文件可以位于你电脑的任何位置。为了方便，我们把这个文件存于 Filebeat 的安装的根目录下。它的内容如下：

sample.log

2019-09-09T13:00:00Z Whose woods these are I think I know.
2019-09-09T14:00:00Z His house is in the village, though;
2019-09-09T15:00:00Z He will not see me stopping here
2019-09-09T16:00:00Z To watch his woods fill up with snow.
2019-09-09T17:00:00Z My little horse must think it queer
2019-09-09T18:00:00Z To stop without a farmhouse near
2019-09-09T19:00:00Z Between the woods and frozen lake
2019-09-09T20:00:00Z The darkest evening of the year.
2019-09-09T21:00:00Z He gives his harness bells a shake
2019-09-09T22:00:00Z To ask if there is some mistake.
2019-09-09T23:00:00Z The only other sound's the sweep
2019-09-09T00:00:00Z Of easy wind and downy flake.
2019-09-09T01:00:00Z The woods are lovely, dark, and deep,
2019-09-09T02:00:00Z But I have promises to keep,
2019-09-09T03:00:00Z And miles to go before I sleep,
2019-09-09T04:00:00Z And miles to go before I sleep

这是一首英文诗。

配置 Logstash

我们可以在你电脑的任何一个位置建立一个 Logstash 的配置文件，比如，我们建立一个叫做 logstash-beats.conf 的文件：

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    port => 5044
  }
}

filter {
	grok {
		match => [
			"message", "%{TIMESTAMP_ISO8601:timestamp_string}%{SPACE}%{GREEDYDATA:line}"
		]
	}

	date  {
		match => ["timestamp_string", "ISO8601"]
	}

	mutate {
		remove_field => [message, timestamp_string]
	}	
}


output {	
	elasticsearch {
		hosts =>  ["http://localhost:9200"]
		user => elastic
		password => "your password"
	}


	stdout {
		codec => rubydebug
	}
}

在这里，我想来说明一下：

• 这里的 input 使用了 beat，并监听端口5044。很多 beats 都使用一个端口，并向这个端口发送数据。在接下来的 Filebeat，我们可以看到如何配置这个端口，并发送log数据到这个端口
• 在 filter 的这个部分，我们首先使用 grok。它的基本语法是 %{SYNTAX:SEMANTIC}。SYNTAX 是与您的文本匹配的模式的名称。这个有点类似于正则表达式。例如，3.44 将与 NUMBER 模式匹配，55.3.244.1 将与 IP 模式匹配。SEMANTIC 是您为匹配的文本提供的标识符。针对我们的例子：

2019-09-09T13:00:00Z Whose woods these are I think I know.

      2019-09-09T13:00:00Z：使用 TIMESTAMP_ISO8601 匹配，并形成一个叫做 timestamp_string 的字符串。SPACE 用来匹配时间后面的空格，而剩下的部分由 GREEDYDATA 进行匹配，并存于 line 这个字符串中。比如针对我们的例子 line 就是 “Whose woods these are I think I know.” 字符串

• date 这个 filter 可以帮我们把一个字符串变成一个 ISO8601 的时间，并最终存于一个叫做 @timestamp 的字段中。如果没有这个 filter，我们可以看到最终的 @timestamp 是采用当前的运行时间，而不是在 log 里的时间
• 我们使用 mutate 过滤器把之前的 message 和 timestamp_string 都删除掉

在接下来的实验中，我们可以把上面的这些 filter 注释掉，你可以看看在输出中有什么变化。这里的 filter 是安装顺序从上到下执行的。在实际的练习中，我们可以把上面的 mutate 和 date 进行调换，我们应该看到不同的结果。

• 在 output 中，我们把数据输出到 Elasticsearch，并同时输出到 stdout，这样我们也可以在 terminal 的输出中看到输出的信息。这个也很方便我们做测试。如果大家已经安装了 x-pack 安全，可以在 Elasticsearch 中加入用户名及密码：

output {	
	elasticsearch {
		hosts =>  ["localhost:9200"]
		index => "logstash-poem"
		document_type => "_doc"
		user => elastic
		password => "YourPassword"
	 }


	stdout {
		codec => rubydebug
	}
}

更多关于 filter 的介绍，可以参阅我们的官方网址 “Filter plugins”。

在 Logstash 中，按照顺序执行的处理方式被叫做一个 pipeline。一个 pipeline 含有一个按照顺序执行的逻辑数据流。pipeline 从 input 里获取数据，并传送给一个队列，并接着传入到一些 worker 去处理：

配置好 Logstash 后，我们可以使用如下的命令来执行：

$ ./bin/logstash -f ~/data/logstash-beats.conf 

在这里，我把 logstash-beats.conf 放置于我 home 目录下的一个叫做 data 的子目录。大家需要根据自己的文件位置进行修改。这样我们的 Logstash 就已经运行起来了。

安装 Filebeat

deb:

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.3.1-amd64.deb
sudo dpkg -i filebeat-7.3.1-amd64.deb

rpm:
 

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.3.1-x86_64.rpm
sudo rpm -vi filebeat-7.3.1-x86_64.rpm

mac:

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.3.1-darwin-x86_64.tar.gz
tar xzvf filebeat-7.3.1-darwin-x86_64.tar.gz

brew:

brew tap elastic/tap
brew install elastic/tap/filebeat-full

linux:

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.3.1-linux-x86_64.tar.gz
tar xzvf filebeat-7.3.1-linux-x86_64.tar.gz

这样我们的 Filebeat 就安装好了。请注意：由于 ELK 迭代比较快，我们可以把上面的版本 7.3.1 替换成我们需要的版本即可。

配置 Filebeat

在 Filebeat 的根目录下，我们可以发现一个叫做 filebeat.yml 的文件。它里面有些描述可以帮我们理解如何配置这个文件。为了方便，我们直接把这个文件改成如下的内容：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - ./sample.log

output.logstash:
  hosts: ["localhost:5044"]

这里需要注意的是之前有的文章里第一行写的是 filebeat.prospectors。经过测试在新的版本里不再适用。

我们在这里可以看到 Filebeat 会帮我们把 sample.log 文件读入，并写入到 Logstash 中。它使用了一个 5044 端口地址。这个地址在我们如下的 Longstash 配置中，我们将会使用到。我们可以通过如下的命令来运行 filebeat:

$ ./filebeat 

在默认的情况下，filebeat 会自动寻找定义在 filebeat.yml 文件里的配置。如果我们的配置文件是另外的名字，我们可以通过如下的命令来执行 filebeat:

$ ./filebeat -c YourYmlFile.yml

Filebeat 的 registry 文件存储 Filebeat 用于跟踪上次读取位置的状态和位置信息。 

• data/registry 针对 .tar.gz and .tgz 归档文件安装
• /var/lib/filebeat/registry 针对 DEB 及 RPM 安装包
• c:\ProgramData\filebeat\registry 针对 Windows zip 文件

如果我们想重新运行数据一遍，我们可以直接到相应的目录下删除那个叫做 registry 的目录即可。针对 .tar.gz 的安装包来说，我们可以直接删除这个文件：

localhost:data liuxg$ pwd
/Users/liuxg/elastic/filebeat-7.3.0-darwin-x86_64/data
localhost:data liuxg$ rm -rf registry/

那么重新运行上面的 ./filebeat 命令即可。它将会重新把数据从头再进行处理一遍。这对于我们的调试来说是非常用用的。

在输出确认请求之前，Filebeat 不会考虑发新的送日志行。 由于将行传送到配置的输出的状态保留在注册表文件中，因此您可以安全地假定事件将至少一次传送到配置的输出，并且不会丢失任何数据

当我们运行完我们的 filebeat 应用，我们可以在 Logstash 的屏幕上看到：

这里除了我们之前看到的那些数据之外，我们也可以看到其他很多不同的字段。这些都是filebeat自动帮我们加入进去的。因为我们加入了 date 过滤器，我们可以看 @timestamp 也取自之前的 timestamp_string 的值。在实际的调试中，我们可以把 filter 一个个地去掉或加上，这样在我们的屏幕上，我们可以看到不同的输出。这个留给大家来做自己的实验。

我们可以回到 Kibana，通过查看一个叫做 “logstash_poem” 的索引：

至此，我们已经完成了通过 filebeat 的方法把一个 log 文件传输到 Elasticsearch 之中。