🔍

Logstashを使用してApacheのログをElasticsearchに取り込んでみた

2022/03/20に公開

Elasticsearch

logstash

Elastic Stack

tech

elasticの公式リポジトリにサンプルデータがいくつも存在しており、この中にApacheのアクセスログがあるので、これをLogstashを使ってElasticsearchに取り込んでみました。

サンプルデータのリポジトリ

GitHub - elastic/examples: Home for Elasticsearch examples available to everyone. It's a great way to get started.

Apacheのアクセスログ

examples/Common Data Formats/apache_logs at master · elastic/examples · GitHub

環境

Elasticsearch、LogstashはDockerにて動かします。取り込むログファイルやLogstashの設定ファイルはDockerVolumeとして読み込ませました。Kibanaもいますけれど、今回は関係ないです。

version: '3.9'
services:
  es01:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.0.0
    ports:
      - 9200:9200
      - 9300:9300
    environment:
      - discovery.type=single-node
      - action.destructive_requires_name=true
      - xpack.security.enabled=false
      - path.repo=/usr/share/elasticsearch/backup
    networks:
      - elastic
    volumes:
      - backup:/usr/share/elasticsearch/backup
    deploy:
      resources:
        limits:
          memory: 2G
  ki01:
    image: docker.elastic.co/kibana/kibana:8.0.0
    environment:
      ELASTICSEARCH_HOSTS: "http://es01:9200"
    ports:
      - 5601:5601
    networks:
      - elastic
    deploy:
      resources:
        limits:
          memory: 2G
  lo01:
    image: docker.elastic.co/logstash/logstash-oss:8.0.0
    environment:
      - MONITORING_ENABLED=false
    volumes:
      - ./logstash/example-logstash.conf:/usr/share/logstash/pipeline/logstash.conf
      - ./logstash/apache_log:/usr/share/logstash/example-data/apache_log
      - ./logstash/completed_log_file:/usr/share/logstash/example-data/completed_log_file
    networks:
      - elastic
    deploy:
      resources:
        limits:
          memory: 1G
  fi01:
    image: docker.elastic.co/beats/filebeat:8.0.0
    volumes:
      - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml
      - ./filebeat/apache_log:/var/apache_log
    networks:
      - elastic
    deploy:
      resources:
        limits:
          memory: 1G
volumes:
  backup:
    driver: local
networks:
  elastic:
    driver: bridge

DockerでLogstashを起動する場合は /usr/share/logstash/pipeline/ ディレクトリに設定ファイルを配置します。すると、コンテナを起動すれば勝手にファイルを読み込んで実行してくれます。補足ですが、複数の設定ファイルがある場合それぞれパイプラインが作成されることになります（パイプライン数の数は別の設定ファイルで設定できたはず）。elasticが提供しているコンテナはデフォルトの設定ファイルが logstash.conf として既に存在するので、docker-compose.ymlにてボリュームをマウントする時に上書きするようにしています。

LogstashをDockerで使用するやり方について詳しくは以下のページに記載されています。

Configuring Logstash for Docker | Logstash Reference [8.1] | Elastic

今回使っているDockerイメージですが、ポストフィックスに -oss とあります。これがついていないもので起動してみたのですが、ライセンスの関係でうまくいきませんでした。-oss は無償で使って良いライセンスになっているようです。elasticが公式に出しているDockerイメージはこちらから確認できます。

Docker @ Elastic | Elastic

Logstashの設定

Logstashの使い方についてざっくりと読んで、ざっくりと設定しました。

Logstashは大きく分けて次のような処理をするみたいです。

How Logstash Works | Logstash Reference [8.1] | Elastic

入力 ... データを読み込む。
フィルタ ... データを分析、加工する。
出力 ... データを吐き出す。

上記の3つの処理をまとめて、パイプラインと言うらしいです。パイプラインは設定ファイルで定義します。

Structure of a config file | Logstash Reference [8.1] | Elastic

今回作成した設定ファイルは次の通りです。推測ですが、Rubyのシンタックスで記載するみたいですね。input filter output が上記の処理ステップにそれぞれ対応しています。個々のセクションの中で、それぞれパラメータを設定するような作りになっています。

input {
  file {
    path => "/usr/share/logstash/example-data/apache_log"
    mode => "read"
    file_completed_action => "log"
    file_completed_log_path => "/usr/share/logstash/example-data/completed_log_file"
  }
}
filter {
  grok {
    match => { "message" => '%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}' }
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    locale => "en"
  }
}
output {
  elasticsearch {
    hosts => [ "es01:9200" ]
    index => "logtest-%{{yyyy.MM}}"
  }
}

input では指定したファイルを入力データとして扱うようにしました。mode を read に設定してファイルを先頭から読み込ませるようにしています。デフォルトの動作だとtailするため新たに追加された行しか読み込んでくれません。なので今回のように事前にログが書き込まれている場合はデフォルトの動作だとうまくいきません。また、 file_completed_action を log にしていますが、デフォルトの動作だと読み込んだ後にファイルが削除されてしまうようなのでこのようにしています。

filter ではgrokプラグインを使ってログを構造化しています。grokの使い方はネットにたくさん転がっているのでそれらを参考にすれば一般的なログフォーマットであれば事足りそうです。ただ、特殊な形式か、独自の形式をサポートしたい場合などはしっかり調べないとダメそうです。grokに関しては少し奥が深そうですね。

