◽️ ☕️ 📫 私たちはELKとExchangeと友達です。パート2 🤹🏿 🛏️ 🆙

ExchangeとELKの友達を作る方法についての私の話を続けます（ここから始めてください）。この組み合わせでは、非常に多くのログをためらうことなく処理できることを思い出してください。今回は、ExchangeをLogstashおよびKibanaコンポーネントと連携させる方法について説明します。

ELKスタックのLogstashは、ログをインテリジェントに処理し、ドキュメントの形式でElasticに配置する準備をするために使用されます。これに基づいて、Kibanaでさまざまな視覚化を構築すると便利です。

インストール

2つの段階で構成されます。

OpenJDKパッケージのインストールと構成。
Logstashパッケージのインストールと構成。

OpenJDK

パッケージのインストールと構成OpenJDKパッケージをダウンロードして、特定のディレクトリに解凍する必要があります。次に、このディレクトリへのパスを、Windowsオペレーティングシステムの$ env：Path変数と$ env：JAVA_HOME変数に入力する必要があります

。Javaバージョンを確認します。

PS C:\> java -version
openjdk version "13.0.1" 2019-10-15
OpenJDK Runtime Environment (build 13.0.1+9)
OpenJDK 64-Bit Server VM (build 13.0.1+9, mixed mode, sharing)

Logstashパッケージのインストールと構成

ここから Logstashディストリビューションを含むアーカイブファイルをダウンロードします。アーカイブはディスクのルートに解凍する必要があります。フォルダーにC:\Program Files解凍しないでください。Logstashは正常に起動することを拒否します。次にjvm.options、JavaプロセスにRAMを割り当てる役割を担うファイルに変更を加える必要があります。サーバーのRAMの半分を指定することをお勧めします。彼が16GBのRAMを搭載している場合、デフォルトのキーは次のとおりです。

-Xms1g
-Xmx1g

次のように置き換える必要があります。

-Xms8g
-Xmx8g

行をコメントアウトすることもお勧めし -XX:+UseConcMarkSweepGCます。詳しくはこちらをご覧ください。次のステップは、logstash.confファイルにデフォルト構成を作成することです。

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

この構成では、Logstashはコンソールからデータを読み取り、それを空のフィルターに通して、コンソールに書き戻します。この構成を適用すると、Logstashの機能がテストされます。これを行うには、インタラクティブに起動しましょう。

PS C:\...\bin> .\logstash.bat -f .\logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Logstashはポート9600で正常に起動しまし

た。インストールの最後のステップは、LogstashをWindowsサービスとして起動することです。これは、たとえば、NSSMパッケージを使用して実行できます。

PS C:\...\bin> .\nssm.exe install logstash
Service "logstash" installed successfully!

耐障害性

永続キューメカニズムは、ソースサーバーからの送信中のログの安全性を保証します。

それはどのように機能しますか

ログ処理中のキューのレイアウト：入力→キュー→フィルター+出力。

入力プラグインは、ログソースからデータを受信し、それをキューに書き込み、データの受信確認をソースに送信します。

キューからのメッセージはLogstashによって処理され、フィルターと出力プラグインを通過します。Logstashは、ログ送信の出力から確認を受け取ると、処理されたログをキューから削除します。Logstashが停止した場合、未処理のメッセージと送信の確認を受信していないメッセージはすべてキューに残り、Logstashは次回の起動時にそれらを処理し続けます。

セットアップ

ファイル内のキーによって調整されます C:\Logstash\config\logstash.yml:

