例: 一般ログ形式でのログ解析

この例のチュートリアルでは、ingest pipelineを使用して、インデックス作成前に一般ログ形式のサーバーログを解析します。開始する前に、ingest pipelinesの前提条件を確認してください。

解析したいログは次のようになります:

ログ

  1. 212.87.37.154 - - [05/May/2099:16:21:15 +0000] "GET /favicon.ico HTTP/1.1" 200 3638 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36"

これらのログには、タイムスタンプ、IPアドレス、およびユーエージェントが含まれています。これらの3つの項目をElasticsearchの独自のフィールドに設定して、より迅速な検索と視覚化を実現したいと考えています。また、リクエストがどこから来ているのかを知りたいです。

  • 1. Kibanaで、メインメニューを開き、Stack Management \u0026gt; Ingest
    Pipelinesをクリックします。
    KibanaのIngest Pipelinesリストビュー
  • 2. Create pipeline > New pipelineをクリックします。
  • 3. Nameをmy-pipelineに設定し、オプションでパイプラインの説明を追加します。
  • 4. grok processorを追加してログメッセージを解析します:
    • 4.1. Add a processorをクリックし、Grokプロセスタイプを選択します。
    • 4.2. Fieldをmessageに設定し、Patternsを次のgrok patternに設定します:

Grok

  1. %{IPORHOST:source.ip} %{USER:user.id} %{USER:user.name} \[%{HTTPDATE:@timestamp}\] "%{WORD:http.request.method} %{DATA:url.original} HTTP/%{NUMBER:http.version}" %{NUMBER:http.response.status_code:int} (?:-|%{NUMBER:http.response.body.bytes:int}) %{QS:http.request.referrer} %{QS:user_agent}
  • 4.3. Addをクリックしてプロセッサを保存します。
    • 4.4. プロセッサの説明をExtract fields from 'message'に設定します。
  • 5. タイムスタンプ、IPアドレス、およびユーエージェントフィールドのためのプロセッサを追加します。プロセッサを次のように構成します:
    | プロセスタイプ | フィールド | 追加オプション | 説明 |
    | :— | :— | :— | :— |
    | Date | @timestamp | Formats: dd/MMM/yyyy:HH:mm:ss Z | Format '@timestamp' as 'dd/MMM/yyyy:HH:mm:ss Z' |
    | GeoIP | source.ip | Target field: source.geo | Add 'source.geo' GeoIP data for 'source.ip' |
    | User agent | user_agent | | Extract fields from 'user_agent' |
    あなたのフォームは次のようになります:
    Ingest Pipelinesのプロセッサ
    4つのプロセッサは順次実行されます:
    Grok \u0026gt; Date \u0026gt; GeoIP \u0026gt; User agent
    矢印アイコンを使用してプロセッサの順序を変更できます。
    または、Import processorsリンクをクリックして、プロセッサをJSONとして定義できます:

Js

  1. {
  2.    "processors": [
  3.    {
  4.    "grok": {
  5.    "description": "Extract fields from 'message'",
  6.    "field": "message",
  7.    "patterns": ["%{IPORHOST:source.ip} %{USER:user.id} %{USER:user.name} \\[%{HTTPDATE:@timestamp}\\] \"%{WORD:http.request.method} %{DATA:url.original} HTTP/%{NUMBER:http.version}\" %{NUMBER:http.response.status_code:int} (?:-|%{NUMBER:http.response.body.bytes:int}) %{QS:http.request.referrer} %{QS:user_agent}"]
  8.    }
  9.    },
  10.    {
  11.    "date": {
  12.    "description": "Format '@timestamp' as 'dd/MMM/yyyy:HH:mm:ss Z'",
  13.    "field": "@timestamp",
  14.    "formats": [ "dd/MMM/yyyy:HH:mm:ss Z" ]
  15.    }
  16.    },
  17.    {
  18.    "geoip": {
  19.    "description": "Add 'source.geo' GeoIP data for 'source.ip'",
  20.    "field": "source.ip",
  21.    "target_field": "source.geo"
  22.    }
  23.    },
  24.    {
  25.    "user_agent": {
  26.    "description": "Extract fields from 'user_agent'",
  27.    "field": "user_agent"
  28.    }
  29.    }
  30.    ]
  32.   }
  • 6. パイプラインをテストするには、Add documentsをクリックします。
  • 7. Documentsタブで、テスト用のサンプルドキュメントを提供します:

