エージェントレスなApacheログ解析ツールを試作した話

はじめに

エージェント導入が制限された環境において、Apacheアクセスログを効率的に分析するために
軽量ログ解析ツールを実装したお話をしたいと思います。
近年はAIエージェントにコードを丸ごと生成してもらうことも可能ですが、
今回はあえてチャット形式でのペアプログラミングという形を選びました。
理由としては、自分自身が設計の理由やデータ構造の意図を理解しながら構築したかったからです。

背景と課題

私の中で次のようなニーズがありました。
・特定日のアクセス状況を確認したい
・急増トラフィックの原因を特定したい
・アクセス元の国を把握したい

運用上の要件としては以下の通りです。
・迅速に把握や特定できること

一方で次のような課題もありました。
・サーバーに追加ソフトウェアをインストールできない
・常時ログ転送によるトラフィック増加を避けたい
・SplunkなどのSIEMを導入するほどの規模ではない

これらを満たすため、エージェントレス・Pull型・都度実行可能はログ解析ツールを設計しました。

成果物

ホスト名 or IPアドレス, ユーザー名, パスワード, アクセスログのパスを入力したあと、
日付を指定することでSSH接続　→　ログの読み込み　→　解析が始まります。
解析完了後は対応する番号を入力することで集計結果を表示する仕組みです。

時間帯別アクセスではヒストグラム形式でアクセス数を表示。
気になる時刻を入力すると、接続元のIPアドレスと国名などの情報が表示されます。

実験的に地理情報もヒートマップで表示可能にしてあります。

設計

ログ取得はPCからSSH経由で行います。
PC　→　SSH　→　Apacheアクセスログ　→　SFTP
サーバー側には常駐プロセスを置かず、解析時のみ接続します。
これにより、
・サーバー負荷を最小化
・トラフィック増加を回避
・導入コストの低減
の実現を目指しました。

技術選定

CLIベースで、最低限の視認性を確保する構成としています。

1.apache_log_parser

【公式リポジトリ】
https://github.com/amandasaurus/apache-log-parser
apache_log_parserはapacheのログを1行ずつ分解して集計しやすいように変換してくれるライブラリです。

2.rich

CLI上でプログレスバーを表示させたり、集計結果をきれいにテーブル表示させるため採用しました。

3.plotext

CLI上でグラフを表示したかったので、軽量なplotextを採用しました。

4.GeoLite2

今回はレスポンス重視でMaxMindが提供するGeoLite2を採用しました。
GeoLite2はデータベースで動作するため、定期更新が必要となりますが、
外部APIなどのレート制限やネットワークの影響を受けません。
そのため、安定して高速処理が可能となります。

設計上の工夫

1. ストリーム処理によるメモリ負荷低減

ログを一括でメモリ展開すると、ログサイズによってはPCがフリーズする可能性があります。
そこで、SFTPでログを1行ずつ取得し、逐次解析するstream方式を採用しました。
・ログ全体を保持しない
・メモリ使用量は集計キー数に依存
・大容量ログにも対応可能
この設計によりある程度の規模のログでも安定して処理できます。

def stream(self):
    sftp = self.ssh_client.client.open_sftp()
    with sftp.open(self.log_path, 'r') as f:
        for line in f:
            yield line

2. キャッシュで解析高速化

逐次解析であっても、大量ログでは処理時間がかかります。
そこで、解析結果をpickleで保存し、2回目以降はキャッシュを利用する仕組みを実装しました。
これにより、
・再解析時にかかる時間を短縮
・調査作業の効率化
を実現しています。

CACHE_DIR = "cache"
os.makedirs(CACHE_DIR, exist_ok=True)
if target_date:
    CACHE_FILE = os.path.join(CACHE_DIR, f"analyzer_{target_date}.pkl")

3．GeoIPによる国名・ASNの取得

今まではアクセスの多いIPアドレスから国名を確認していましたが、
ASNの確認も重要であることからASNの組織情報も表示できるように実装しました。
データベースが2つになってしまいましたが、
クラウド業者なのか、ISPなのか、ホスティング業者なのか判別できるので、
判断の1つになるのではと考えています。

self.city_reader = geoip2.database.Reader(
    os.path.join(GEO_DIR, "GeoLite2-City.mmdb")
)

self.asn_reader = geoip2.database.Reader(
    os.path.join(GEO_DIR, "GeoLite2-ASN.mmdb")
)

今後の展望

アクセス元や付随する情報を一括で確認できるようになりましたが、
急増するトラフィックの原因究明まではまだまだといったところです。
今後は、異常検知の実装を検討し、通常時との比較や傾向を可視化することで、
直感的に把握できる仕組みを取り入れていきたいと考えています。

まとめ

制約のある環境においても、工夫次第では解決策を見出すことは可能であると感じました。
同様の制約環境で悩む方の設計のヒントになれば幸いです。

この記事を読んだ方はこちらも読まれています

単純な作業はTeraTermのマクロに任せよう 次世代Wi-Fi規格「Wi-Fi 7」とは？ RushDrive PCクライアント2.5.0で発生中の問題 WSL2とWindows Terminalを使って生産性を上げよう！