在上一篇博文中，我们介绍了使用 Metricbeat 对 Elastic Stack 进行监测这一新方法。使用 Metricbeat 从外部收集有关 Elastic Stack 产品的监测信息的话，您可以提高这些产品的监测服务的可靠性。这种方法还能让您灵活地制定将监测数据发送到 Elasticsearch 监测集群的方法。在本篇博文中，我们将会就第二方面进行更深入的探讨，向您展示用户可以如何借助 Logstash 或 Kafka 将 Metricbeat 所收集的监测数据发送到监测集群中。所以，如果已经针对业务数据在 Metricbeat 配置中使用 logstash 或 kafka 输出，您仍可以继续使用这些输出来发送您 Elastic Stack 的监测数据。
现在我们从上一篇博文结束之处继续说起。上篇博文介绍了下面的架构，该架构会借助 Metricbeat 对 Elastic Stack 产品进行监测。
请注意，每个 Metricbeat 实例监测的都是 Elastic Stack 产品的一个实例或节点。因此，我们必须启用正确的 Metricbeat 模块（*-xpack 配置变体）。举个例子，如要监测一个 Logstash 节点，必须启用 logstash-xpack 模块。
在该架构中，每个 Metricbeat 实例都会将数据发送到监测集群。这也意味着，Metricbeat 主机到监测集群主机之间必须有网络连接。
然而，您有时可能希望减少 Elasticsearch 的接入点数量。您可能更希望将 Metricbeat 发出的所有堆栈监测数据输送到一个 Logstash 实例中，然后再由该 Logstash 实例将数据转发到监测集群。在本篇博文中，我们会看一下如何使用 Metricbeat 实施此类架构以完成堆栈监测。
将 Logstash 添加到您的堆栈监测数据流中
首先我们要设置一个 Logstash 管道，因为我们要通过此管道从 Metricbeat 接收堆栈监测数据并将这些数据转发到监测集群。此管道如下所示，我们还会详细讲解其组成。
input {
beats {
port => 5044
}
}
filter {
# 固定内容，用以兼容各种 Beats 版本
mutate {
rename => { "[@metadata][id]" => "[@metadata][_id]" }
}
}
output {
if [@metadata][index] =~ /^.monitoring-*/ {
# 将堆栈监测数据发送到负责监测的 Elasticsearch 集群
if [@metadata][_id] {
elasticsearch {
index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
document_id => "%{[@metadata][_id]}"
hosts => ["https://node1:9200"]
}
} else {
elasticsearch{
index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
hosts => ["https://node1:9200"]
}
}
} else {
# 发送非堆栈监测类数据
}
}
从整体上来说，这个管道：
- 会使用
beats输入插件来读取 Metricbeat 所发送的堆栈监测数据。 - 会使用
elasticsearch输出插件来将堆栈监测数据发送到监测集群。
请注意管道中 output 部分的 if-else 语句梯子。顶级的 if-else 会允许您将用于堆栈监测的数据（会索引到 monitoring-* 中）与同一 Metricbeat 实例可能收集的其他数据（例如，如果您启用了 system 模块）区别开来。
在针对堆栈监测数据的 if 子句内，还有一个嵌套的 if-else 语句。这一结构能够确保在将数据索引到监测集群时，来自 Metricbeat 的堆栈监测数据事件上的任何 ID（如果已设置）均会传送到 _id 字段。这一点对于正确地索引 Elasticsearch 分片监测数据尤为重要。如果没有这一结构，针对 Elasticsearch 的堆栈监测 UI 将会错误地显示分片数量一直不断增长！
配置 Metricbeat 以将数据发送到 Logstash
Logstash 管道设置完毕之后，您还需要配置 Metricbeat 实例以将相应数据发送到 Logstash 主机，而非直接发送到监测集群。
output.logstash: hosts: [ "logstash_hostname:5044" ]
这一设置还有一个变体，即可以在 Metricbeat 和 Logstash 之间引入一个 Kafka。在这种情况下，Logstash 管道和上面的代码基本一样，只是您需要使用 kafka 输入插件，而非使用 beats 插件。同时，您相应地需要配置自己的 Metricbeat 实例以将数据发送到 Kafka 集群，而非 Logstash。
总结
希望本文对如何通过 Logstash（或 Kafka）将 Elastic Stack 监测数据从 Metricbeat 发送到 Elasticsearch 为您提供了具体的实施指南。我还希望本文向您清晰演示了您可以如何灵活地使用 Metricbeat 从外部收集有关 Elastic Stack 产品的监测数据。
如果您在进行此设置的过程中遇到任何问题或难题，敬请在 discuss.elastic.co 上发帖讨论。而且，如果您想感受一下“堆栈监测”的使用体验，请访问我们的交互式演示并亲自试用。