Logstash Mutate フィルターを使用したログフィールドの解析 - ES

ログデータが単一のパイプ区切りの文字列として受信される場合、logstash-filter-mutate プラグインを使用して、その文字列を個別の名前付きフィールドに分割してから Elasticsearch に書き込むことができます。このトピックでは、パイプラインの設定方法、各 mutate 操作によるデータ変換の仕組み、および Kibana で結果を検証する方法について説明します。

前提条件

開始する前に、以下が準備できていることを確認してください。

Alibaba Cloud Elasticsearch クラスター (この例では V7.10)。作成手順については、「Alibaba Cloud Elasticsearch クラスターを作成する」をご参照ください。
Elasticsearch クラスターで自動インデックス作成が有効になっていること。手順については、「YML ファイルの設定」をご参照ください。

ここでは、自動インデックス作成はテスト目的でのみ使用します。本番環境では、データを転送する前に手動でインデックスを作成してください。自動作成されたインデックスはスキーマと一致しない可能性があります。詳細については、「クイックスタート」をご参照ください。
Elasticsearch クラスターと同じ仮想プライベートクラウド (VPC) 内の Alibaba Cloud Logstash クラスター。詳細については、「Alibaba Cloud Logstash クラスターの作成」をご参照ください。
準備されたテストデータ。この例では、Beats によって収集された LogMessage データレコードを使用します。フィールドは縦棒 (|) で区切られています。
```
LogMessage: |1390000****|jop|byORP|2022-04-18T14:18:16.633|/log/cms/send|200|pluginNums=0,pluginStatus=0||||||
```
パイプラインはこのレコードを分割し、値を順番に mobile、appName、type、timestamp、status、code、component、cid、serviceId、serviceName、serviceType、param に割り当てます。Beats を使用したログの収集方法については、「ECS インスタンスのログを収集する」をご参照ください。

背景情報

logstash-filter-mutate プラグインを使用すると、ログイベントのフィールドを変換 (分割、名前の変更、削除、置き換え、変更) できます。すべてのオプションのリストについては、「Mutate フィルタープラグインリファレンス」をご参照ください。

すべてのフィルタープラグインには、共通の設定オプションがあります。

オプション	データ型
add_field	hash
add_tag	array
enable_metric	boolean
id	string
periodic_flush	boolean
remove_field	array
remove_tag	array

これらのオプションは任意です。詳細については、「共通オプション」をご参照ください。

パイプラインの作成

Alibaba Cloud Elasticsearch コンソールの [Logstash クラスター] ページに移動します。
上部のナビゲーションバーで、クラスターが存在するリージョンを選択します。[Logstash クラスター] ページで、対象のクラスターを見つけてその ID をクリックします。
左側のナビゲーションウィンドウで、[パイプライン] をクリックします。
[パイプライン] ページで、[パイプラインの作成] をクリックします。

[作成] ウィザードで、パイプライン ID を入力し、パイプラインを設定します。Logstash パイプラインには 3 つのセクションがあります。以下の設定を使用してください。

Input — データソース (この例では Beats) を定義します。
Filter — インデックス作成前にイベントを変換します。
出力 — 処理済みデータをその送信先 (Elasticsearch) に送信します

gsub — 正規表現 \| を使用して各縦棒 (|) に一致させ、それを縦棒とスペース (| ) に置き換えます。バックスラッシュは、正規表現で特別な意味を持つパイプ文字をエスケープします。gsub の後：
```
LogMessage: | 1390000****| jop| byORP| 2022-04-18T14:18:16.633| /log/cms/send| 200| pluginNums=0,pluginStatus=0| | | | | |
```
split — メッセージフィールドを縦棒 (|) で分割し、値の配列を生成します。

add_field — 各配列要素をインデックスによって名前付きフィールドに割り当てます。このステップの後、各値には先頭にスペースが含まれます。

