多くのシステムで共通する課題として、開発者がサーバーに直接アクセスできないことや、ログが各サーバーに散在し検索が困難であること、さらに大量のログデータによりリアルタイムなクエリが困難であることが挙げられます。これらの問題を解決するために、集中型のログ収集・分析基盤を構築します。

前提条件とアーキテクチャ概要

本環境では、以下のソフトウェアスタックを使用します：

• Elasticsearch：全文検索エンジン
• Kafka + ZooKeeper：メッセージキューと分散協調
• rsyslog：ログ転送クライアント
• Kibana：可視化ダッシュボード

すべてのコンポーネントは公式サイトまたは信頼できるミラーから入手可能です。

事前準備

ファイアウォールとSELinuxを無効化（本番環境では適切に設定）。NTPによる時刻同期を確立します。

*/5 * * * * /usr/bin/rdate -s time-b.nist.gov &>/dev/null

1. Elasticsearchクラスタのセットアップ

Java 8環境をRPMでインストールし、環境変数を設定します。

export JAVA_HOME=/usr/java/latest
export PATH=$JAVA_HOME/bin:$PATH

Elasticsearchを展開し、elasticsearch.yml を以下のように編集：

cluster.name: my-log-cluster
node.name: node-01
path.data: /data/es/data
path.logs: /data/es/logs
network.host: 192.168.1.10
discovery.seed_hosts: ["192.168.1.10", "192.168.1.11"]
cluster.initial_master_nodes: ["node-01", "node-02"]

専用ユーザーを作成し、権限とリソース制限を設定：

useradd esuser
chown -R esuser:esuser /data/es /usr/local/elasticsearch
echo "esuser soft nofile 65536" >> /etc/security/limits.conf
echo "vm.max_map_count=262144" >> /etc/sysctl.conf
sysctl -p

起動後、curl http://localhost:9200 で正常動作を確認。管理用プラグイン head も併せて導入推奨。

2. Logstashの設定

Logstashを展開し、messages.conf で入力と出力を定義：

input {
  file {
    path => "/var/log/messages"
    start_position => "beginning"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.1.10:9200", "192.168.1.11:9200"]
    index => "syslog-%{+YYYY.MM.dd}"
  }
}

構文チェック後、サービスを起動して動作を検証します。

3. Kafka + ZooKeeperクラスタ構築

Kafkaバンドル版のZooKeeperを使用。各ノードで zookeeper.properties を設定：

dataDir=/var/lib/zookeeper
clientPort=2181
server.1=192.168.1.20:2888:3888
server.2=192.168.1.21:2888:3888

/var/lib/zookeeper/myid に固有ID（1 or 2）を記述。

Kafkaの server.properties：

broker.id=1
listeners=PLAINTEXT://192.168.1.20:9092
log.dirs=/var/lib/kafka-logs
zookeeper.connect=192.168.1.20:2181,192.168.1.21:2181

ZooKeeper → Kafka の順で起動。トピック作成とプロデューサー/コンシューマーテストで通信を検証。

4. rsyslogクライアント設定

バージョン8以上にアップグレードし、Kafkaモジュールを追加：

yum install rsyslog-kafka -y

/etc/rsyslog.d/nginx.conf でNginxログをKafkaへ転送：

module(load="omkafka")
template(name="NginxTemplate" type="string" string="%msg%")

action(
  type="omkafka"
  topic="nginx_logs"
  broker=["192.168.1.20:9092"]
  template="NginxTemplate"
)

5. LogstashによるKafka→Elasticsearch連携

kafka_to_es.conf を作成：

input {
  kafka {
    bootstrap_servers => "192.168.1.20:9092"
    topics => ["nginx_logs"]
    consumer_threads => 3
  }
}
output {
  elasticsearch {
    hosts => ["192.168.1.10:9200"]
    index => "nginx-%{+YYYY-MM-dd}"
  }
}

6. Kibanaによる可視化

Kibanaを展開し、kibana.yml でElasticsearch接続先を指定：

server.host: "0.0.0.0"
elasticsearch.hosts: ["http://192.168.1.10:9200"]

ブラウザで http://host:5601 にアクセスし、インデックスパターンを登録。フィルタリングやグラフ作成でログ分析が可能になります。

トラブルシューティングのポイント

• JDKバージョンは全ノードで厳密に一致させる
• Elasticsearchのヒープサイズは32GB未満に抑える
• ZooKeeper/Kafka/Elasticsearchのノード数は奇数が望ましい
• サービスはsystemdまたはscreen/nohupで常駐させる