MetricbeatとLogstashやKafkaを併用してElastic Stackを監視する

前回の記事で、Metricbeatを使ってElastic Stackを監視する新しいメソッドの登場についてお伝えしました。Metricbeatを使って外部からElastic Stackプロダクトに関する監視情報を収集すると、プロダクト監視の信頼性が高まります。また、監視データをElasticsearch監視クラスターに送る方法も、よりフレキシブルになります。今回の記事では後者に注目し、Metricbeatで収集した監視データをLogstashやKafkaを経由して監視クラスターに転送する手法を説明しながら詳しく解説してゆきます。すでにビジネスデータを活用する目的でMetricbeatの設定にlogstash、またはkafkaアウトプットを使用している場合、それをElastic Stack監視データの転送にそのまま使うこともできます。

まず前回の記事のおさらいです。前回、Elastic StackプロダクトをMetricbeatで監視するために、下図のアーキテクチャーを構築しました。

ご注目いただきたいのは、各Metricbeatインスタンスで特定のElastic Stackプロダクトのインスタンス、またはノードを監視している点です。このように監視させるには、適切なMetricbeatモジュール（*-xpack configのバリアント）を有効化する必要があります。たとえばLogstashノードを監視するには、logstash-xpackモジュールを有効化しなければなりません。

このアーキテクチャーのすべてのMetricbeatインスタンスは、監視クラスターにデータをシッピングしています。ということは、Metricbeatホストと監視クラスターホストの間に、ネットワーク接続が必要です。

しかしElasticsearchにデータ投入されるポイントの数は、最小限が望ましいというケースは少なくありません。MetricbeatインスタンスからLogstashインスタンスに向かうすべてのスタック監視トラフィックをまとめてから監視クラスターに転送できればより理想的、というケースもあるでしょう。この記事では、そうしたアーキテクチャーを実装してMetricbeatを使い、スタックを監視するアプローチを検討します。

スタック監視データフローにLogstashを追加する

はじめに、Metricbeatからスタック監視データを受け取り、それを監視クラスターに転送するLogstashパイプラインを立ち上げます。このパイプラインは以下の通りです。各部については後で説明します。

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アウトプットプラグインを使用して、スタック監視データを監視クラスターに送る

パイプラインのアウトプットにあるif-else文のラダーに注目してください。最上位のif-elseは、同じMetricbeatインスタンスで収集される可能性のあるスタック監視用ではないデータ（例：systemモジュールを有効化している場合など）を分離し、スタック監視のデータを.monitoring-*インデックス内にインデックスします。

スタック監視データ用のif構文内には、ネストされたif-else文があります。この構造により、データを監視クラスターにインデックスする際、Metricbeatからくるスタック監視データイベント上に設定されたあらゆるIDが_idフィールドに確実に渡されます。この構造は、特にElasticsearchシャード監視データを適切にインデックスする上で不可欠です。この構造がない場合、ElasticsearchのStack Monitoring UIが、時間とともに増え続ける誤ったシャード数を表示してしまいます。

Metricbeatを設定してLogstashにシッピングさせる

Logstashパイプラインの立ち上げが終わったら、データを直接監視クラスターに送るのではなく、LogstashホストにシッピングするようMetricbeatインスタンスを設定します。

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

設定を少し変えて、MetricbeatとLogstashの間にKafkaを入れるセットアップを行うこともできます。その場合もLogstashパイプラインは上述のものとほぼ同じですが、beatsインプットプラグインに代えて、kafkaインプットプラグインを使用する点が異なります。同様に、Metricbeatインスタンスも、データをLogstashではなく、Kafkaクラスターに送るよう設定します。

まとめ

今回の記事が、Elastic Stack監視データをMetricbeatから、Logstash（またはKafka）を経由して転送する実装の手順を具体的に理解する手助けとなれば幸いです。また、Metricbeatで外部からElastic Stackプロダクトの監視データを収集する手法で、柔軟性が向上するメリットを実感していただければと思います。 

ご質問や、セットアップに関してお困りのことがおありの場合は、discuss.elastic.coにお気軽にご相談ください。スタック監視を導入するメリットについて詳しく知りたい、という方は、インタラクティブデモンストレーションをお試しください。