"mobile":" 1390000****",
"appName":" jop",
"type":" byORP",
"timestamp":" 2022-04-18T14:18:16.633",
"status":" /log/cms/sen",
"code":" 200",
"component":" pluginNums=0,pluginStatus=0",
"cid":" ",
"serviceId":" ",
"serviceName":" ",
"serviceType":" ",
"param":" "

strip (2 番目の mutate ブロック) — 各フィールド値から先頭のスペースを削除します。

重要

input.beats.port の値は 8000～9000 の範囲である必要があります。

警告

パイプラインを保存してデプロイすると、Logstash クラスターの再起動がトリガーされます。続行する前に、再起動が実行中のワークロードに影響を与えないことを確認してください。

プレースホルダー	説明
`<yourIndexName>`	送信先インデックスの名前
`es-cn-7mz2mu1zp0006****.elasticsearch.aliyuncs.com:9200`	ご利用の Elasticsearch クラスターのホストアドレス
`<yourPassword>`	`elastic` アカウントのパスワード

input {
    beats {
        port => 8001
    }
}
filter {
    mutate {
        gsub => ["message","\|","| "]
        split => ["message","|"]
        add_field => {
            "mobile" => "%{[message][1]}"
            "appName" => "%{[message][2]}"
            "type" => "%{[message][3]}"
            "timestamp" => "%{[message][4]}"
            "status" => "%{[message][5]}"
            "code" => "%{[message][6]}"
            "component" => "%{[message][7]}"
            "cid" => "%{[message][8]}"
            "serviceId" => "%{[message][9]}"
            "serviceName" => "%{[message][10]}"
            "serviceType" => "%{[message][11]}"
            "param" => "%{[message][12]}"
        }
    }
    mutate {
        strip => ["mobile","appName","type","timestamp","status","code","component","cid","serviceId","serviceName","serviceType","param"]
    }
}
output {
    elasticsearch {
        index => "<yourIndexName>"
        hosts => ["es-cn-7mz2mu1zp0006****.elasticsearch.aliyuncs.com:9200"]
        user => "elastic"
        password => "<yourPassword>"
    }
}

output セクションの次のプレースホルダーを置き換えます。 フィルターがデータを変換する方法 フィルターは 4 つの操作をシーケンスで適用します。パイプライン構成の詳細については、「構成ファイルを使用してパイプラインを管理する」および「Logstash 構成ファイル」をご参照ください。

[保存] または [保存してデプロイ] をクリックします。
- [保存] — パイプライン設定を適用せずに保存します。[パイプライン] ページでパイプラインを見つけ、[操作] 列の [今すぐデプロイ] をクリックして設定を適用し、クラスターを再起動します。
- [保存してデプロイ] — 設定を保存し、ただちに Logstash クラスターを再起動して適用します。

結果の検証

Elasticsearch クラスターの Kibana コンソールにログインします。詳細については、「Kibana コンソールにログインする」をご参照ください。

この例では Elasticsearch V7.10.0 を使用します。手順はバージョンによって異なる場合があります。実際のコンソールをご参照ください。
右上隅にある [開発ツール] をクリックします。

[コンソール] タブで、次のクエリを実行してインデックスからドキュメントを取得します：

GET <yourIndexName>/_search
{
  "query": {
    "match_all": {}
  }
}

パイプライン設定の index パラメーターと同じインデックス名を使用します。成功した結果は次のようになります。

{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "<yourIndexName>",
        "_type" : "_doc",
        "_id" : "Lb1UWoAB-6Zo6en4luDi",
        "_score" : 1.0,
        "_source" : {
          "mobile" : "1390000****",
          "appName" : "jop",
          "type" : "byORP",
          "timestamp" : "2022-04-18T14:18:16.633",
          "status" : "/log/cms/sen",
          "code" : "200",
          "component" : "pluginNums=0,pluginStatus=0",
          "cid" : "",
          "serviceId" : "",
          "serviceName" : "",
          "serviceType" : "",
          "param" : ""
        }
      }
    ]
  }
}