導入

別のシステムを導入する際、多数の異なるログを処理する必要性に直面しました。 ELK がツールとして選択されました。この記事では、このスタックのセットアップに関する私たちの経験について説明します。

すべての機能を説明するという目標は設定しませんが、特に実際的な問題の解決に集中したいと考えています。これは、かなり大量のドキュメントと既製のイメージがあるにもかかわらず、少なくとも私たちが見つけた落とし穴がかなりたくさんあるという事実によるものです。

docker-compose を介してスタックをデプロイしました。さらに、よく書かれた docker-compose.yml があったため、ほぼ問題なくスタックを引き上げることができました。そして、私たちにはすでに勝利が近づいているように見えました。今度は私たちのニーズに合わせて少し調整して、それで終わりです。

残念ながら、アプリケーションからログを受信して処理するようにシステムを構成する試みは、すぐには成功しませんでした。したがって、各コンポーネントを個別に検討してから、それらの接続に戻ることが価値があると判断しました。

そこで、logstash から始めました。

環境、デプロイメント、コンテナー内での Logstash の実行

デプロイには docker-compose を使用します。ここで説明する実験は MacOS と Ubuntu 18.0.4 で実行されました。

オリジナルの docker-compose.yml に登録された logstash イメージは docker.elastic.co/logstash/logstash:6.3.2 です。

実験に使用させていただきます。

logstash を実行するために別の docker-compose.yml を作成しました。もちろん、コマンドラインからイメージを起動することも可能でしたが、私たちは docker-compose からすべてを実行するという特定の問題を解決していました。

設定ファイルについて簡単に説明します

説明から分かるように、logstash は XNUMX つのチャネルに対して実行できます (その場合は *.conf ファイルを渡す必要があります)、または複数のチャネルに対して実行できます (この場合は Pipelines.yml ファイルを渡す必要があります)。、各チャンネルのファイル .conf にリンクします。
私たちはXNUMX番目の道を選びました。私たちには、それがより普遍的で拡張性があるように見えました。そこで、pipelines.yml を作成し、各チャネルの .conf ファイルを配置する Pipelines ディレクトリを作成しました。

コンテナー内には、別の構成ファイル logstash.yml があります。うちは触らずにそのまま使ってます。

したがって、ディレクトリ構造は次のようになります。

入力データを受信するには、今のところポート 5046 上の TCP であると想定し、出力には stdout を使用します。

ここでは、初回起動時の簡単な構成を示します。なぜなら、最初のタスクは起動することだからです。

これが docker-compose.yml です。

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

ここで何が見えますか？

ネットワークとボリュームは元の docker-compose.yml (スタック全体が起動されるもの) から取得されており、ここでの全体像に大きな影響を与えるものではないと思います。
docker.elastic.co/logstash/logstash:6.3.2 イメージから XNUMX つの logstash サービスを作成し、logstash_one_channel という名前を付けます。
コンテナ内のポート 5046 を同じ内部ポートに転送します。
パイプ設定ファイル ./config/pipelines.yml をコンテナ内のファイル /usr/share/logstash/config/pipelines.yml にマップします。ここで、logstash がそれを取得し、念のため読み取り専用にします。
チャネル設定を含むファイルがある ./config/pipelines ディレクトリを /usr/share/logstash/config/pipelines ディレクトリにマップし、読み取り専用にします。

Pipelines.yml ファイル

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

ここでは、HABR 識別子を持つ XNUMX つのチャネルとその構成ファイルへのパスについて説明します。

そして最後にファイル「./config/pipelines/habr_pipeline.conf」

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

今はその説明には立ち入らないで、実行してみましょう。

docker-compose up

何が見えますか？

コンテナが起動しました。その動作を確認できます。

echo '13123123123123123123123213123213' | nc localhost 5046

そして、コンテナーコンソールに応答が表示されます。

しかし同時に、次のこともわかります。

