JDKの事前準備

LogstashはJavaランタイムに依存するため、まずJDK 8をシステムに展開します。

# JDKアーカイブの展開とシンボリックリンク作成
sudo tar xzf jdk-8u202-linux-x64.tar.gz -C /usr/local/
sudo ln -sf /usr/local/jdk1.8.0_202 /usr/local/java

# 環境変数の永続化（/etc/profile.d/java.sh）
echo 'export JAVA_HOME=/usr/local/java' | sudo tee /etc/profile.d/java.sh
echo 'export PATH=$JAVA_HOME/bin:$PATH' | sudo tee -a /etc/profile.d/java.sh
source /etc/profile.d/java.sh

Logstashのインストール手法

パッケージ管理による導入が推奨されます。RPMベース環境では以下のように実行します。

sudo yum install -y https://artifacts.elastic.co/downloads/logstash/logstash-7.17.3.rpm
# サービス登録と起動
sudo systemctl daemon-reload
sudo systemctl enable logstash
sudo systemctl start logstash

基本的な動作確認

コマンドラインで簡易パイプラインを構築し、標準入力から受信したデータを整形して出力します。

logstash -e '
input { stdin { } }
filter { mutate { add_field => { "received_at" => "%{@timestamp}" } } }
output { stdout { codec => json_lines } }
'

Nginxログ向けカスタムパイプライン設定

以下の/etc/logstash/conf.d/nginx-pipeline.confは、アクセスログとエラーログを分岐処理し、地理情報や時刻正規化を適用します。

input {
  beats {
    port => 5044
    codec => json
  }
}

filter {
  if "nginx" in [tags] {
    # アクセスログ処理（GeoIP付与）
    if "access" in [tags] {
      grok {
        match => { "message" => '%{IPORHOST:remote_addr} - %{DATA:user_name} \[%{HTTPDATE:timestamp}\] "%{WORD:http_method} %{DATA:url} HTTP/%{NUMBER:http_version}" %{NUMBER:status_code} %{NUMBER:body_bytes_sent} "%{DATA:referrer}" "%{DATA:user_agent}"' }
      }
      date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
        target => "@timestamp"
      }
      geoip {
        source => "remote_addr"
        database => "/usr/share/GeoIP/GeoLite2-City.mmdb"
        fields => ["country_name", "region_name", "city_name", "latitude", "longitude"]
      }
      mutate {
        convert => { "latitude" => "float" "longitude" => "float" }
      }
    }

    # エラーログ処理（厳密な日時抽出）
    if "error" in [tags] {
      grok {
        match => { "message" => '%{TIMESTAMP_ISO8601:log_time} \[%{LOGLEVEL:level}\] %{NUMBER:pid:int}#%{NUMBER}: \*%{NUMBER} %{GREEDYDATA:details}' }
      }
      date {
        match => ["log_time", "ISO8601"]
        target => "@timestamp"
      }
      mutate {
        remove_field => ["log_time", "message"]
      }
    }
  }
}

output {
  if "nginx-access-hc" in [tags] {
    elasticsearch {
      hosts => ["http://192.168.1.126:9200"]
      index => "nginx-hc-access-%{+YYYY.MM.dd}"
      template_name => "nginx-access"
    }
  }
  else if "nginx-error-sj" in [tags] {
    elasticsearch {
      hosts => ["http://192.168.1.126:9200"]
      index => "nginx-sj-error-%{+YYYY.MM.dd}"
      template_name => "nginx-error"
    }
  }
  stdout { codec => dots }
}

Filebeat側の対応設定

NginxログをLogstashへ送信するには、Filebeatのfilebeat.ymlで適切なタグとパスを指定します。

filebeat.inputs:
- type: filestream
  enabled: true
  paths:
    - /var/log/nginx/hc_access.log
  tags: ["nginx", "access", "nginx-access-hc"]

- type: filestream
  enabled: true
  paths:
    - /var/log/nginx/sj_error.log
  tags: ["nginx", "error", "nginx-error-sj"]

output.logstash:
  hosts: ["192.168.1.126:5044"]