在我之前的文章“Elastic：运用 Elastic Stack 分析 Spring Boot 微服务日志 (一）”，我详细讲述了如何使用如下架构来进行对 Spring Boot 应用进行日志分析：

细心的开发者可能已经发现：我们的 Logstash 的配置文件每次只配置一个 log 文件，也就是说每个 Logstash 的实例只能处理一个文件。这在很多的情况下不是最优的。比如说，我们有10个 log 文件，那么我们可能需要10个 Logstash 的实例来处理这个。由于 Logstash 的使用成本是非常高的。Logstash 的运行需要很多的资源来完成，通常会甚至需要专门的服务器来完成这样的工作。那么我们该如何完善我们的设计呢？

我们可以使用 Filebeat 来配合完成我们的工作。关于 Filebeat 如何配合 Logstash，并把数据发送 Elasticsearch 中，我们可以参阅我之前的另外一篇文章“Beats：通过Filebeat把日志传入到Elasticsearch”来进行详述。参阅那篇文章，我们可以使用如下的架构来完善我们的设计：

在上面的架构中，我们使用 Filebeat 来分别读取每个 log 文件。由于 Filebeat 是一个轻量级的数据摄入器，它并不消耗很多的资源。它可以和我们的每个微服务的应用部署在同一个服务器中，并轻松地把数据传入到 Logstash 中。在这里，我们可以使用 Logstash 超强的数据处理，转换，丰富等能力对数据进一步地加工，并最终导入到 Elasticsearch 之中。最后，由 Kibana 来对数据进行分析我展现。 在这个架构中，我们所有的 Filebeat 共享一个 Logstash 的实例。
 

安装

Filebeat

相比之前的架构，我们新增加了一个 Filebeat 的部分。我们打开 Kibana 界面：

点击 Add log data 按钮：

点击上面的 System logs 链接：

我们首先现在 Filebeat 运行的平台，并选择相应的指令进行相应的安装。在我们今天的练习中，我们并不需要启动任何的模块，这是因为我们这个 log 不是属于任何的模块。是我们自己定义格式的日志。我们有不需要配置任何的 Elasticsearch 及 Kibana，这是因为我们会把数据传入到 Logstash 中。在上面的指令中，我们只需要选择指令进行安装。

等我们安装好 Filebeat，我们给 Filebeat 配置相应的配置文件：

filebeat_logstash.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /Users/liuxg/tmp/spring-boot-elastic.log
  multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
  multiline.negate: true
  multiline.match: after

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

 在这里，我们在 paths 里定义了日志的路径。同时我们使用了 mulitiline 来把多行日志变为一个文档而不是每一行都变成一个文档。如果大家对这个还是不很了解的话，请参阅我之前的文章“Beats：使用Filebeat传送多行日志”。

我们可以使用如下的指令来测试我们的配置文件：

/filebeat -c filebeat_logstash.yml test config

$ ./filebeat -c filebeat_logstash.yml test config
Config OK

上面显示我们的配置文件没有任何的错误。

我们可以使用如下的指令来测试我们的 output:

./filebeat -c filebeat_logstash.yml test output

$ ./filebeat -c filebeat_logstash.yml test output
logstash: localhost:5044...
  connection...
    parse host... OK
    dns lookup... OK
    addresses: ::1, 127.0.0.1
    dial up... ERROR dial tcp 127.0.0.1:5044: connect: connection refused

上面显示我们的 output 是有问题的。这个原因是很简单的，这是因为我们的 localhost:5044 服务没有启动。

启动 Logstash

针对我们的最新的架构，我们需要重新设计我们的 Logstash 的配置文件：

logstash.conf

# Read input from filebeat by listening to port 5044 on which filebeat will send the data
input {
    beats {
       type => "test"
       port => "5044"
    }
}
 
filter {
  #If log line contains tab character followed by 'at' then we will tag that entry as stacktrace
  if [message] =~ "\tat" {
    grok {
      match => ["message", "^(\tat)"]
      add_tag => ["stacktrace"]
    }
  }
 
}
 
output {
   
  stdout {
    codec => rubydebug
  }
 
  # Sending properly parsed log events to elasticsearch
  elasticsearch {
    hosts => ["localhost:9200"]
  }
}

在上面，我们主要修改了 Input 部分：

input {
    beats {
       type => "test"
       port => "5044"
    }
}

其它的和之前的配置是一样的。

为了更清楚地表明 Elasticsearch 的数据是来自这次实验的数据，我们在 Kibana 中对之前的 Logstash 的索引的数据进行删除：

DELETE logstash-*

然后，我们启动 Logstash：

sudo ./bin/logstash -f logstash.conf

上面显示我们的 Logstash 已经成功运行了。

运行 Filebeat

接下来，我们开始运行 Filebeat。首先我们再次测试一下我们的 output:

$ ./filebeat -c filebeat_logstash.yml test output
logstash: localhost:5044...
  connection...
    parse host... OK
    dns lookup... OK
    addresses: ::1, 127.0.0.1
    dial up... OK
  TLS... WARN secure connection disabled
  talk to server... OK

这次显示我们的测试是成功的。我们接下来使用如下的指令来运行 Filebeat:

./filebeat -e -c filebeat_logstash.yml 

上面显示我们的 Filebeat 连接到 Logstash是成功的。我们可以到 Logstash 的 console 来查看一下：

上面显示每一条被处理的文档。

在 Kibana 中展示数据

在上面，我们看出数据已经被 Logstash 处理了。在 Kibana 中，打入如下的命令：

GET _cat/indices

我们可以看到一个新的以 Logstash 为开头的日志索引已经生成了（还记得在上面我们已经把所有 logstash-* 的索引都删除的指令吗）。按照在上一个教程 “Elastic：运用 Elastic Stack 分析 Spring Boot 微服务日志 (一）”中所描述的，我们创建一个index pattern，并展示数据：

我们可以使用之前的微服务接口生成更多的日志：

我们再次冲查看最新的文档：

再查看具有 stacktrace 的文档：