CloudWatch Logs Insights — ログクエリと分析パターン

CloudWatch Logs Insights の概要

CloudWatch Logs InsightsはCloudWatchロググループのログデータをSQLライクな構文でインタラクティブに分析するサービスだ。

特徴:
  → Logs グループを横断したクエリ
  → スケールに関係なく高速（数十秒以内）
  → 結果の可視化（棒グラフ、折れ線グラフ）
  → CloudWatchダッシュボードへの埋め込み
  
コスト:
  → スキャンしたデータ量に対して課金
  → $0.005/GB スキャン（東京リージョン）
  → 時間範囲とロググループを絞ることで削減

基本クエリ構文

コマンド体系:
  fields: 表示するフィールドを選択
  filter: 条件でフィルタリング
  stats: 集計（count, sum, avg, min, max, percentile）
  sort: ソート
  limit: 件数制限
  parse: 正規表現やパターンでテキストを解析
  dedup: 重複排除

Lambda エラー分析

-- Lambda関数のエラーを分析
fields @timestamp, @message, @requestId
| filter @message like /ERROR/
| sort @timestamp desc
| limit 100

-- Lambda関数のコールドスタートを検出
fields @timestamp, @message, @duration, @billedDuration
| filter @message like /Init Duration/
| parse @message "Init Duration: * ms" as initDuration
| stats 
    count(*) as coldStarts,
    avg(initDuration) as avgInitMs,
    max(initDuration) as maxInitMs
| sort coldStarts desc

-- Lambda関数のタイムアウト検出
filter @message like /Task timed out/
| stats count(*) as timeouts by bin(5m)

APIレイテンシー分析

-- API Gatewayのレイテンシーパーセンタイル分析
fields @timestamp, @requestId, @latency, @status
| filter @type = "ACCESS"
| stats 
    percentile(@latency, 50) as p50,
    percentile(@latency, 90) as p90,
    percentile(@latency, 99) as p99,
    avg(@latency) as avgLatency,
    count(*) as requestCount
by bin(1h)

-- ステータスコード別の集計
fields @timestamp, status
| stats count(*) as count by status
| sort count desc

ALBアクセスログ分析

-- ALBのアクセスログでスロークエリを検出
fields time, elb, client_ip, request_url, 
       target_processing_time, response_processing_time
| filter target_processing_time > 1.0
| stats 
    avg(target_processing_time) as avgProcessingTime,
    count(*) as requestCount
by request_url
| sort avgProcessingTime desc
| limit 20

VPCフローログ分析

-- 拒否されたトラフィックのTOP10送信元
fields srcaddr, dstaddr, srcport, dstport, protocol, action
| filter action = "REJECT"
| stats count(*) as rejectCount by srcaddr
| sort rejectCount desc
| limit 10

-- 特定ポートへの接続試行（セキュリティ監査）
fields @timestamp, srcaddr, dstaddr, dstport, action
| filter dstport in [22, 3389, 1433]
| stats count(*) as attempts by srcaddr, dstport
| sort attempts desc

構造化ログの分析

-- JSON形式のアプリケーションログ（自動フィールド抽出）
fields @timestamp, level, message, userId, requestId
| filter level = "ERROR"
| stats count(*) as errorCount by userId
| sort errorCount desc

# Python ログ出力をJSON形式にする
import json
import logging

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name,
        }
        if hasattr(record, 'user_id'):
            log_data['userId'] = record.user_id
        return json.dumps(log_data)

# Lambda関数でJSON形式でログ出力
logger = logging.getLogger()
logger.handlers[0].setFormatter(JsonFormatter())

def lambda_handler(event, context):
    extra = {'user_id': event.get('userId', 'unknown')}
    logger.info('Processing request', extra=extra)

コスト最適化

スキャン範囲の絞り込み:
  1. 時間範囲を最小限にする（例: 直近1時間）
  2. クエリ対象のロググループを限定する
  3. filterコマンドでできるだけ早期にデータを絞る
  
ログ保持期間の最適化:
  → ロググループの保持期間を設定（デフォルトは無期限）
  → S3へのエクスポートでコスト削減（Cold Storage）
  
aws logs put-retention-policy \
  --log-group-name /aws/lambda/my-function \
  --retention-in-days 30

CloudWatch Logs サブスクリプション

# ログをリアルタイムでKinesisに転送（分析・アーカイブ用）
aws logs put-subscription-filter \
  --log-group-name /aws/lambda/my-function \
  --filter-name "all-events" \
  --filter-pattern "" \
  --destination-arn arn:aws:kinesis:ap-northeast-1:123456789012:stream/log-stream

試験頻出ポイント

シナリオ	回答
Lambda エラーのパターン分析	Logs Insights + filter @message like /ERROR/
ログデータのコスト削減	保持期間設定 + 時間範囲を絞ったクエリ
JSON ログのフィールドを自動認識	Logs Insights が JSON を自動パース
リアルタイムログ転送	サブスクリプションフィルター
Logs Insightsの課金基準	スキャンしたデータ量（GB単位）

まとめ

CloudWatch Logs InsightsはSQLライクな構文でロググループを横断分析できる強力なツールだ。Lambda、API Gateway、ALB、VPCフローログ等の分析に活用し、時間範囲を絞ることでスキャンコストを最小化する。

S	公式ソース確認済み
A	成功実績多数・失敗例少数
B	賛否両論
C	動作未確認・セキュリティリスク高
Z	個人所感