また、 filter にてdateプラグインを使ってログの時刻データを処理しています。

Date filter plugin | Logstash Reference [8.1] | Elastic

dateプラグインを使うことで時刻を表す文字列をパースすることができます。さらに、ロケール情報も合わせて追加できます。パースしたデータは時刻情報としてそれ以降の処理で利用することができます。今回は時刻情報を使いログを異なるインデックスに振り分けています。

dateプラグインはパースした文字列をデフォルトで @timestamp フィールドに保存します。このフィールドはLogstashが初めから予約しているフィールド名になっています。別のフィールドに格納することもできますが、基本的に変える必要はないでしょう。

ここでフィールドという言葉が出てきましたが、フィールドが何であるのか、どういった使い方ができるのかについてはこの辺りに書いてありました。

Accessing event data and fields in the configuration | Logstash Reference [8.1] | Elastic

output ではelasticsearchに解析した結果を出力しています。 index に年月を指定することで、月ごとに別のインデックスにしています。ログデータの場合、大体は月または日ごとにインデックスを分けるんじゃないでしょうか。このように分ければ、ログの参照頻度に応じてデータの圧縮や削除を行うことが容易になります。便利ですね。

時刻データ（@timestamp）のフィールドへの参照方法は他と違っていて、 {{<フォーマット形式>}} とできるみたいです。おそらくよく使われる参照データなので簡単に書けるようにしているんでしょうね。フォーマット形式の指定は、JavaのDateTimeFormatterに従うようです。

DateTimeFormatter | Java 11 API

実行する

次のコマンドにより実行します。事前にelasticsearchは立ち上げておく必要があります。

$ docker compose up lo01

起動後のログはこんな感じ。一部を抜粋しています。

elastic-stack-lo01-1  | [2022-03-20T03:22:23,517][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>250, "pipeline.sources"=>["/usr/share/logstash/pipeline/logstash.conf"], :thread=>"#<Thread:0x523b2745 run>"}
elastic-stack-lo01-1  | [2022-03-20T03:22:25,144][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
elastic-stack-lo01-1  | [2022-03-20T03:22:25,217][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
elastic-stack-lo01-1  | [2022-03-20T03:22:25,244][INFO ][filewatch.observingread  ][main][3541ad80183caa8e3cba3fce295c047e2e81e3e5484c373c62a635dc40efbf18] START, creating Discoverer, Watch with file and sincedb collections

pipeline.sourcesの箇所で読み込んだパイプラインの設定ファイルが書かれていますね。正しい設定ファイルが読み込まれているのかどうかはここを見ればわかりますね。

Kibanaを立ち上げてログがどのように取り込まれたのかを見てみます。

GET _cat/indices

yellow open logtest-2022.03 ITAM0Vp-QWK6MNHlqctaQw 1 1    1 0 9.9kb 9.9kb
yellow open logtest-2015.05 ldm4q7jIQU2n5IrhqEEToQ 1 1 9999 0 6.1mb 6.1mb

２つのインデックスが作成されているのがわかりました。調べてみたところ読み込みに使用したログファイルは全て2015年の時刻なのですが一部のログが想定していない形式になっているため、誤った時刻データとして処理されているようです。事前に想定していないような形式のログがある場合、どうすればいいんでしょうかね？今回はスルーします。

検索してドキュメントを確認します。

GET logtest-*/_search
{
  "query": {
    "match_all": {}
  }
}

検索結果の一部抜粋です。

"hits" : [
      {
        "_index" : "logtest-2015.05",
        "_id" : "g9NZpX8BMC2yzG7fBnHy",
        "_score" : 1.0,
        "_source" : {
          "timestamp" : "18/May/2015:13:05:37 +0000",
          "message" : "208.115.113.88 - - [18/May/2015:13:05:37 +0000] \"GET /articles/ppp-over-ssh HTTP/1.1\" 301 336 \"-\" \"Mozilla/5.0 (compatible; Ezooms/1.0; help@moz.com)\"",
          "agent" : "\"Mozilla/5.0 (compatible; Ezooms/1.0; help@moz.com)\"",
          "referrer" : "\"-\"",
          "ident" : "-",
          "httpversion" : "1.1",
          "verb" : "GET",
          "auth" : "-",
          "@timestamp" : "2015-05-18T13:05:37Z",
          "clientip" : "208.115.113.88",
          "log" : {
            "file" : {
              "path" : "/usr/share/logstash/example-data/apache_log"
            }
          },
          "host" : {
            "name" : "14e739f913d0"
          },
          "response" : 301,
          "@version" : "1",
          "bytes" : 336,
          "event" : {
            "original" : "208.115.113.88 - - [18/May/2015:13:05:37 +0000] \"GET /articles/ppp-over-ssh HTTP/1.1\" 301 336 \"-\" \"Mozilla/5.0 (compatible; Ezooms/1.0; help@moz.com)\""
          },
          "request" : "/articles/ppp-over-ssh"
        }
      },

正しくドキュメントが作成されているのがわかりました。grokプラグインを用いてログを構造化しましたがそちらも正しく動作しているようです。@timestampのフィールドも正しく時刻データを示していますね。

環境

Logstashの設定

実行する

Discussion