ログスタッシュワンチャンネル | [2019-04-29T11:28:59,790][エラー][logstash.licensechecker.licensereader] ライセンスサーバーからライセンス情報を取得できません {:message=>「Elasticsearch に到達できません: [http://elasticsearch:9200/][Manticore]」 ::ResolutionFailure] elasticsearch", ...

ログスタッシュワンチャンネル | [2019-04-29T11:28:59,894][情報][logstash.pipeline ] パイプラインが正常に開始されました {:pipeline_id=>".monitoring-logstash", :thread=>"# "}

ログスタッシュワンチャンネル | [2019-04-29T11:28:59,988][情報][logstash.agent ] パイプラインが実行中です {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[ ]}
ログスタッシュワンチャンネル | [2019-04-29T11:29:00,015][エラー][logstash.inputs.metrics] X-Pack は Logstash にインストールされていますが、Elasticsearch にはインストールされていません。モニタリング機能を使用するには、Elasticsearch に X-Pack をインストールしてください。他の機能も利用できる場合があります。
ログスタッシュワンチャンネル | [2019-04-29T11:29:00,526][情報][logstash.agent ] Logstash API エンドポイント {:port=>9600} が正常に開始されました
ログスタッシュワンチャンネル | [2019-04-29T11:29:04,478][情報][logstash.outputs.elasticsearch] ヘルスチェックを実行して Elasticsearch 接続が機能しているかどうかを確認します {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
ログスタッシュワンチャンネル | [2019-04-29T11:29:04,487][警告][logstash.outputs.elasticsearch] 停止した ES インスタンスへの接続を復活させようとしましたが、エラーが発生しました。 {:url=>”エラスティックサーチ:9200/"、:error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError、:error=>"Elasticsearch に到達できません: [http://elasticsearch:9200/][Manticore::ResolutionFailure]エラスティックサーチ"}
ログスタッシュワンチャンネル | [2019-04-29T11:29:04,704][情報][logstash.licensechecker.licensereader] ヘルスチェックを実行して Elasticsearch 接続が機能しているかどうかを確認します {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
ログスタッシュワンチャンネル | [2019-04-29T11:29:04,710][警告][logstash.licensechecker.licensereader] 停止した ES インスタンスへの接続を復活させようとしましたが、エラーが発生しました。 {:url=>”エラスティックサーチ:9200/"、:error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError、:error=>"Elasticsearch に到達できません: [http://elasticsearch:9200/][Manticore::ResolutionFailure]エラスティックサーチ"}

そして私たちのログは常に増加しています。

ここでは、パイプラインが正常に起動したことを示すメッセージを緑色で強調表示し、エラーメッセージを赤色で強調表示し、接続の試行に関するメッセージを黄色で強調表示しています。エラスティックサーチ：9200。
これは、イメージに含まれる logstash.conf に elasticsearch の可用性のチェックが含まれているために発生します。結局のところ、logstash は Elk スタックの一部として動作することを前提としていますが、それを分離しました。

働くことは可能ですが、不便です。

解決策は、XPACK_MONITORING_ENABLED 環境変数を使用してこのチェックを無効にすることです。

docker-compose.yml を変更して、再度実行してみましょう。

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

さて、すべて順調です。コンテナは実験の準備が整いました。

次のコンソールにもう一度入力します。

echo '13123123123123123123123213123213' | nc localhost 5046

そして、以下を参照してください。

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

XNUMX つのチャネル内で作業する

そこで私たちは立ち上げました。これで、実際に時間をかけて logstash 自体を構成できるようになります。ここでは、pipelines.yml ファイルには触れずに、XNUMX つのチャネルを操作することで何が得られるかを見てみましょう。

チャネル構成ファイルを使用する一般原則は、公式マニュアルに詳しく説明されていると言わざるを得ません。ここで
ロシア語で読みたい場合は、これを使用しました記事(ただし、クエリ構文は古いため、これを考慮する必要があります)。

入力セクションから順に行ってみましょう。私たちはすでに TCP に関する取り組みを見てきました。他に何が面白いでしょうか?

ハートビートを使用してメッセージをテストする

自動テストメッセージを生成する非常に興味深い機会があります。
これを行うには、入力セクションでハートビーンプラグインを有効にする必要があります。

input {
  heartbeat {
    message => "HeartBeat!"
   }
  }

電源を入れて、XNUMX分にXNUMX回受信を開始します

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

もっと頻繁に受信したい場合は、interval パラメータを追加する必要があります。
このようにして、10 秒ごとにメッセージを受信します。

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

ファイルからデータを取得する

ファイルモードについても検討することにしました。ファイルが正常に動作する場合は、少なくともローカルで使用する場合にはエージェントは必要ない可能性があります。

説明によると、動作モードは tail -f と同様である必要があります。つまり、新しい行を読み取るか、オプションでファイル全体を読み取ります。

それで、私たちが取得したいものは次のとおりです。

XNUMX つのログファイルに追加された行を受信したいと考えています。
複数のログファイルに書き込まれたデータを、何をどこから受信したかを分離できるように受信したいと考えています。
logstash が再起動されたときに、このデータを再度受信しないようにしたいと考えています。
logstash がオフになっていて、データが引き続きファイルに書き込まれている場合、logstash を実行するとこのデータが受信されることを確認したいと考えています。

実験を行うために、docker-compose.yml に別の行を追加して、ファイルを置いたディレクトリを開きます。

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

そして、habr_pipeline.conf の入力セクションを変更します。

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

はじめましょう：

docker-compose up

ログファイルを作成して書き込むには、次のコマンドを使用します。

 echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

はい、うまくいきます！

同時に、パスフィールドが自動的に追加されたことがわかります。これは、将来的には、それによってレコードをフィルターできるようになることを意味します。

もう一度試してみましょう:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

次に、別のファイルに移動します。

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

素晴らしい！ファイルが選択され、パスが正しく指定されており、すべて問題ありません。

logstash を停止して、再度開始します。待とう。沈黙。それらの。これらの記録を再度受け取ることはありません。

そして今度は最も大胆な実験です。

logstash をインストールして実行します。

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

logstash を再度実行して、次を確認します。

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

万歳！すべてが拾われました。

しかし、次の点について警告しなければなりません。 logstash コンテナーが削除された場合 (docker stop logstash_one_channel && docker rm logstash_one_channel)、何も取得されません。ファイルが読み取られた位置はコンテナ内に保存されます。最初から実行する場合は、新しい行のみを受け入れます。

既存のファイルの読み取り

初めて logstash を起動するが、すでにログがあり、それらを処理したいとします。
上記で使用した入力セクションを使用して logstash を実行しても、何も取得されません。 logstash では新しい行のみが処理されます。

既存のファイルから行を取得するには、入力セクションに追加の行を追加する必要があります。

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

さらに、微妙な違いがあります。これは、logstash がまだ認識していない新しいファイルにのみ影響します。 logstash の視野内にすでに存在する同じファイルについては、そのサイズがすでに記憶されているため、新しいエントリのみがそのファイルに取り込まれるようになります。

ここで停止して、入力セクションを勉強しましょう。オプションはまだたくさんありますが、現時点ではさらなる実験を行うにはこれで十分です。

ルーティングとデータ変換

次の問題を解決してみましょう。XNUMX つのチャネルからのメッセージがあり、その一部は情報メッセージであり、一部はエラーメッセージであるとします。タグによって異なります。情報となるものもあれば、エラーとなるものもあります。

出口で彼らを分離する必要があります。それらの。あるチャネルには情報メッセージを書き込み、別のチャネルにはエラーメッセージを書き込みます。

これを行うには、入力セクションからフィルターと出力に移動します。

フィルターセクションを使用して、受信メッセージを解析し、そこからハッシュ (キーと値のペア) を取得します。これは既に処理できます。条件に応じて分解します。そして出力セクションではメッセージを選択し、それぞれを独自のチャネルに送信します。

grok を使用したメッセージの解析

テキスト文字列を解析し、そこから一連のフィールドを取得するために、フィルターセクションに特別なプラグイン - grok があります。

ここでそれについて詳しく説明するという目標は設定しません（これについては、公式ドキュメント）、私の簡単な例を示します。

これを行うには、入力文字列の形式を決定する必要があります。私は次のようなものを持っています：

1 情報メッセージ1
2 エラーメッセージ2

それらの。最初に識別子、次に INFO/ERROR、そしてスペースを含まない単語が続きます。
それほど難しいことではありませんが、動作原理を理解するだけで十分です。

したがって、grok プラグインのフィルターセクションで、文字列を解析するためのパターンを定義する必要があります。

次のようになります。

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  }

基本的には正規表現です。 INT、LOGLEVEL、WORD などの既製のパターンが使用されます。それらの説明と他のパターンは、ここで見つけることができます。ここで

このフィルタを通過すると、文字列は XNUMX つのフィールド (message_id、message_type、message_text) のハッシュに変わります。

これらは出力セクションに表示されます。

if コマンドを使用してメッセージを出力セクションにルーティングする

出力セクションでは、メッセージを XNUMX つのストリームに分割する予定でした。一部の iNFO はコンソールに出力され、エラーが発生した場合はファイルに出力されます。

これらのメッセージをどのように分離すればよいでしょうか? 問題の状況はすでに解決策を示唆しています。結局のところ、専用の message_type フィールドがすでにあり、このフィールドは INFO と ERROR の XNUMX つの値のみを取ることができます。これに基づいて、if ステートメントを使用して選択を行います。

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

フィールドと演算子の操作の説明は、このセクションにあります。公式マニュアル.

さて、実際の結論自体について。

コンソール出力、ここではすべてが明確です - stdout {}

ただし、ファイルへの出力 - これらすべてをコンテナから実行していることに注意してください。結果を書き込むファイルに外部からアクセスできるようにするには、docker-compose.yml でこのディレクトリを開く必要があります。

合計：

ファイルの出力セクションは次のようになります。

 output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "custom format: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

docker-compose.yml では、出力用に別のボリュームを追加します。

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

それを立ち上げて試してみると、XNUMX つの流れに分かれていることがわかります。

出所： habr.com

ELKの実用化。 logstash のセットアップ

導入

環境、デプロイメント、コンテナー内での Logstash の実行

設定ファイルについて簡単に説明します

XNUMX つのチャネル内で作業する

ハートビートを使用してメッセージをテストする

ファイルからデータを取得する

既存のファイルの読み取り

ルーティングとデータ変換

grok を使用したメッセージの解析

if コマンドを使用してメッセージを出力セクションにルーティングする

コメントを追加します返信をキャンセル

ELKの実用化。 logstash のセットアップ

導入

環境、デプロイメント、コンテナー内での Logstash の実行

設定ファイルについて簡単に説明します

XNUMX つのチャネル内で作業する

ハートビートを使用してメッセージをテストする

ファイルからデータを取得する

既存のファイルの読み取り

ルーティングとデータ変換

grok を使用したメッセージの解析

if コマンドを使用してメッセージを出力セクションにルーティングする

コメントを追加します 返信をキャンセル

コメントを追加します返信をキャンセル