ELK实践（一）：基础入门 (4)

这说明直接使用filebeat，也是可以发送数据到ES的，为什么还用logstash呢？原因是filebeat采集的是原始日志内容，发送到ES的也是原始内容，如果不需要处理的话，使用filebeat也行。

我们可以看到返回的文档json里有一个字段 message ，这个是日志原始内容。filebeat还默认加了一些字段：

@timestamp 时间

beat filebeat相关信息，数组类型

input_type 日志类型，一般是log

message 日志原文

offset 当前采集的日志的偏移量

source 日志所在文件

tags 自定义标签，数组类型

type document_type字段定义的内容

kibana里查看

打开kibana web地址：:5601，依次打开：Management
-> Kibana -> Index Patterns ,选择Create Index Pattern：

a. Index pattern 输入：test-filebeat ；

b. Time Filter field name 选择 @timestamp。

c. 点击Create。

然后打开Discover，选择 test-filebeat 就能看到日志数据了。

这时候我们去访问下自己配置了nginx日志的应用，这里能实时更新，再也不用去命令行使用tail查看了。

filebeat内容发送到logstash

接下来，我们将日志使用filebeat发送到logstash，然后通过logstash处理后发送到ElasticSearch。

首先我们要修改上一节里filebeat的配置：

vim beats/filebeat/filebeat.test.yml

改为：

filebeat.prospectors: - type: log paths: - /tmp/test1.log tags: ["test1"] document_type: test1 - type: log paths: - /tmp/test2.log tags: ["test2"] document_type: test2 output.logstash: hosts: ["127.0.0.1:5046"] #output.elasticsearch: # hosts: ["127.0.0.1:9200"] # index: "test-filebeat"

我们把output.elasticsearch注释了，新增了output.logstash。

然后新增logstash配置：

vim logstash/config/conf.d/filebeat.test.conf input { beats { port => 5046 } } output { elasticsearch { hosts => ["127.0.0.1:9200"] index => "test-filebeat-%{type}" } stdout { codec => rubydebug } }

这里的type变量就是filebeat里面的document_type。端口指定为5046（自定义即可），和filebeat里面配置的一致。logstash可以有多个子配置，所以也就能配置多个端口。此时，logstash是作为服务端运行的，filebeat是客户端。

接下来我们启动logstash和filebeat：

./logstash/bin/logstash &

&表示后台运行：

./beats/filebeat/filebeat -c beats/filebeat/filebeat.test.yml

我们新开终端往日志里加点新内容：

echo "new msg" >> /tmp/test2.log echo "new msg3" >> /tmp/test2.log

稍等几秒钟，我们可以查看ES里的数据：

curl :9200/test-filebeat-test2/_search?q=*&sort=@timestamp:desc { "took": 3, "timed_out": false, "_shards": { "total": 5, "successful": 5, "skipped": 0, "failed": 0 }, "hits": { "total": 2, "max_score": null, "hits": [ { "_index": "test-filebeat-test2", "_type": "test2", "_id": "AWYMF-PeJzfnbYlB_DSo", "_score": null, "_source": { "@timestamp": "2018-09-24T14:59:38.188Z", "offset": 49, "@version": "1", "input_type": "log", "beat": { "name": "2106567e5bce", "hostname": "2106567e5bce", "version": "5.6.2" }, "host": "2106567e5bce", "source": "/tmp/test2.log", "message": "new msg3", "type": "test2", "tags": [ "test2", "beats_input_codec_plain_applied", "_grokparsefailure" ] }, "sort": [ 1537801178188 ] }, { "_index": "test-filebeat-test2", "_type": "test2", "_id": "AWYMF-PeJzfnbYlB_DSn", "_score": null, "_source": { "@timestamp": "2018-09-24T14:59:38.186Z", "offset": 40, "@version": "1", "input_type": "log", "beat": { "name": "2106567e5bce", "hostname": "2106567e5bce", "version": "5.6.2" }, "host": "2106567e5bce", "source": "/tmp/test2.log", "message": "new msg", "type": "test2", "tags": [ "test2", "beats_input_codec_plain_applied", "_grokparsefailure" ] }, "sort": [ 1537801178186 ] } ] } }

本节里我们虽然用到了logstash，但没有发挥它的强大处理功能。仅仅是演示了将日志使用filebeat发送到logstash，然后通过logstash处理后发送到ElasticSearch。处理功能后续再做讲解。