Elastic：通过 Logstash 或 Kafka 使用 Metricbeat 监控 Elastic Stack

Posted 2021-07-06 Elastic 中国社区官方博客

在之前的一篇博文中，我们介绍了一种使用 Metricbeat 监控 Elastic Stack 的新方法。 使用 Metricbeat 从外部收集有关 Elastic Stack 产品的监控信息，提高了监控这些产品的可靠性。 它还为如何将监控数据路由到 Elasticsearch 监控集群提供了灵活性。 在这篇博文中，我们通过展示用户如何将 Metricbeat 收集的监控数据通过 Logstash 或 Kafka 路由到监控集群来更深入地研究第二个方面。 因此，如果你已经在 Metricbeat 配置中将 logstash 或 kafka 输出用于业务数据，你也可以继续使用这些输出来路由 Elastic Stack 监控数据。

让我们从上一篇博客文章停止的地方开始。 它引入了以下架构，用于使用 Metricbeat 监控 Elastic Stack 产品。

请注意，每个 Metricbeat 实例都在监控 Elastic Stack 产品的一个实例或节点。 为此，必须启用正确的 Metricbeat 模块（*-xpack 配置变体）。 例如，要监控 Logstash 节点，必须启用 logstash-xpack 模块。如果你想了解更多，请参阅我之前的文章：

• Elastic：监控 Elasticsearch 及 Kibana

• Elastic：监控 Beats 及 APM Server

• Logstash：使用 Metricbeat 监控 Logstash

在这种架构中，每个 Metricbeat 实例都会将数据传送到一个监控集群。 这意味着需要有从 Metricbeat 主机到监控集群主机的网络连接。

然而，有时可能需要尽量减少进入 Elasticsearch 的入口点数。 将 Metricbeat 实例发出的所有堆栈监控流量汇集到 Logstash 实例中，然后让它将数据转发到监控集群可能更理想。 在这篇博文中，我们将看看如何使用 Metricbeat 实现这种用于堆栈监控的架构。

将 Logstash 添加到你的堆栈监控数据流

首先，我们将设置一个 Logstash 管道，该管道从 Metricbeat 接收堆栈监控数据并将其转发到监控集群。 该管道如下所示，并对其部分进行了进一步说明。

input {
  beats {
    port => 5044
  }
}

filter {
  # Boilerplate for compatibility across Beats versions
  mutate {
    rename => { "[@metadata][id]" => "[@metadata][_id]" }
  }
}

output {
  stdout {
  }

  if [@metadata][index] =~ /^.monitoring-*/ {
    # route stack monitoring data to monitoring Elasticsearch cluster
    if [@metadata][_id] {
      elasticsearch {
        index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
        document_id => "%{[@metadata][_id]}"
        hosts => ["http://localhost:9200"]
        user => "elastic"
        password => "password"
      }
    } else {
      elasticsearch{
        index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
        hosts => ["http://localhost:9200"]
        user => "elastic"
        password => "password"
      }
    }
  } else {
    # route non-stack monitoring data
  }  
}

简单地说，这个管道：

• 使用 beats 输入插件读取 Metricbeat 发送的堆栈监控数据。
• 使用 elasticsearch 输出插件将堆栈监控数据发送到监控集群。

请注意管道输出部分中的 if-else 语句。顶级 if-else 允许你将用于堆栈监控的数据（将被索引到 .monitoring-* 索引中）与可能由相同 Metricbeat 实例收集的其他数据分开，例如如果你启用了系统模块。

在堆栈监控数据的 if 子句中，有一个嵌套的 if-else 语句。这种结构确保在将数据索引到监控集群时，来自 Metricbeat 的堆栈监控数据事件上的任何 ID（如果设置）都会传递到 _id 字段。这对于正确索引 Elasticsearch 分片监控数据至关重要。如果没有这种结构，Elasticsearch 的 Stack Monitoring UI 将错误地显示随着时间的推移不断增加的分片数量！

配置 Metricbeat 以发送到 Logstash

设置 Logstash 管道后，您需要配置 Metricbeat 实例以将其数据发送到 Logstash 主机，而不是直接发送到监控集群。

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

此设置的一个变体可能是在 Metricbeat 和 Logstash 之间引入 Kafka。 在这种情况下，Logstash 管道看起来与上面几乎相同，除了你将使用 kafka 输入插件而不是 beats 插件。 相应地，你可以将 Metricbeat 实例配置为将它们的数据发送到 Kafka 集群而不是 Logstash。

经过上面的配置，我们可以监控 Elasticsearch 集群：

在上面我们启动 Metericbeat 来监控 Elasticsearch。我们可以查看到 .monitoring 开头的索引：

总结

希望这篇文章为你提供了一个具体的实现，用于通过 Logstash（或 Kafka）将你的 Elastic Stack 监控数据从 Metricbeat 路由到 Elasticsearch。 我还希望它清楚地展示了使用 Metricbeat 从外部收集来自 Elastic Stack 产品的监控数据的灵活性。

如果你对此设置有任何疑问或遇到任何问题，请随时将它们发布到 discussion.elastic.co。 如果你想了解 Stack Monitoring 带来的好处，请前往我们的交互式演示并尝试一下。