Js

  1. [
  2.    {
  3.    "_source": {
  4.    "message": "212.87.37.154 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\""
  5.    }
  6.    }
  7.   ]
  • 8. Run the pipelineをクリックして、パイプラインが期待通りに動作したことを確認します。
  • 9. すべてが正しく見える場合は、パネルを閉じて、Create
    pipelineをクリックします。
    これで、ログデータをdata streamにインデックス作成する準備が整いました。
  • 10. index templateを作成し、data stream enabledを設定します。

Python

  1. resp = client.indices.put_index_template(
  2.    name="my-data-stream-template",
  3.    index_patterns=[
  4.    "my-data-stream*"
  5.    ],
  6.    data_stream={},
  7.    priority=500,
  8.   )
  9.   print(resp)

Js

  1. const response = await client.indices.putIndexTemplate({
  2.    name: "my-data-stream-template",
  3.    index_patterns: ["my-data-stream*"],
  4.    data_stream: {},
  5.    priority: 500,
  6.   });
  7.   console.log(response);

Console

  1. PUT _index_template/my-data-stream-template
  2.   {
  3.    "index_patterns": [ "my-data-stream*" ],
  4.    "data_stream": { },
  5.    "priority": 500
  6.   }
  • 11. 作成したパイプラインを使用してドキュメントをインデックスします。

Python

  1. resp = client.index(
  2.    index="my-data-stream",
  3.    pipeline="my-pipeline",
  4.    document={
  5.    "message": "89.160.20.128 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\""
  6.    },
  7.   )
  8.   print(resp)

Js

  1. const response = await client.index({
  2.    index: "my-data-stream",
  3.    pipeline: "my-pipeline",
  4.    document: {
  5.    message:
  6.    '89.160.20.128 - - [05/May/2099:16:21:15 +0000] "GET /favicon.ico HTTP/1.1" 200 3638 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36"',
  7.    },
  8.   });
  9.   console.log(response);

Console

  1. POST my-data-stream/_doc?pipeline=my-pipeline
  2.   {
  3.    "message": "89.160.20.128 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\""
  4.   }
  • 12. 確認するために、データストリームを検索してドキュメントを取得します。次の検索は、filter_pathを使用して、document sourceのみを返します。

Python

  1. resp = client.search(
  2.    index="my-data-stream",
  3.    filter_path="hits.hits._source",
  4.   )
  5.   print(resp)

Ruby

  1. response = client.search(
  2.    index: 'my-data-stream',
  3.    filter_path: 'hits.hits._source'
  4.   )
  5.   puts response

Js

  1. const response = await client.search({
  2.    index: "my-data-stream",
  3.    filter_path: "hits.hits._source",
  4.   });
  5.   console.log(response);

Console

  1. GET my-data-stream/_search?filter_path=hits.hits._source

APIは次のように返します：

コンソール-結果

  1. {
  2.    "hits": {
  3.    "hits": [
  4.    {
  5.    "_source": {
  6.    "@timestamp": "2099-05-05T16:21:15.000Z",
  7.    "http": {
  8.    "request": {
  9.    "referrer": "\"-\"",
  10.    "method": "GET"
  11.    },
  12.    "response": {
  13.    "status_code": 200,
  14.    "body": {
  15.    "bytes": 3638
  16.    }
  17.    },
  18.    "version": "1.1"
  19.    },
  20.    "source": {
  21.    "ip": "89.160.20.128",
  22.    "geo": {
  23.    "continent_name" : "Europe",
  24.    "country_name" : "Sweden",
  25.    "country_iso_code" : "SE",
  26.    "city_name" : "Linköping",
  27.    "region_iso_code" : "SE-E",
  28.    "region_name" : "Östergötland County",
  29.    "location" : {
  30.    "lon" : 15.6167,
  31.    "lat" : 58.4167
  32.    }
  33.    }
  34.    },
  35.    "message": "89.160.20.128 - - [05/May/2099:16:21:15 +0000] \"GET /favicon.ico HTTP/1.1\" 200 3638 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\"",
  36.    "url": {
  37.    "original": "/favicon.ico"
  38.    },
  39.    "user": {
  40.    "name": "-",
  41.    "id": "-"
  42.    },
  43.    "user_agent": {
  44.    "original": "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36\"",
  45.    "os": {
  46.    "name": "Mac OS X",
  47.    "version": "10.11.6",
  48.    "full": "Mac OS X 10.11.6"
  49.    },
  50.    "name": "Chrome",
  51.    "device": {
  52.    "name": "Mac"
  53.    },
  54.    "version": "52.0.2743.116"
  55.    }
  56.    }
  57.    }
  58.    ]
  59.    }
  60.   }