queue.type：（可能な値はpersistedとmemory (default))。
path.queue：(デフォルトでC：\ Logstash \ queueに保存されているキューファイルを含むフォルダーへのパス）。
queue.page_capacity：(キューの最大ページサイズ。デフォルトは64MBです）。
queue.drain：( true / false-Logstashをオフにする前にキュー処理の停止を有効/無効にします。サーバーのシャットダウン速度に直接影響するため、オンにすることはお勧めしません）。
queue.max_events：(キュー内のイベントの最大数、デフォルト-0（無制限））。
queue.max_bytes：(バイト単位の最大キューサイズ、デフォルトは1024mb（1gb））。

queue.max_eventsおよびが構成queue.max_bytesされている場合、これらの設定のいずれかの値に達すると、メッセージはキューで受信されなくなります。永続キューの詳細については、こちらをご覧ください。

キューの設定を担当するlogstash.ymlの部分の例：

queue.type: persisted
queue.max_bytes: 10gb

セットアップ

Logstash構成は通常、受信ログの処理のさまざまなフェーズ（受信（入力セクション）、解析（フィルターセクション）、Elasticへの送信（出力セクション））を担当する3つの部分で構成されます。以下では、それぞれについて詳しく見ていきます。

入力

生のログを含む着信ストリームは、ファイルビートエージェントから受信されます。入力セクションで指定するのはこのプラグインです。

input {
  beats {
    port => 5044
  }
}

この設定後、Logstashはポート5044でリッスンを開始し、ログを受信すると、フィルターセクションの設定に従ってログを処理します。必要に応じて、ファイルビットからログを受信するためのチャネルをSSLでラップできます。ビートプラグインの設定について詳しくは、こちらをご覧ください。

フィルタ

Exchangeが処理のために生成するすべての興味深いテキストログは、ログファイル自体に記述されたフィールドを持つcsv形式です。：CSVレコードを解析するために、Logstashは私達に3つのプラグイン提供解剖、CSVやGROKを。1つ目は最速ですが、最も単純なログしか解析できません。

たとえば、次のレコードを2つに分割します（フィールド内にコンマが存在するため）。これにより、ログが誤って解析されます。

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

IISなどのログを解析するときに使用できます。この場合、フィルターセクションは次のようになります。

filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
    }
  }
}

Logstash構成では条件付きステートメントが許可されているため、filebeatタグでマークされたdissectプラグインにのみログを送信できますIIS。プラグイン内で、フィールド値をそれらの名前と照合しmessage、エントリを含む元のフィールドをログから削除します。たとえば、ログを収集するアプリケーションの名前を含む任意のフィールドを追加できます。

ログを追跡する場合は、csvプラグインを使用することをお勧めします。これにより、複雑なフィールドを正しく処理できます。

filter {
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
}

プラグイン内で、フィールド値をその名前と照合し、エントリを含む元のフィールドmessage（tenant-idおよびとフィールドschema-version）をログから削除します。たとえば、ログを収集するアプリケーションの名前を含む任意のフィールドを追加できます。

フィルタリング段階の終了時に、最初の近似でドキュメントを取得し、Kibanaでレンダリングする準備をします。私たちは以下を見逃します：

数値フィールドはテキストとして認識され、操作を実行できなくなります。つまり、time-takenIISログフィールド、recipient-countおよびtotal-bites追跡フィールドとログです。
標準のドキュメントタイムスタンプには、サーバー側の記録時間ではなく、ログ処理時間が含まれます。
フィールドrecipient-addressは単一の構造のように見えますが、文字の受信者を数えることによる分析はできません。

今こそ、ログ処理プロセスに魔法を加えるときです。

数値フィールドの変換

dissectプラグインには、convert_datatypeテキストフィールドをデジタル形式に変換するために使用できるオプションがあります。たとえば、次のようになります。

dissect {
  …
  convert_datatype => { "time-taken" => "int" }
  …
}

この方法は、フィールドに文字列が確実に含まれる場合にのみ適していることを覚えておく価値があります。このオプションは、フィールドからのnull値を処理せず、例外にスローされます。

ログの変換を追跡する場合、フィールドrecipient-countとtotal-bitesは空白になる可能性があるため、同様の方法は避ける必要があります。これらのフィールドを変換するには、mutateプラグインを使用することをお勧めします。

mutate {
  convert => [ "total-bytes", "integer" ]
  convert => [ "recipient-count", "integer" ]
}

receive_addressを個々の受信者に分割する

このタスクは、mutateプラグインを使用して解決することもできます。

mutate {
  split => ["recipient_address", ";"]
}

タイムスタンプの変更

ログを追跡する場合、タスクは日付プラグインによって非常に簡単に解決されます。これはtimestamp、フィールドから必要な形式でフィールドに日付と時刻を書き込むのに役立ちますdate-time。

date {
  match => [ "date-time", "ISO8601" ]
  timezone => "Europe/Moscow"
  remove_field => [ "date-time" ]
}

IISログのケースでは、フィールドのデータを結合する必要がありますdateとtime、のmutateプラグインを使用して、私たちに必要な時間帯を登録し、このタイムスタンプを置くtimestamp日付プラグインを使用しました：

mutate { 
  add_field => { "data-time" => "%{date} %{time}" }
  remove_field => [ "date", "time" ]
}
date { 
  match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
  timezone => "UTC"
  remove_field => [ "data-time" ]
}

出力

出力セクションは、処理されたログをログレシーバーに送信するために使用されます。Elasticに直接送信する場合は、elasticsearchプラグインが使用されます。これは、生成されたドキュメントを送信するためのサーバーアドレスとインデックス名のテンプレートを指定します。

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

最終構成

最終的な構成は次のようになります。

input {
  beats {
    port => 5044
  }
}
 
filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
      convert_datatype => { "time-taken" => "int" }
    }
    mutate { 
      add_field => { "data-time" => "%{date} %{time}" }
      remove_field => [ "date", "time" ]
    }
    date { 
      match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
      timezone => "UTC"
      remove_field => [ "data-time" ]
    }
  }
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
    mutate {
      convert => [ "total-bytes", "integer" ]
      convert => [ "recipient-count", "integer" ]
      split => ["recipient_address", ";"]
    }
    date {
      match => [ "date-time", "ISO8601" ]
      timezone => "Europe/Moscow"
      remove_field => [ "date-time" ]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

便利なリンク：

私たちはELKとExchangeと友達です。パート2

インストール

Logstashパッケージのインストールと構成

耐障害性

それはどのように機能しますか

セットアップ

セットアップ

入力

フィルタ

数値フィールドの変換

receive_addressを個々の受信者に分割する

タイムスタンプの変更

出力

最終構成

More articles: