AWS Observability本番運用Vol2 CloudWatch Logs深掘り編｜Logs Insights × Lambda統合 × Centralized Logging × コスト最適化

1. なぜCloudWatch Logs深掘り編か — Vol1分散トレース編からの架橋 + 4本柱選定の現実解

Vol1「Observability実践 分散トレース実践編」では、X-Ray・Application Signals・ADOTという分散トレース3本柱でリクエストのEnd-to-End可視化を習得しました。トレースはサービス間の呼び出し連鎖と遅延原因を特定する強力な手段ですが、本番運用を6ヶ月以上継続すると、分散トレースだけでは解消できない「ログ観測の壁」に必ずぶつかります。

Vol1完遂者が直面するLogs観測の5つの壁

4本柱選定の現実解 — なぜこの4本柱か

本Vol2の4本柱は「ログ運用の本番品質を最短で確立する」観点で選定しました。

• Logs Insights (§2): ログの「見える化」起点。Query Language習熟なしに他の施策は効果半減。最初に習得すべき基盤技術。
• Lambda統合 (§3): Real-timeアラートとデータ変換を実現。障害検知レイテンシを「数分→5秒以内」に短縮する経路設計。
• Centralized Logging (§4): 100 Account規模では避けられない集約設計。Cross-Account可視化なしに本番障害対応は機能しない。
• コスト最適化 (§5): Retention戦略とTiered Storageで月額50%削減を実証。コスト制御なしに長期継続運用は不可能。

Vol1との連続性 — Observability 3本柱の完成へ

Observabilityは「Metrics・Logs・Traces」の3本柱で構成されます。Vol1がTracesを深掘りしたのに対し、Vol2はLogsを深掘りします。Vol1 (分散トレース) + Vol2 (Logs深掘り) の2作を完遂することで、Observabilityの2軸を本番品質で確立できます。

本Vol2で得られる4成果

1. Logs Insights QL習熟: Error Rate集計・Percentile分析・Cross-LogGroup横断クエリを5秒以内で実行するクエリ設計力
2. Lambda統合パターン実装: Subscription Filter × Lambda × Kinesis Firehoseを使い分けたReal-time処理とバッチ処理の本番設計
3. Centralized Logging Hub設計: Kinesis Data Streams中継によるCross-Account/Multi-Region集約とOrganizations統合の一元管理アーキテクチャ
4. Logsコスト50%削減実装: Retention戦略 (LogGroup別) + Tiered Storage (Hot/Warm/Cold) + Athena移行による実証済みコスト削減パターン

Observability実践 Vol1 — 分散トレース実践編 (X-Ray / Application Signals / ADOT)

2. CloudWatch Logs Insights本番運用 — Query Language × SQL風構文 × 集計 × 可視化

2.1 Logs Insights Query Language 基礎 — 5コマンド完全解説

CloudWatch Logs Insights は独自の Query Language (QL) を持ちます。すべてのクエリは以下のコマンドの組み合わせで構成されます。

SQL風構文との対比 — 実例

fields @timestamp, @message, level, service
| filter level = "ERROR"
| stats count(*) as ErrorCount by service, bin(5m)
| sort ErrorCount desc
| limit 20

SQL換算: SELECT timestamp, level, service, COUNT(*) AS ErrorCount FROM logs WHERE level='ERROR' GROUP BY service, DATE_TRUNC('5min',timestamp) ORDER BY ErrorCount DESC LIMIT 20

2.2 主要関数 — parse / pct / bin / count_distinct

parse コマンド — 非構造化ログから動的フィールドを抽出します。

fields @message
| parse @message "userId=* requestId=* duration=*ms" as userId, requestId, duration
| stats avg(duration) as AvgDuration by userId
| sort AvgDuration desc

2.3 本番で使う実践クエリ集

① エラー率集計 (5分バケット)

fields @timestamp, @message
| filter @message like /ERROR/
| stats count(*) as ErrorCount by bin(5m)
| sort @timestamp asc

② レイテンシ Percentile 分析 (p95/p99)

fields @timestamp, duration
| filter ispresent(duration)
| stats pct(duration, 95) as p95, pct(duration, 99) as p99 by bin(1h)
| sort @timestamp asc

③ ユーザー別アクセス数 Top20

fields @timestamp, userId
| filter ispresent(userId)
| stats count(*) as AccessCount by userId
| sort AccessCount desc
| limit 20

④ JSON構造化ログのネストフィールドクエリ

fields @timestamp, requestContext.requestId, httpMethod, status
| filter status >= 500
| stats count(*) as ServerErrors by httpMethod, status
| sort ServerErrors desc

⑤ Cross-Log-Group 横断検索 — Logs Insightsコンソールで複数LogGroupを選択することで、組織横断の横串検索が可能です。

fields @timestamp, @log, @message
| filter @message like /CRITICAL/
| sort @timestamp desc
| limit 50

2.4 可視化 × Saved Queries × Performance最適化

可視化 — stats クエリ結果は以下のグラフタイプで可視化し、CloudWatch Dashboardへ直接追加できます。

Saved Queries — 組織内クエリ集をフォルダ整理で共有し、命名規約 (サービス名_用途_更新日) で管理します。チーム内の車輪の再発明を防ぎ、新メンバーが即戦力クエリを利用できます。

Performance最適化 — スキャン量削減3原則

1. 時間範囲を最短から開始: 24時間ではなく5分窓から探索し、必要に応じて拡張。スキャン量は時間範囲に比例して課金されます。
2. filter を stats より前に適用: 先に件数を絞り込むことでスキャン量を削減。| filter level = "ERROR" | stats count(*)の順序が正解。
3. LogGroup絞り込み: 全LogGroup横断より対象サービスのLogGroupを明示指定することでスキャン対象を最小化。

mermaid01: Logs Insights クエリ評価フロー

flowchart LR
 A[ログデータ受信] --> B["fields\n表示フィールド選択"]
 B --> C["filter\n条件絞り込み"]
 C --> D["stats\n集計・集約"]
 D --> E["sort\n並べ替え"]
 E --> F["limit\n件数制限"]
 F --> G["可視化\nLine / Bar / Pie"]

 style A fill:#232f3e,color:#ffffff
 style B fill:#0073bb,color:#ffffff
 style C fill:#0073bb,color:#ffffff
 style D fill:#ff9900,color:#232f3e
 style E fill:#1a9c3e,color:#ffffff
 style F fill:#1a9c3e,color:#ffffff
 style G fill:#ff9900,color:#232f3e

3. Lambda統合パターン本番運用 — Subscription Filter × Lambda Destination × Kinesis Firehose連携

3-1. Subscription Filter — Filter Pattern構文完全ガイド

Subscription FilterはCloudWatch Logsのリアルタイム配信機構であり、LogGroupに新着するログイベントをフィルタリングしてLambda / Kinesis Data Streams / Kinesis Data Firehoseへ即時送信する。本番運用で必須となるFilter Pattern構文を完全網羅する。

① 文字列マッチ (String Term)

最も基本的なパターン。スペース区切りで複数指定するとAND条件になる。

# 単一文字列
ERROR

# AND条件 (ERRORかつDatabaseを含む)
ERROR Database

# OR条件 (ERRORまたはWARN)
?ERROR ?WARN

# NOT条件 (ERRORを含まないログ)
-ERROR

# 完全一致 (ダブルクォート)
"Connection refused"

# ワイルドカード (* = 0文字以上)
"TimeoutException*"

② JSON形式ログのフィルタリング

JSON構造化ログに対しては .fieldName 記法でネストフィールドを指定できる。

# JSONフィールドの値マッチ
{ $.level = "ERROR" }

# 数値比較 (レスポンスタイム500ms超)
{ $.responseTime > 500 }

# 複数条件 AND
{ $.level = "ERROR" && $.statusCode >= 500 }

# 複数条件 OR
{ $.level = "ERROR" || $.level = "FATAL" }

# フィールド存在チェック (IS EXISTS)
{ $.errorCode IS NOT NULL }

# 否定マッチ
{ $.level != "INFO" }

# 文字列前方一致
{ $.requestPath = "/api/payment*" }

③ スペース区切りログ (Space-delimited)

Apache / Nginx のアクセスログなど固定カラム形式の場合は [field1, field2, ...] 記法を使う。

# Apache Combined Log Format (status=5xx を抽出)
[host, ident, authuser, date, request, status=5*, bytes, referrer, agent]

# ELBアクセスログ (ターゲット処理時間1秒超)
[type, time, elb, client, target, request_processing_time, target_processing_time > 1.0, ...]

# ワイルドカードカラムをスキップ
[w1, w2, w3, status=4*, ...]

④ 数値比較演算子一覧

Subscription Filter上限と回避策

1LogGroupにつきSubscription Filterは最大2個 (2024年のアップデートで1個→2個に緩和。Account Subscription Filterも同じカウントに含まれる)。複数のDownstreamへ配信が必要な場合の設計パターン:

• Lambda内Fan-out: 1つのLambdaがSNS / SQS / Firehose / OpenSearchへ並列送信
• EventBridge Pipes: CloudWatch Logs → Kinesis → EventBridge Pipes → 複数Ruleで配信先分岐
• Kinesis Data Streams中継: 1Filter → KDS → 複数Lambda/Firehose Consumer

3-2. Lambda統合パターン — CloudWatch Logs → Lambda → SNS/Slack通知

実装構成: リアルタイムエラーアラート

CloudWatch Logs LogGroup
 ↓ Subscription Filter (Filter Pattern: { $.level = "ERROR" })
Lambda Function (log-alert-processor)
 ├─ SNS Topic → PagerDuty (P1インシデント)
 ├─ Slack Webhook → 開発チームチャンネル
 └─ SQS DLQ (Lambda失敗時のバックアップ)

Terraform実装

resource "aws_cloudwatch_log_subscription_filter" "error_alert" {
  name= "error-alert-subscription"
  log_group_name  = aws_cloudwatch_log_group.app.name
  filter_pattern  = "{ $.level = \"ERROR\" }"
  destination_arn = aws_lambda_function.log_alert_processor.arn
  distribution = "ByLogStream"

  depends_on = [aws_lambda_permission.allow_cloudwatch]
}

resource "aws_lambda_permission" "allow_cloudwatch" {
  statement_id  = "AllowCloudWatchLogs"
  action  = "lambda:InvokeFunction"
  function_name = aws_lambda_function.log_alert_processor.function_name
  principal  = "logs.amazonaws.com"
  source_arn = "${aws_cloudwatch_log_group.app.arn}:*"
}

resource "aws_lambda_function" "log_alert_processor" {
  function_name = "log-alert-processor"
  runtime = "python3.12"
  handler = "handler.lambda_handler"
  role = aws_iam_role.lambda_exec.arn
  filename= data.archive_file.lambda_zip.output_path

  environment {
 variables = {
SNS_TOPIC_ARN= aws_sns_topic.alerts.arn
SLACK_WEBHOOK= var.slack_webhook_url
DLQ_URL= aws_sqs_queue.dlq.url
 }
  }

  dead_letter_config {
 target_arn = aws_sqs_queue.dlq.arn
  }

  reserved_concurrent_executions = 100
}

resource "aws_sqs_queue" "dlq" {
  name = "log-processor-dlq"
  message_retention_seconds = 1209600  # 14日
}

resource "aws_cloudwatch_metric_alarm" "dlq_depth" {
  alarm_name = "log-processor-dlq-depth"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name= "ApproximateNumberOfMessagesVisible"
  namespace  = "AWS/SQS"
  period  = 60
  statistic  = "Sum"
  threshold  = 0
  alarm_actions = [aws_sns_topic.alerts.arn]
  dimensions = {
 QueueName = aws_sqs_queue.dlq.name
  }
}

Lambda処理コード (Python 3.12)

import base64
import gzip
import json
import os
import urllib.request

def lambda_handler(event, context):
 # CloudWatch Logsイベントのデコード (gzip + base64)
 compressed = base64.b64decode(event["awslogs"]["data"])
 payload = json.loads(gzip.decompress(compressed))

 log_group  = payload["logGroup"]
 log_stream = payload["logStream"]
 log_events = payload["logEvents"]

 error_messages = []
 for log_event in log_events:
  try:
record = json.loads(log_event["message"])
if record.get("level") == "ERROR":
 # PII/クレジットカードマスキング
 masked_msg = mask_pii(record.get("message", ""))
 error_messages.append({
  "timestamp": log_event["timestamp"],
  "message":masked_msg,
  "requestId": record.get("requestId", "unknown"),
 })
  except json.JSONDecodeError:
pass

 if error_messages:
  send_sns_alert(log_group, log_stream, error_messages)
  send_slack_alert(log_group, error_messages)

 return {"statusCode": 200, "processed": len(error_messages)}

def mask_pii(text: str) -> str:
 import re
 # クレジットカード番号マスキング (例: 4111-1111-1111-1111 → ****-****-****-1111)
 text = re.sub(r"\b(\d{4})[- ]?(\d{4})[- ]?(\d{4})[- ]?(\d{4})\b",
r"****-****-****-\4", text)
 # メールアドレスマスキング
 text = re.sub(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}",
"****@****.***", text)
 return text

def send_sns_alert(log_group, log_stream, errors):
 import boto3
 sns = boto3.client("sns")
 sns.publish(
  TopicArn=os.environ["SNS_TOPIC_ARN"],
  Subject=f"[ERROR] {log_group}",
  Message=json.dumps({
"logGroup":  log_group,
"logStream": log_stream,
"errors": errors[:5],  # 最大5件
  }, ensure_ascii=False, indent=2),
 )

def send_slack_alert(log_group, errors):
 webhook = os.environ.get("SLACK_WEBHOOK")
 if not webhook:
  return
 payload = json.dumps({
  "text": f":rotating_light: *ERROR detected* in `{log_group}`\n"
 + "\n".join(f"• {e['message'][:200]}" for e in errors[:3])
 }).encode()
 req = urllib.request.Request(webhook, data=payload,
headers={"Content-Type": "application/json"})
 urllib.request.urlopen(req, timeout=5)

コールドスタート対策: Provisioned Concurrency

エラーアラートのLambdaは遅延が許容されないため、Provisioned Concurrencyを設定してコールドスタートを排除する。

resource "aws_lambda_provisioned_concurrency_config" "log_alert" {
  function_name= aws_lambda_function.log_alert_processor.function_name
  qualifier = aws_lambda_alias.log_alert_live.name
  provisioned_concurrent_executions = 5
}

resource "aws_lambda_alias" "log_alert_live" {
  name = "live"
  function_name = aws_lambda_function.log_alert_processor.function_name
  function_version = aws_lambda_function.log_alert_processor.version
}

Lambda Layer活用: ログパーサー共通化

複数Lambda間でログパース・マスキングロジックを共通化するには Lambda Layer が有効。

resource "aws_lambda_layer_version" "log_parser" {
  filename= "log_parser_layer.zip"
  layer_name = "log-parser-common"
  compatible_runtimes = ["python3.12"]
  description= "共通ログパーサー / PIIマスキングライブラリ"
}

resource "aws_lambda_function" "log_alert_processor" {
  # ... (前述の設定)
  layers = [aws_lambda_layer_version.log_parser.arn]
}

3-3. Lambda Destination — Dead Letter Queue (DLQ) × 失敗ハンドリング本番設計

Lambda Destinationは非同期呼び出しの成功/失敗を別のAWSサービスへ転送する機能。CloudWatch LogsからのSubscription Filter経由呼び出しは同期呼び出しのためDestinationは使えないが、Lambda内部でSQS DLQを明示的に実装することで同等の失敗ハンドリングを実現する。

失敗ハンドリング設計パターン

非同期Lambda (EventBridge起点など) でのDestination設定 (CDK)

import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as destinations from 'aws-cdk-lib/aws-lambda-destinations';

const dlq = new sqs.Queue(this, 'LogProcessorDlq', {
  retentionPeriod: cdk.Duration.days(14),
  encryption: sqs.QueueEncryption.KMS_MANAGED,
});

const successQueue = new sqs.Queue(this, 'LogProcessorSuccess', {
  retentionPeriod: cdk.Duration.days(1),
});

const logProcessor = new lambda.Function(this, 'LogProcessor', {
  runtime: lambda.Runtime.PYTHON_3_12,
  handler: 'handler.lambda_handler',
  code: lambda.Code.fromAsset('lambda/log_processor'),
  // 非同期呼び出し用Destination設定
  onSuccess: new destinations.SqsDestination(successQueue),
  onFailure: new destinations.SqsDestination(dlq),
  retryAttempts: 2,  // 失敗時に最大2回リトライ後DLQ送信
  maxEventAge: cdk.Duration.hours(6),
});

3-4. Kinesis Data Firehose連携 — CloudWatch Logs → S3 → Athena 長期保存パイプライン

全体アーキテクチャ

CloudWatch Logs LogGroup
 ↓ Subscription Filter (Filter Pattern: "" = 全ログ)
Kinesis Data Firehose Delivery Stream
 ├─ 変換Lambda (JSON正規化 / gzip解凍 / タイムスタンプ正規化)
 ├─ Buffer: size=128MB / interval=300秒
 ├─ S3 Prefix: logs/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/
 └─ S3 Intelligent-Tiering (30日後→IA / 90日後→Archive)

S3 (長期保存バケット)
 ↓ Glue Crawler (日次スケジュール)
AWS Glue Data Catalog
 ↓
Amazon Athena (アドホッククエリ / コスト最適化)

Terraform実装

resource "aws_kinesis_firehose_delivery_stream" "cloudwatch_logs" {
  name  = "cloudwatch-logs-to-s3"
  destination = "extended_s3"

  extended_s3_configuration {
 role_arn= aws_iam_role.firehose_role.arn
 bucket_arn = aws_s3_bucket.logs_archive.arn

 # Hive形式パーティション (Athenaパフォーマンス最適化)
 prefix  = "logs/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
 error_output_prefix = "errors/!{firehose:error-output-type}/!{timestamp:yyyy}/!{timestamp:MM}/"

 buffering_size  = 128  # MB (最大128MB)
 buffering_interval = 300  # 秒 (最大900秒)
 compression_format = "GZIP"

 # 変換Lambda
 processing_configuration {
enabled = true
processors {
  type = "Lambda"
  parameters {
 parameter_name  = "LambdaArn"
 parameter_value = "${aws_lambda_function.firehose_transformer.arn}:$LATEST"
  }
  parameters {
 parameter_name  = "BufferSizeInMBs"
 parameter_value = "3"
  }
  parameters {
 parameter_name  = "BufferIntervalInSeconds"
 parameter_value = "60"
  }
}
 }
  }
}

resource "aws_cloudwatch_log_subscription_filter" "to_firehose" {
  name= "all-logs-to-firehose"
  log_group_name  = aws_cloudwatch_log_group.app.name
  filter_pattern  = ""  # 全ログ配信
  destination_arn = aws_kinesis_firehose_delivery_stream.cloudwatch_logs.arn
  role_arn  = aws_iam_role.cwlogs_to_firehose.arn
  distribution = "Random"
}

変換Lambda (JSON正規化 / gzip解凍)

CloudWatch LogsからFirehoseへ配信されるレコードはbase64+gzip圧縮されたCloudWatch Logsイベント形式。変換Lambdaで解凍・正規化してS3に保存することで、Athenaから直接クエリできる形式にする。

import base64
import gzip
import json

def lambda_handler(event, context):
 output = []
 for record in event["records"]:
  # base64デコード + gzip解凍
  compressed = base64.b64decode(record["data"])
  payload = json.loads(gzip.decompress(compressed))

  if payload.get("messageType") == "CONTROL_MESSAGE":
output.append({"recordId": record["recordId"],
"result": "Dropped", "data": record["data"]})
continue

  # 各ログイベントをNewline-delimited JSONに変換
  transformed_lines = []
  for log_event in payload.get("logEvents", []):
try:
 record_data = json.loads(log_event["message"])
except json.JSONDecodeError:
 record_data = {"message": log_event["message"]}

record_data.update({
 "cloudwatch_log_group":  payload["logGroup"],
 "cloudwatch_log_stream": payload["logStream"],
 "cloudwatch_timestamp":  log_event["timestamp"],
})
transformed_lines.append(json.dumps(record_data, ensure_ascii=False))

  new_data = "\n".join(transformed_lines) + "\n"
  output.append({
"recordId": record["recordId"],
"result":"Ok",
"data":  base64.b64encode(new_data.encode()).decode(),
  })

 return {"records": output}

S3 Intelligent-Tiering設定

resource "aws_s3_bucket_lifecycle_configuration" "logs_lifecycle" {
  bucket = aws_s3_bucket.logs_archive.id

  rule {
 id  = "intelligent-tiering"
 status = "Enabled"

 transition {
days = 0
storage_class = "INTELLIGENT_TIERING"
 }

 # 180日後に削除 (要件に応じて調整)
 expiration {
days = 180
 }
  }
}

resource "aws_s3_intelligent_tiering_configuration" "logs" {
  bucket = aws_s3_bucket.logs_archive.id
  name= "logs-tiering"

  tiering {
 access_tier = "DEEP_ARCHIVE_ACCESS"
 days  = 180
  }
  tiering {
 access_tier = "ARCHIVE_ACCESS"
 days  = 90
  }
}

Athenaクエリ例 (S3からログ分析)

-- 直近7日間のエラー件数集計 (パーティション指定でコスト削減)
SELECT
  date_format(from_unixtime(cloudwatch_timestamp / 1000), '%Y-%m-%d %H:00:00') AS hour,
  count(*) AS error_count,
  cloudwatch_log_group
FROM logs_archive
WHERE year  = '2026'
  AND month = '05'
  AND level = 'ERROR'
GROUP BY 1, 3
ORDER BY 1 DESC;

-- レスポンスタイム P99 (パスごと)
SELECT
  requestPath,
  approx_percentile(responseTime, 0.99) AS p99_ms,
  count(*) AS request_count
FROM logs_archive
WHERE year = '2026' AND month = '05' AND day >= '11'
GROUP BY requestPath
ORDER BY p99_ms DESC
LIMIT 20;

3-5. Kinesis Data Streams中継 — 1万RPS超のスケールアウト設計

Firehose単体では1シャードあたり最大500MB/s / 2000 records/s の制約がある。1万RPS超の大規模ログ処理にはKinesis Data Streams (KDS) を中継層に挟む。

CloudWatch Logs
 ↓ Subscription Filter
Kinesis Data Streams (Enhanced Fan-Out + On-demand mode)
 ├─ Consumer1: Kinesis Data Firehose → S3 (長期保存)
 ├─ Consumer2: Lambda (リアルタイムアラート)
 └─ Consumer3: OpenSearch Ingestion Pipeline (検索インデックス)

resource "aws_kinesis_stream" "log_stream" {
  name = "cloudwatch-log-stream"
  stream_mode_details {
 stream_mode = "ON_DEMAND"  # シャード数自動スケール
  }
  encryption_type = "KMS"
  kms_key_id= aws_kms_key.kinesis.arn
}

resource "aws_cloudwatch_log_subscription_filter" "to_kds" {
  name= "all-logs-to-kds"
  log_group_name  = aws_cloudwatch_log_group.app.name
  filter_pattern  = ""
  destination_arn = aws_kinesis_stream.log_stream.arn
  role_arn  = aws_iam_role.cwlogs_to_kds.arn
  distribution = "ByLogStream"  # ログストリーム単位でシャード分配
}

3-6. 監視 — Subscription Filter DeliveryStatus × Lambda失敗率

監視すべきメトリクス一覧

Terraform: DeliveryErrors アラート

resource "aws_cloudwatch_metric_alarm" "subscription_delivery_errors" {
  alarm_name = "subscription-filter-delivery-errors"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name= "DeliveryErrors"
  namespace  = "AWS/Logs"
  period  = 60
  statistic  = "Sum"
  threshold  = 0
  alarm_description= "Subscription Filter配信エラー検知 — ログロスト発生の可能性"
  alarm_actions = [aws_sns_topic.critical_alerts.arn]
  treat_missing_data  = "notBreaching"
  dimensions = {
 LogGroupName = aws_cloudwatch_log_group.app.name
 DestinationType = "LambdaFunction"
 DestinationArn  = aws_lambda_function.log_alert_processor.arn
  }
}

4. Centralized Logging本番運用 — Cross-Account × Multi-Region集約 × Organizations統合

Centralized Loggingの設計は「どのアカウントにどのログを、どの経路で集約するか」という3軸の決定から始まります。本番規模（50アカウント以上・Multi-Region）では、Hub Account集約モデルを採用し、Kinesis Data StreamをCross-Account Subscriptionの受信エンドポイントとして構成することが標準パターンです。

4.1 Hub Account設計パターン — ログ集約専用アカウントの構造

Hub Accountとセキュリティアカウントの分離原則

多くの組織がセキュリティアカウントをHub Accountとして流用しますが、本番規模ではログ専用の Log Archive Account を独立させることがAWSのベストプラクティスです。

分離すべき理由は3点あります。
– スケール影響の遮断: 大規模環境では1日数百GBから数TBのログが集約される。同居するとセキュリティ機能のレスポンスに影響
– 権限モデルの独立: ログ読み取り権限（監査チーム・開発チーム）とセキュリティ機能の権限を独立して設計可能
– コスト可視化: ログストレージコストをLog Archiveアカウント単体で追跡し、部署別チャージバックが実現可能

Hub Accountの構成コンポーネント：

アーキテクチャ全体像：

Spoke Account A (ap-northeast-1)
  └─ LogGroup: /aws/lambda/api-handler
 └─ Subscription Filter ──→ CWL Destination (Hub Account)
└─ Kinesis Data Stream (Hub / ap-northeast-1)
└─ Kinesis Firehose
└─ S3 (log-archive-bucket)

Spoke Account B (us-east-1)
  └─ LogGroup: /aws/ecs/web-service
 └─ Subscription Filter ──→ CWL Destination (Hub Account / us-east-1)
└─ Kinesis Data Stream (Hub / us-east-1)
└─ S3 CRR / Firehose
└─ S3 (log-archive-bucket / ap-northeast-1)

4.2 Cross-Account Subscription Filter設定 — Spoke→Hub方向のログ転送

Cross-Account SubscriptionにはAccount側とHub側で相互の設定が必要です。設定順序を誤るとPermission deniedでログが届かない落とし穴があります。

Step 1: Hub Account — Kinesis Data Stream作成

resource "aws_kinesis_stream" "log_aggregation" {
  name = "centralized-logs-${var.environment}"
  shard_count= 4  # 1シャード = 1MB/s入力・2MB/s出力。100LogGroup想定で4シャード
  retention_period = 24

  stream_mode_details {
 stream_mode = "PROVISIONED"
  }

  tags = {
 Name  = "centralized-logs"
 Environment = var.environment
 Purpose  = "log-aggregation"
  }
}

Step 2: Hub Account — CloudWatch Logs DestinationとResource Policy設定

resource "aws_cloudwatch_log_destination" "hub" {
  name = "centralized-log-destination"
  role_arn= aws_iam_role.cloudwatch_logs_to_kinesis.arn
  target_arn = aws_kinesis_stream.log_aggregation.arn
}

resource "aws_cloudwatch_log_destination_policy" "hub" {
  destination_name = aws_cloudwatch_log_destination.hub.name

  access_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Sid = "AllowSpokeAccountsSubscribe"
  Effect = "Allow"
  Principal = {
 AWS = var.spoke_account_ids
  }
  Action= "logs:PutSubscriptionFilter"
  Resource = aws_cloudwatch_log_destination.hub.arn
}
 ]
  })
}

Step 3: Hub Account — CWL→Kinesis IAMロール

resource "aws_iam_role" "cloudwatch_logs_to_kinesis" {
  name = "cloudwatch-logs-to-kinesis-role"

  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = {
  Service = "logs.amazonaws.com"
}
Action = "sts:AssumeRole"
Condition = {
  StringLike = {
 "aws:SourceArn" = "arn:aws:logs:ap-northeast-1:${var.hub_account_id}:*"
  }
}
 }]
  })
}

resource "aws_iam_role_policy" "cloudwatch_logs_to_kinesis" {
  name = "put-records-to-kinesis"
  role = aws_iam_role.cloudwatch_logs_to_kinesis.id

  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect= "Allow"
Action= ["kinesis:PutRecord", "kinesis:PutRecords"]
Resource = aws_kinesis_stream.log_aggregation.arn
 }]
  })
}

Step 4: Spoke Account — Subscription Filter設定

resource "aws_cloudwatch_log_subscription_filter" "to_hub" {
  name= "forward-to-hub-centralized"
  log_group_name  = "/aws/lambda/api-handler"
  filter_pattern  = ""
  destination_arn = "arn:aws:logs:ap-northeast-1:${var.hub_account_id}:destination:centralized-log-destination"
  distribution = "ByLogStream"
}

Resource Policy vs IAM Policy のトレードオフ：

4.3 Organizations統合 — 全アカウント一括Subscription Filter設定

Account-level Subscription Filter（Organizations全体）

Organizations統合機能を使うと、Organizations単位でCloudWatch Logsのサブスクリプションを一括設定できます。

import json
import boto3


def setup_org_subscription(destination_arn: str) -> None:
 """Organizations全アカウントにAccount-level Subscription Filterを設定"""
 logs_client = boto3.client("logs")

 logs_client.put_account_policy(
  policyName="centralized-logging-policy",
  policyDocument=json.dumps({
"DestinationArn": destination_arn,
"FilterPattern": "",
"Distribution": "ByLogStream"
  }),
  policyType="SUBSCRIPTION_FILTER_POLICY",
  scope="ALL",
 )

この設定により、Organizations配下の全アカウント・全LogGroupに自動でSubscription Filterが適用されます。新規アカウント追加時も自動適用されるため、オンボーディング工数がゼロになります。

CloudTrail Lakeとの比較 — ユースケース別使い分け：

Service Control Policy (SCP) との連携

SCPでSpokeアカウントからの logs:DeleteSubscriptionFilter を禁止することで、Centralized Loggingの設定削除を防ぎます。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "DenyDeleteCentralizedLoggingConfig",
"Effect": "Deny",
"Action": [
  "logs:DeleteSubscriptionFilter",
  "logs:DeleteAccountPolicy"
],
"Resource": "*",
"Condition": {
  "StringEquals": {
 "aws:RequestedRegion": ["ap-northeast-1", "us-east-1"]
  }
}
 }
  ]
}

AWS Control Tower統合 — Log Archive Account自動プロビジョニング

AWS Control Towerを利用している環境では、Log Archive AccountはLanding Zone構築時に自動作成されます。Control Towerが管理する構成要素：

• Log Archive Account: Organizations配下のCloudTrailログ・Configスナップショットを集約
• Audit Account: コンプライアンスチェック用（読み取り専用）
• Account Factory: 新規Account作成時に自動でSCP・ログ集約設定を適用

推奨は 別途専用Hub Account の設置です。Control Tower管理外のリソース（Kinesis Data Stream・Firehose等）をLog Archive Accountに配置すると、Control Tower Guardrailsとの競合が発生するためです。

4.4 Multi-Region集約 — クロスリージョンログ統合戦略

2段階集約アーキテクチャ

大規模Multi-Region環境では、単純に全リージョンのログを1リージョンのKinesis Data Streamに集約しようとするとネットワーク転送コストとレイテンシが問題になります。推奨は2段階集約です。

【第1段階: Region内集約】
ap-northeast-1 の全Spoke Account
  → Kinesis Data Stream (ap-northeast-1 / Hub Account)
  → Firehose → S3 (ap-northeast-1 / log-archive-bucket)

us-east-1 の全Spoke Account
  → Kinesis Data Stream (us-east-1 / Hub Account)
  → Firehose → S3 (us-east-1 / log-archive-bucket)

【第2段階: Global集約 (S3 Cross-Region Replication)】
S3 (ap-northeast-1) ──→ S3 (Global Archive / ap-northeast-1)
S3 (us-east-1)──→ S3 (Global Archive / ap-northeast-1)

Kinesis Data Stream経由のクロスリージョン転送（Lambda経由）

リアルタイム要件がある場合は KinesisのCross-Region転送をLambdaで実装します。

import base64
import boto3

target_kinesis = boto3.client("kinesis", region_name="ap-northeast-1")


def handler(event, context):
 """us-east-1 Kinesis Consumer Lambda → ap-northeast-1 Global Stream に転送"""
 records = []
 for record in event["Records"]:
  payload = base64.b64decode(record["kinesis"]["data"])
  records.append({
"Data": payload,
"PartitionKey": record["kinesis"]["partitionKey"]
  })

 if records:
  for i in range(0, len(records), 500):
target_kinesis.put_records(
 StreamName="global-centralized-logs",
 Records=records[i:i + 500]
)

S3 Cross-Region Replication（推奨パターン）

アーカイブ目的では S3 CRR がコスト効率最優。

resource "aws_s3_bucket_replication_configuration" "us_to_ap" {
  bucket = aws_s3_bucket.log_archive_us_east_1.id
  role= aws_iam_role.s3_replication.arn

  rule {
 id  = "replicate-to-global-archive"
 status = "Enabled"

 filter {
prefix = "logs/"
 }

 destination {
bucket  = "arn:aws:s3:::global-log-archive-ap-northeast-1"
storage_class = "STANDARD_IA"

replication_time {
  status = "Enabled"
  time {
 minutes = 15
  }
}

metrics {
  status = "Enabled"
  event_threshold {
 minutes = 15
  }
}
 }
  }
}

レプリケーションラグとコスト試算：

4.5 CloudFormation StackSetsによる一括展開

StackSetsをOrganizations統合と組み合わせることで、全Spoke AccountへのSubscription Filter設定を一括展開できます。

# stackset-subscription-filter.yaml
AWSTemplateFormatVersion: "2010-09-09"
Description: "Centralized Logging Subscription Filter - Deployed via StackSets"

Parameters:
  HubAccountId:
 Type: String
  DestinationName:
 Type: String
 Default: "centralized-log-destination"
  FilterPattern:
 Type: String
 Default: ""

Resources:
  AccountSubscriptionPolicy:
 Type: "AWS::Logs::AccountPolicy"
 Properties:
PolicyName: "centralized-logging-subscription"
PolicyDocument: !Sub |
  {
 "DestinationArn": "arn:aws:logs:${AWS::Region}:${HubAccountId}:destination:${DestinationName}",
 "FilterPattern": "${FilterPattern}",
 "Distribution": "ByLogStream"
  }
PolicyType: "SUBSCRIPTION_FILTER_POLICY"
Scope: "ALL"

StackSets展開コマンド（CLI）：

aws cloudformation create-stack-set--stack-set-name "centralized-logging-subscription"--template-body file://stackset-subscription-filter.yaml--parameters  ParameterKey=HubAccountId,ParameterValue=123456789012  ParameterKey=FilterPattern,ParameterValue=""--permission-model SERVICE_MANAGED--auto-deployment Enabled=true,RetainStacksOnAccountRemoval=false--capabilities CAPABILITY_IAM

aws cloudformation create-stack-instances--stack-set-name "centralized-logging-subscription"--deployment-targets OrganizationalUnitIds=["ou-xxxx-xxxxxxxx"]--regions ap-northeast-1 us-east-1 eu-west-1--operation-preferences  MaxConcurrentPercentage=20,FailureTolerancePercentage=10

4.6 Terraform IaC — Hub Account完全実装例

Hub Account Terraform（Kinesis + Firehose + S3 + Destination）：

module "centralized_logging_hub" {
  source = "./modules/centralized-logging/hub"

  environment = "production"
  hub_account_id = data.aws_caller_identity.current.account_id
  spoke_account_ids = var.spoke_account_ids

  kinesis_config = {
 shard_count= 4
 retention_period = 24
  }

  s3_config = {
 bucket_name = "org-log-archive-${var.environment}"
 lifecycle_rules = {
standard_ia_days = 30
glacier_days  = 90
expiration_days  = 2555
 }
  }

  firehose_config = {
 buffer_size  = 128
 buffer_interval = 300
 compression  = "GZIP"
 prefix = "year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
 error_prefix = "errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/"
  }
}

Spoke Account Terraform（Subscription Filter）：

module "centralized_logging_spoke" {
  source = "./modules/centralized-logging/spoke"

  hub_account_id= var.hub_account_id
  hub_region = "ap-northeast-1"
  destination_name = "centralized-log-destination"

  log_group_prefixes = [
 "/aws/lambda/",
 "/aws/ecs/",
 "/aws/apigateway/",
 "/app/"
  ]

  filter_pattern = ""
}

4.7 AWS CDK実装例 — TypeScript

import * as cdk from "aws-cdk-lib";
import * as kinesis from "aws-cdk-lib/aws-kinesis";
import * as s3 from "aws-cdk-lib/aws-s3";
import * as logs from "aws-cdk-lib/aws-logs";
import * as iam from "aws-cdk-lib/aws-iam";

interface CentralizedLoggingProps extends cdk.StackProps {
  environment: string;
  spokeAccountIds: string[];
}

export class CentralizedLoggingHubStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string, props: CentralizedLoggingProps) {
 super(scope, id, props);

 const logArchiveBucket = new s3.Bucket(this, "LogArchiveBucket", {
bucketName: `org-log-archive-${props.environment}`,
versioned: true,
encryption: s3.BucketEncryption.S3_MANAGED,
lifecycleRules: [
  {
 transitions: [
{
  storageClass: s3.StorageClass.INFREQUENT_ACCESS,
  transitionAfter: cdk.Duration.days(30),
},
{
  storageClass: s3.StorageClass.GLACIER_INSTANT_RETRIEVAL,
  transitionAfter: cdk.Duration.days(90),
},
 ],
 expiration: cdk.Duration.days(2555),
  },
],
 });

 const logStream = new kinesis.Stream(this, "CentralizedLogStream", {
streamName: `centralized-logs-${props.environment}`,
shardCount: 4,
retentionPeriod: cdk.Duration.hours(24),
 });

 const cwlToKinesisRole = new iam.Role(this, "CwlToKinesisRole", {
assumedBy: new iam.ServicePrincipal("logs.amazonaws.com"),
 });
 logStream.grantWrite(cwlToKinesisRole);

 new logs.CfnDestination(this, "LogDestination", {
destinationName: "centralized-log-destination",
roleArn: cwlToKinesisRole.roleArn,
targetArn: logStream.streamArn,
destinationPolicy: JSON.stringify({
  Version: "2012-10-17",
  Statement: [
 {
Effect: "Allow",
Principal: { AWS: props.spokeAccountIds },
Action: "logs:PutSubscriptionFilter",
Resource: `arn:aws:logs:${this.region}:${this.account}:destination:centralized-log-destination`,
 },
  ],
}),
 });
  }
}

4.8 アクセス制御設計 — Log Group Policy / PutLogEvents権限

Log Group Resource Policy

resource "aws_cloudwatch_log_resource_policy" "hub_log_group" {
  policy_name = "allow-cross-account-put-log-events"

  policy_document = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Sid = "AllowSpokeAccountsPutLogEvents"
  Effect = "Allow"
  Principal = {
 Service = "logs.amazonaws.com"
  }
  Action = [
 "logs:CreateLogGroup",
 "logs:CreateLogStream",
 "logs:PutLogEvents"
  ]
  Resource = "arn:aws:logs:ap-northeast-1:${var.hub_account_id}:log-group:/centralized/*:*"
  Condition = {
 StringEquals = {
"aws:PrincipalOrgID" = var.org_id
 }
  }
}
 ]
  })
}

PutLogEvents権限設計の原則：

CloudWatch Logs Cross-Account Observability（閲覧専用）

CloudWatch Logs Cross-Account Observability 機能を使うと、Spoke AccountのLog Groupを Hub Account のコンソールから直接閲覧できます。

# Hub Account (Monitoring Account) でSink作成
aws oam create-sink--name "centralized-observability-sink"--resource-types "AWS::Logs::LogGroup"

SINK_ARN="arn:aws:oam:ap-northeast-1:123456789012:sink/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

# 各Spoke Accountで実行
aws oam create-link--label-template "\$AccountName"--resource-types "AWS::Logs::LogGroup"--sink-identifier "$SINK_ARN"

この機能の制限：
– Log Group の 閲覧・クエリのみ（転送・永続化は別途Subscription Filterが必要）
– Logs InsightsのCross-Account queryもこの設定で有効化
– 1つのMonitoring Accountに接続できるSource Accountは 最大100アカウント

閲覧権限設計（最小権限原則）：

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Sid": "CentralizedLoggingReadOnly",
"Effect": "Allow",
"Action": [
  "logs:DescribeLogGroups",
  "logs:DescribeLogStreams",
  "logs:GetLogEvents",
  "logs:FilterLogEvents",
  "logs:StartQuery",
  "logs:GetQueryResults"
],
"Resource": [
  "arn:aws:logs:*:*:log-group:/centralized/*",
  "arn:aws:logs:*:*:log-group:/centralized/*:log-stream:*"
]
 }
  ]
}

4.9 Athena集約クエリ — Hub Accountでの全社横断分析

Glue Data Catalog + Athena構成

S3に蓄積されたログを Glue Data Catalog でテーブル化し、Athena でSQL横断クエリを実行します。Parquet変換を挟むことでクエリコストを大幅削減できます。

Glue Crawlerによる自動スキーマ検出：

resource "aws_glue_crawler" "log_archive" {
  name = "log-archive-crawler"
  database_name = aws_glue_catalog_database.logs.name
  role = aws_iam_role.glue_crawler.arn
  schedule= "cron(0 1 * * ? *)"

  s3_target {
 path = "s3://${aws_s3_bucket.log_archive.bucket}/year="
 exclusions = ["errors/**"]
  }

  configuration = jsonencode({
 Version = 1.0
 CrawlerOutput = {
Partitions = { AddOrUpdateBehavior = "InheritFromTable" }
 }
  })
}

代表的なAthena横断クエリ：

-- 全アカウント × 全リージョンのエラーログ集計（過去7日）
SELECT
 account_id,
 aws_region,
 log_group,
 COUNT(*) AS error_count,
 DATE_FORMAT(FROM_UNIXTIME(timestamp / 1000), '%Y-%m-%d')  AS log_date
FROM
 "log_archive_db"."centralized_logs"
WHERE
 year >= '2026'
 AND message LIKE '%ERROR%'
 AND FROM_UNIXTIME(timestamp / 1000) >= CURRENT_DATE - INTERVAL '7' DAY
GROUP BY
 account_id,
 aws_region,
 log_group,
 DATE_FORMAT(FROM_UNIXTIME(timestamp / 1000), '%Y-%m-%d')
ORDER BY
 error_count DESC
LIMIT 100;

-- アカウント別ログ量ランキング（Ingestion量多い順）
SELECT
 account_id,
 COUNT(*) AS log_lines,
 SUM(LENGTH(message)) / 1024 / 1024 AS total_mb
FROM
 "log_archive_db"."centralized_logs"
WHERE
 year = '2026' AND month = '05'
GROUP BY
 account_id
ORDER BY
 total_mb DESC;

ログ保管期間 Tiered Strategy と Athena の連携：

5. ログコスト最適化本番運用 — Retention戦略 × Tiered Storage × Logs Live Tail × Athena経由クエリ

5-1. CloudWatch Logs課金構造の全体像

CloudWatch Logsの料金体系は3つの課金軸で構成される。本番運用でコスト爆発が起きるケースのほぼ全てが、いずれか一つの軸に対する無策から発生する。

100 GiB/日のLogs Ingestion環境での月額試算:

• Ingestion: 100 GB × 30日 × $0.76 = $2,280/月
• Storage (Never Expire、12ヶ月後): 36,500 GB × $0.033 = $1,205/月
• Logs Insights: 日100 GB × 5クエリ × $0.0076 = $1,140/月
• 合計: ~$4,625/月 (年間 $55,500)

適切なRetention戦略 + Tiered Storage + Athena移行を実施すると:

• Ingestion: DEBUG/INFO除去で 30 GB/日 → $684/月
• Storage: Retention短縮 + IA移行で → $180/月
• Logs Insights: 短時間範囲+Athena移行で → $228/月
• 最適化後合計: ~$1,092/月 (約76%削減)

5-2. Retention Policy設定 — LogGroup別最適化戦略

ログクラス別最適Retention設定:

コスト試算例: Lambda/Application Logs合計 1 GB/日 の環境では、Never Expire 12ヶ月後は Storage 365 GB × $0.033 = $12.05/月、2年後は 730 GB × $0.033 = $24.09/月 (累積増加) となる。Retention 30日設定では Storage 30 GB × $0.033 = $0.99/月 (固定) で、12ヶ月後の削減効果は 92%、2年後は 96% 削減。

Terraform によるRetention一括設定:

resource "aws_cloudwatch_log_group" "lambda_app" {
  name  = "/aws/lambda/${var.function_name}"
  retention_in_days = 30

  tags = {
 LogClass = "application"
 Environment = var.environment
  }
}

resource "aws_cloudwatch_log_group" "audit" {
  name  = "/custom/audit/${var.service_name}"
  retention_in_days = 2555

  tags = {
 LogClass= "audit"
 Compliance = "SOC2-PCI-DSS"
  }
}

locals {
  log_group_retentions = {
 "/aws/lambda"  = 30
 "/aws/rds"  = 7
 "/aws/eks"  = 30
 "/aws/apigateway" = 30
 "/custom/audit"= 2555
  }
}

resource "aws_cloudwatch_log_group" "managed" {
  for_each = local.log_group_retentions
  name  = each.key
  retention_in_days = each.value
}

EventBridge + Lambda による新規LogGroupの自動Retention設定:

import boto3
import os

logs_client = boto3.client('logs')

RETENTION_MAP = {
 '/aws/lambda':int(os.environ.get('LAMBDA_RETENTION', 30)),
 '/aws/rds':int(os.environ.get('RDS_RETENTION', 7)),
 '/aws/eks':int(os.environ.get('EKS_RETENTION', 30)),
 '/custom/audit': int(os.environ.get('AUDIT_RETENTION', 2555)),
 'default': int(os.environ.get('DEFAULT_RETENTION', 90)),
}

def get_retention(log_group_name):
 for prefix, days in RETENTION_MAP.items():
  if prefix == 'default':
continue
  if log_group_name.startswith(prefix):
return days
 return RETENTION_MAP['default']

def handler(event, context):
 detail = event.get('detail', {})
 log_group_name = detail.get('requestParameters', {}).get('logGroupName', '')

 if not log_group_name:
  return {'statusCode': 400, 'body': 'No log group name found'}

 retention_days = get_retention(log_group_name)
 logs_client.put_retention_policy(
  logGroupName=log_group_name,
  retentionInDays=retention_days
 )
 print(f"Set retention {retention_days} days for {log_group_name}")
 return {'statusCode': 200}

EventBridgeルール (CreateLogGroup API コールをトリガー):

resource "aws_cloudwatch_event_rule" "new_log_group" {
  name  = "auto-retention-on-log-group-create"
  description = "Trigger Lambda when new CloudWatch Log Group is created"

  event_pattern = jsonencode({
 source= ["aws.logs"]
 detail-type = ["AWS API Call via CloudTrail"]
 detail = {
eventSource = ["logs.amazonaws.com"]
eventName= ["CreateLogGroup"]
 }
  })
}

resource "aws_cloudwatch_event_target" "retention_lambda" {
  rule= aws_cloudwatch_event_rule.new_log_group.name
  target_id = "RetentionSetterLambda"
  arn = aws_lambda_function.retention_setter.arn
}

resource "aws_lambda_permission" "allow_eventbridge" {
  statement_id  = "AllowEventBridge"
  action  = "lambda:InvokeFunction"
  function_name = aws_lambda_function.retention_setter.function_name
  principal  = "events.amazonaws.com"
  source_arn = aws_cloudwatch_event_rule.new_log_group.arn
}

#!/bin/bash
DEFAULT_RETENTION=90
AUDIT_PATTERN="/custom/audit"
LAMBDA_PATTERN="/aws/lambda"

aws logs describe-log-groups \
  --query 'logGroups[?retentionInDays==<code>null</code>].logGroupName' \
  --output text | tr '\t' '\n' | while read log_group; do

  if [[ "$log_group" == *"$AUDIT_PATTERN"* ]]; then
 retention=2555
  elif [[ "$log_group" == *"$LAMBDA_PATTERN"* ]]; then
 retention=30
  else
 retention=$DEFAULT_RETENTION
  fi

  aws logs put-retention-policy \
 --log-group-name "$log_group" \
 --retention-in-days $retention

  echo "Set retention $retention days: $log_group"
done

5-3. CloudWatch Logs Tiered Storage — Standard vs Infrequent Access

2023年11月にGA (Generally Available) となった CloudWatch Logs Infrequent Access (IA) Storage Class を活用することで、Standard Storage比10分の1のコストでログを長期保持できる。

Standard Storage vs IA Storage 完全比較:

mermaid02: ログTiered Storage判定フロー

flowchart LR
 A[ログ書き込み] --> B{アクセス頻度・用途}
 B -->|高頻度 リアルタイム処理必要\n0-30日| C[Hot: Standard Storage\nSubscription Filter / Live Tail 利用可]
 B -->|低頻度 クエリ偶発的\n30-90日| D[Warm: IA Storage\nストレージ 90% 削減]
 B -->|アーカイブ 規制対応\n90日+| E[Cold: S3 Glacier\nAthena経由クエリ]
 C -->|30日超過| D
 D -->|90日超過| E
 E --> F{クエリ必要?}
 F -->|Yes| G[Glue Crawler + Athena]
 F -->|No| H[Glacier Deep Archive\n最終アーカイブ]

IA Storage への移行 — Terraform設定:

IA StorageはLogGroup単位でLog Class設定する:

resource "aws_cloudwatch_log_group" "app_logs_ia" {
  name  = "/app/production/access-logs"
  retention_in_days = 90
  log_group_class= "INFREQUENT_ACCESS"

  tags = {
 StorageClass = "IA"
 Purpose= "long-term-storage"
  }
}

IA Storage LogGroupにはSubscription FilterとMetric Filterが設定不可。リアルタイム処理が必要なLogGroupはStandard Storageを維持すること。

IA Storage コスト試算例 (1 TB のログを90日保持): Standard Storage: 1,000 GB × $0.033 × 3ヶ月 = $99/月。IA Storage: 1,000 GB × $0.0033 × 3ヶ月 = $9.9/月。削減額: $89.1/月 (約90%削減)。Logs Insights クエリ回数が月10回以下なら IA が有利。

IA Storage 適用判断基準 (4要件を全て満たす場合に適用):

1. Subscription Filter 不要: リアルタイムLambda処理が不要
2. Metric Filter 不要: アラートトリガーに使わない
3. Live Tail 不要: リアルタイムデバッグ対象外
4. 参照頻度低: 月1-2回のAd-hocクエリのみ

S3 Lifecycle Policy との組み合わせによる自動Tiering:

resource "aws_s3_bucket_lifecycle_configuration" "log_archive" {
  bucket = aws_s3_bucket.log_archive.id

  rule {
 id  = "log-tiering"
 status = "Enabled"

 transition {
days = 30
storage_class = "STANDARD_IA"
 }

 transition {
days = 90
storage_class = "GLACIER"
 }

 transition {
days = 365
storage_class = "DEEP_ARCHIVE"
 }

 expiration {
days = 2555
 }
  }
}

5-4. Logs Live Tail — リアルタイムデバッグの実践

Logs Live Tail (2023年リリース) はCloudWatch Logs ConsoleまたはCLIからLogGroupのログをリアルタイムストリーミング表示する機能。インシデント対応時のデバッグ効率を大幅に向上させる。

Logs Live Tail の主要ユースケース:

Logs Live Tail コスト: Live Tailはセッション単位での課金 (約$0.01/分)。1時間セッション: 60分 × $0.01 = $0.60。インシデント対応 (4時間): 240分 × $0.01 = $2.40。リアルタイムデバッグ目的ではLogs Insights (スキャン量課金) より安価な場合が多い。

AWS CLI でLogs Live Tail (Filter Pattern付き):

aws logs start-live-tail \
  --log-group-identifiers \
 "arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/lambda/my-function" \
  --log-event-filter-pattern "ERROR" \
  --output json

Console上のFilter Pattern例:

Live Tail の制限事項:

5-5. Glue Table経由Athenaクエリ — 長期ログのS3移行と低コスト分析

90日以上保持する大量ログはCloudWatch Logsに格納し続けるとStorage費が累積する。S3 Export + Glue + Athena パイプラインに移行することで、ストレージコストを1/10以下に削減しつつ、SQLクエリでの分析を維持できる。

パイプライン全体アーキテクチャ: CloudWatch Logs → S3 Export → Glue Crawler → Glue Data Catalog → Athena の順で構成する。Export後はS3 Lifecycle PolicyでGlacier移行まで自動化し、Athena側でPartition Projectionを使って長期データへの高速クエリを実現する。

STEP 1: CloudWatch Logs → S3 Export

aws logs create-export-task \
  --task-name "daily-log-export-$(date +%Y%m%d)" \
  --log-group-name "/aws/lambda/my-function" \
  --from $(date -v-1d +%s)000 \
  --to $(date +%s)000 \
  --destination "my-log-archive-bucket" \
  --destination-prefix "cloudwatch-logs/lambda/my-function/$(date +%Y/%m/%d)"

Lambda による自動日次Export:

import boto3
import time
from datetime import datetime, timedelta, timezone

logs_client = boto3.client('logs')

LOG_GROUPS = ['/aws/lambda/my-function', '/app/production/api']
EXPORT_BUCKET = 'my-log-archive-bucket'

def handler(event, context):
 jst = timezone(timedelta(hours=9))
 yesterday = datetime.now(jst) - timedelta(days=1)
 date_str = yesterday.strftime('%Y/%m/%d')
 from_time = int(yesterday.replace(
  hour=0, minute=0, second=0, microsecond=0).timestamp() * 1000)
 to_time = int(yesterday.replace(
  hour=23, minute=59, second=59, microsecond=0).timestamp() * 1000)

 for log_group in LOG_GROUPS:
  prefix = f"cloudwatch-logs{log_group}/{date_str}"
  response = logs_client.create_export_task(
taskName=f"export{log_group.replace('/', '-')}-{date_str.replace('/', '-')}",
logGroupName=log_group,
fromTime=from_time,
to=to_time,
destination=EXPORT_BUCKET,
destinationPrefix=prefix
  )
  task_id = response['taskId']

  for _ in range(30):
status = logs_client.describe_export_tasks(taskId=task_id)
state = status['exportTasks'][0]['status']['code']
if state == 'COMPLETED':
 break
elif state in ['FAILED', 'CANCELLED']:
 raise Exception(f"Export failed for {log_group}: {state}")
time.sleep(10)

STEP 2: Glue Crawler によるスキーマ自動検出

resource "aws_glue_catalog_database" "logs_db" {
  name = "cloudwatch_logs_archive"
}

resource "aws_glue_crawler" "cloudwatch_logs" {
  name = "cloudwatch-logs-crawler"
  role = aws_iam_role.glue_crawler.arn
  database_name = aws_glue_catalog_database.logs_db.name
  schedule= "cron(0 2 * * ? *)"

  s3_target {
 path = "s3://${var.log_archive_bucket}/cloudwatch-logs/"
  }

  schema_change_policy {
 delete_behavior = "LOG"
 update_behavior = "UPDATE_IN_DATABASE"
  }
}

STEP 3: Partition設計 (date/hour) によるクエリコスト最適化

CREATE EXTERNAL TABLE cloudwatch_logs.lambda_logs (
  `timestamp`  BIGINT,
  `message` STRING,
  `log_stream` STRING,
  `owner`STRING
)
PARTITIONED BY (
  `year`  STRING,
  `month` STRING,
  `day`STRING,
  `hour`  STRING
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://my-log-archive-bucket/cloudwatch-logs/lambda/my-function/'
TBLPROPERTIES ('has_encrypted_data' = 'false');

ALTER TABLE cloudwatch_logs.lambda_logs
ADD PARTITION (year='2026', month='05', day='17', hour='12')
LOCATION 's3://my-log-archive-bucket/cloudwatch-logs/lambda/my-function/2026/05/17/12/';

STEP 4: Athena SQL クエリ例

-- 直近7日のERROR集計 (Partition活用でスキャン最小化)
SELECT
  year, month, day,
  COUNT(*) AS error_count
FROM cloudwatch_logs.lambda_logs
WHERE year = '2026'
  AND month = '05'
  AND day BETWEEN '11' AND '17'
  AND message LIKE '%ERROR%'
GROUP BY year, month, day
ORDER BY day DESC;

-- エラーパターン分析 (正規表現)
SELECT
  regexp_extract(message, '"error":"([^"]+)"', 1) AS error_type,
  COUNT(*) AS cnt,
  MAX(timestamp) AS last_occurrence
FROM cloudwatch_logs.lambda_logs
WHERE year = '2026' AND month = '05'
  AND message LIKE '%"error"%'
GROUP BY 1
ORDER BY cnt DESC
LIMIT 20;

-- 時間別ログ量推移 (キャパシティプランニング用)
SELECT
  hour,
  COUNT(*) AS log_count,
  SUM(LENGTH(message)) / 1024 / 1024 AS total_mb
FROM cloudwatch_logs.lambda_logs
WHERE year = '2026' AND month = '05' AND day = '17'
GROUP BY hour
ORDER BY hour ASC;

CloudWatch Logs Insights vs Athena コスト比較表:

5-6. 月額50%削減ケーススタディ

初期状態 (最適化前): マイクロサービス10サービス × 各10 GiB/日

最適化施策 (3フェーズ):

Phase 1 — Ingestion削減 (即時効果): Lambda関数のDEBUG/INFO Logsを10% Samplingに変更し、構造化Logs (JSON minify + gzip) で圧縮。結果: 100 GiB/日 → 25 GiB/日 (75%削減)。Ingestion節約: $2,280 → $570 (月$1,710削減)。

Phase 2 — Storage最適化 (30日後効果): Lambda: Retention 7日、Application: Retention 30日、Audit: Retention 2555日 + IA Storage移行。結果: 累積3,650 GB → 600 GB相当。Storage節約: $120 → $20 (月$100削減)。

Phase 3 — Insights最適化 (即時効果): クエリ時間範囲を過去30日→過去1日に短縮、90日超クエリをAthena移行、Saved Queries再利用徹底。Insights節約: $685 → $115 (月$570削減)。

最適化後合計: $705/月 (初期 $3,085 → 77%削減)

コスト監視Dashboard 必須メトリクス:

6. 詰まりポイント7選 — CloudWatch Logs本番運用の地雷とフィックス

本番環境で CloudWatch Logs を運用すると、ドキュメントには書かれていない落とし穴に頻繁に遭遇する。以下の7つは実際のプロジェクトで繰り返し発生するパターンで、「なぜ詰まるか」「どう解くか」の2段構成で解説する。

詰まり1: Logs Insights クエリが30分以上完了しない

なぜ詰まるか: 時間範囲を1週間以上に設定した状態で stats count(*) を実行すると、CloudWatch Logs の逐次スキャンアーキテクチャでは数十GB超のスキャンが発生し著しく遅延する。stats コマンドの前に filter を置かない場合、全件に集計処理が走りスキャン量がさらに増加する。複数 LogGroup の横断クエリも重なるとタイムアウトになりやすい。

どう解くか:
1. 時間範囲を最初は15〜30分に絞り、クエリが動くことを確認してから徐々に拡張する
2. filter コマンドを stats より前に配置してスキャン前に件数を絞り込む（これだけで50%削減できるケースが多い）
3. bin(5m) で時間バケット集計しサンプリング的に件数を減らす
4. 複数 LogGroup 横断クエリは 1 つずつ動作確認してからマルチ指定に変更する

# OK: filter 先頭でスキャン量を削減 (高速 + 低コスト)
fields @timestamp, @message, level
| filter level = "ERROR"
| stats count(*) as error_count by bin(5m)
| sort error_count desc

詰まり2: Filter Pattern が意図しないマッチをする

なぜ詰まるか: CloudWatch Logs Filter Pattern はスペース区切りのトークンマッチング・JSON属性指定・数値比較が混在する独自構文で、直感に反する動作が起きやすい。ERROR と記述するだけで「ERROR を含む行を全て」マッチするため、WARN-ERROR-NOTICE のような文字列でも誤マッチする。大文字小文字も区別するため error では ERROR がヒットしない。

どう解くか:
1. JSON 構造化ログを採用し { $.level = "ERROR" } で厳密な属性指定マッチを使用する（最も確実）
2. テスト専用の LogGroup で Filter Pattern 検証を行ってから本番 LogGroup に適用する
3. テキストログの場合はダブルクォートで完全一致: "[ERROR]" 形式で囲む
4. CLF / Apache 形式のアクセスログには positional matching: [ip, -, -, timestamp, -, status>=400, bytes]

詰まり3: Subscription Filter 上限 (1LogGroup 2個) に到達

なぜ詰まるか: CloudWatch Logs は 1 つの LogGroup につき Subscription Filter を最大 2 個しか設定できない（Account-level subscription と合計して上限）。リアルタイム Lambda 通知 + Firehose 転送 + セキュリティ分析 Lambda の3系統を設定しようとして LimitExceededException: Subscription filter limit exceeded が発生する。この制限を知らずにアーキテクチャを設計すると後で大きな変更を強いられる。

どう解くか:
1. Lambda 集約 Fan-out: 1 つの Lambda がリクエストを受け取り、複数の Downstream（SNS / SQS / Firehose）へ配信する（シンプルだが Lambda 障害で全 Downstream が停止するリスクがある）
2. Kinesis Data Streams + Enhanced Fan-out: Subscription Filter を 1 本の Kinesis Stream に集約し、複数の独立した Lambda Consumer が各自でデータを消費する（推奨・Consumer 間の障害が独立）
3. EventBridge Pipes 経由 Fan-out: Kinesis → EventBridge Pipes → 複数ターゲットへのフィルタリング可能ルーティング（複雑なルーティング要件に最適）

詰まり4: Cross-Account Subscription 権限エラー（無音で失敗）

なぜ詰まるか: Cross-Account Logs Subscription は 3 つのポリシーが全て揃わないと無音で失敗する。エラーメッセージが CloudWatch Logs コンソールに出力されないため原因特定に時間がかかる。送信元アカウントの Filter Role 設定だけでは不十分な点が最大の盲点になりやすい。

どう解くか: 以下の 3 点セットを順番に確認する。

1. Destination Policy（受信側）: logs.amazonaws.com からの logs:PutSubscriptionFilter を許可する Resource Policy
2. IAM Cross-Account Trust（受信側）: 送信元 Account からの AssumeRole を許可する Trust Relationship
3. Filter Role（送信側）: logs.<region>.amazonaws.com が AssumeRole でき、kinesis:PutRecord 権限を持つ IAM Role

{
  "Version": "2012-10-17",
  "Statement": [{
 "Effect": "Allow",
 "Principal": { "Service": "logs.ap-northeast-1.amazonaws.com" },
 "Action": "logs:PutSubscriptionFilter",
 "Resource": "arn:aws:logs:ap-northeast-1:DEST_ACCOUNT_ID:destination:CentralHub",
 "Condition": {
"StringLike": {
  "aws:SourceArn": "arn:aws:logs:ap-northeast-1:SOURCE_ACCOUNT_ID:*"
}
 }
  }]
}

詰まり5: ログコスト爆発（月額1万ドル超）

なぜ詰まるか: 開発環境の DEBUG ログ設定を本番に持ち込み、Retention を Never Expire のままにすると、Ingestion 費と Storage 費が複利で膨張する。Lambda が毎秒 1000 リクエストを処理するサービスでは DEBUG ログだけで月額数十万円超になる事例が頻発している。

どう解くか:
1. Lambda 内ログレベル制御: 本番は INFO 以上のみ出力。DEBUG は環境変数 LOG_LEVEL=INFO で無効化する
2. DEBUG Sampling: 本番で DEBUG が必要な場合は 1〜5% サンプリングで Ingestion を最大 80% 削減する
3. LogGroup 別 Retention 戦略: DEBUG=7日 / アプリ INFO=90日 / 監査ログ=2555日（7年）を必ず設定する
4. 90日以上は Athena 移行: S3 Export して Glue カタログ登録 → Athena 経由クエリで月額 50% 削減可能

詰まり6: ログ欠損・遅延（Kinesis Firehose / Lambda 設定ミス）

なぜ詰まるか: Lambda がタイムアウトまたは同期エラーで強制終了すると、そのリクエストの CloudWatch Logs への書き込みが未完了のまま欠損する。Kinesis Firehose のデフォルトバッファ設定（5分 / 5MB）ではリアルタイム通知が必要な用途で遅延が発生する。Firehose 配信失敗時に S3 Error Bucket が設定されていないとログがサイレントに消える点が見落とされやすい。

どう解くか:
1. Lambda DLQ 設定: SQS DLQ を設定し、失敗したイベントを捕捉して再送処理を実装する
2. Firehose バッファ短縮: リアルタイム重視の場合は Buffer Interval=60秒 / Buffer Size=1MB に変更する
3. Firehose Error Output 設定: S3 Error Bucket を必ず設定し、配信失敗ログを保全する
4. DeliveryStatus 監視: CloudWatch Metrics の DeliveryThrottling / DeliveryErrors に CloudWatch Alarm を設定する

詰まり7: CloudWatch Logs Export 制限（1日1タスク制限）

なぜ詰まるか: CloudWatch Logs から S3 への create-export-task API は 1 LogGroup あたり同時実行タスクが 1 つに制限されており、前のタスクが完了するまで次のタスクを開始できない。過去ログの大量移行をこの API で自動化しようとすると処理上限に引っかかる。また S3 バケットポリシーに logs.amazonaws.com の許可がないと Export が無音で失敗するという二重の罠がある。

どう解くか:
1. 継続的転送は Kinesis Firehose 経由: 制限なしでリアルタイム S3 転送。新規ログは全て Firehose 経由にする
2. 過去ログ大量 Export は直列バッチ化: 各 LogGroup の Export タスクが完了してから次を実行する直列処理に変更する
3. S3 バケットポリシーの事前確認: logs.amazonaws.com Principal に s3:PutObject を許可するポリシーをインフラコードに必ず含める

7. アンチパターン→正解パターン変換演習5問

本番運用でよく見かけるアンチパターンを「現状の問題点」「正解への移行手順」「Trade-off」の3段構成で解説する。Terraform / Python / SQL の実コード例を交えた実践的な内容にしている。

演習1: 全LogGroup Retention Never Expire → LogGroup別Retention戦略

現状の問題点: アカウント内の全 LogGroup が Retention Never Expire のまま運用されている。Lambda の実行ログが年単位で蓄積し、Storage 費が月を追うごとに増加し続ける。特に頻繁に呼び出される Lambda のログが数TB に達する事例も多い。

正解への移行手順:
1. Retention が設定されていない LogGroup を一覧表示する:

aws logs describe-log-groups \
  --query 'logGroups[?!retentionInDays].[logGroupName]' \
  --output text

1. 用途別 Retention ポリシーを設計する（DEBUG=7日 / アプリ INFO=90日 / 監査・コンプライアンス=2555日）
2. Terraform で LogGroup 別 Retention を設定する:

resource "aws_cloudwatch_log_group" "app_info" {
  name  = "/aws/lambda/my-app"
  retention_in_days = 90
  tags = { Environment = "production", Type = "app" }
}

resource "aws_cloudwatch_log_group" "audit" {
  name  = "/app/audit"
  retention_in_days = 2555
  tags = { Environment = "production", Type = "audit" }
}

1. 既存 LogGroup には aws logs put-retention-policy CLI で一括適用し、CI/CD パイプラインで管理する

Trade-off: Retention を短縮するとログが自動削除されるため、長期障害調査やコンプライアンス監査で履歴参照できなくなる。監査対象ログは短縮禁止とし、重要ログは 90 日以前に S3 Export して保全しておく。

演習2: 1LogGroupに3つ目のSubscription Filter追加 → Kinesis Data Streams Enhanced Fan-out

現状の問題点: /aws/lambda/payment-api LogGroup に Lambda 通知 Filter（1個目）と Firehose 転送 Filter（2個目）が設定されており、セキュリティ監査用 Lambda（3個目）を追加しようとして LimitExceededException: Subscription filter limit exceeded エラーが発生している。

正解への移行手順:
1. 既存の Subscription Filter を 1 本の Kinesis Data Stream への Subscription に変更する
2. Kinesis Data Stream に Enhanced Fan-out で複数 Consumer を登録する:

# Enhanced Fan-out Consumer を登録
aws kinesis register-stream-consumer \
  --stream-arn arn:aws:kinesis:ap-northeast-1:ACCOUNT_ID:stream/payment-logs-stream \
  --consumer-name payment-alert-consumer

aws kinesis register-stream-consumer \
  --stream-arn arn:aws:kinesis:ap-northeast-1:ACCOUNT_ID:stream/payment-logs-stream \
  --consumer-name firehose-delivery-consumer

1. 各 Consumer（Lambda 通知 / Firehose 転送 / セキュリティ分析 Lambda）を独立して設定する（Consumer1 件ごとに 2MB/秒の独立スループットが保証される）

Trade-off: Kinesis Data Streams のシャード費用（0.015USD/シャード/時間）が追加発生する。10 アカウント未満の小規模な場合は Lambda 内での複数 Downstream 配信（1 Lambda → 複数 SNS/SQS）のほうがコスト効率が良い。

演習3: Cross-Account Logs個別調査 → Hub Account Centralized Logging

現状の問題点: 15 の AWS アカウントがあり、障害発生時に各アカウントの Logs Insights を個別に調査している。手動での横断調査に 1 時間以上かかり、障害対応の MTTR が悪化している。

正解への移行手順:
1. Organizations の Log Archive Account（または専用 Observability Account）を Hub Account として指定する
2. 各 Source Account から Cross-Account Subscription を設定する（Destination → Kinesis Data Stream in Hub Account）
3. Hub Account の Kinesis → Firehose → S3 → Glue Crawler → Athena で全アカウントログを横断クエリ可能にする
4. Hub Account の Logs Insights で全アカウント横串クエリを実行する:

# Hub Account での Logs Insights 横串クエリ例
fields @timestamp, @logStream, @message
| filter level = "ERROR"
| stats count(*) by @logStream, bin(5m)
| sort count desc
| limit 20

Trade-off: Hub Account への Kinesis Data Streams 料金と Firehose 転送費が追加発生する。しかし横断可視化による MTTR 改善と運用工数削減の効果が大きく、20 アカウント以上では投資対効果が明確にプラスになる。

演習4: Logs Insights 90日範囲クエリ → 直近14日Insights + 古データはAthena

現状の問題点: コンプライアンス調査で 90 日前のアクセスログを Logs Insights でクエリしている。1 クエリのスキャン量が 50GB 超になり、クエリ費用が 1 回あたり 25USD 超かかっている。さらにタイムアウトで結果が取得できないケースも発生している。

正解への移行手順:
1. 90 日以上前のログは S3 に自動 Export するパイプラインを構築する（Firehose → S3 → Glacier 移行）
2. S3 のログに Glue Crawler を適用して Athena Table を作成する（パーティション: year / month / day / account）
3. 古いデータは Athena 経由クエリに切り替える:

SELECT timestamp, account_id, source_ip, action
FROM access_logs
WHERE year = '2025' AND month = '02'
  AND day BETWEEN '01' AND '28'
  AND status_code >= 400
ORDER BY timestamp DESC
LIMIT 1000;

1. Logs Insights は直近 14 日以内の調査に特化させ、それ以前は Athena 経由とするルールを運用手順書に明記する

Trade-off: 二重のクエリ経路を管理する複雑性が増す。Glue Crawler の定期実行費用と S3 Storage 費が追加発生するが、Logs Insights 直接クエリより 70〜90% のコスト削減効果がある。

演習5: DEBUG Logs全保持（Ingestion費膨張）→ Lambda内Sampling + 構造化ログ

現状の問題点: Lambda のログレベルを rootLogger=DEBUG に設定し、全てのリクエストで DEBUG ログを出力している。1 日の Ingestion 量が 100GB を超え、月額の Ingestion 費だけで 30 万円超になっている。

正解への移行手順:
1. ログレベルを環境変数で制御し、本番は INFO または WARN に変更する
2. DEBUG ログが必要な場合はサンプリングを実装する:

import os, random, logging

LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
DEBUG_SAMPLE = float(os.environ.get("DEBUG_SAMPLE_RATE", "0.01"))

logger = logging.getLogger()
logger.setLevel(getattr(logging, LOG_LEVEL))

def handler(event, context):
 if LOG_LEVEL == "DEBUG" and random.random() > DEBUG_SAMPLE:
  logger.setLevel(logging.INFO)
 logger.info("request received",
 extra={"request_id": context.aws_request_id})
 logger.debug("event payload: %s", event)

1. JSON 構造化ログ（Lambda Powertools 等）を採用してフィールド指定クエリでスキャン量を削減する
2. POWERTOOLS_LOGGER_SAMPLE_RATE=0.01 環境変数で DEBUG サンプリングを設定管理する

Trade-off: DEBUG サンプリングにより一部のリクエストのみ詳細ログが残る。バグ調査時に DEBUG ログが取得できないケースが発生するため、Correlation ID を必ず付与しトレースと組み合わせて代替可視性を確保する。

8. まとめ + Observability Vol1↔Vol2双方向リンク + 落とし穴10選 + 全クロスリンク

Vol2 完遂で習得する4本柱

本記事 Vol2 で習得した4本柱を振り返る。

1. Logs Insights 本番運用: Query Language 習熟から可視化・Saved Queries・パフォーマンス最適化まで。全件スキャン回避と filter 先頭配置の2原則がコストと速度を決める。

2. Lambda 統合パターン: Subscription Filter × Kinesis Data Streams × Enhanced Fan-out による柔軟なログルーティング。DLQ + Error Bucket + DeliveryStatus 監視の3点セットで欠損ゼロを実現する。

3. Centralized Logging: Hub Account 集約 × Cross-Account Subscription × Organizations 統合。Athena 横串クエリで全アカウントのログを 1 か所から分析し、MTTR を大幅に短縮する。

4. ログコスト最適化: Retention 戦略 × DEBUG サンプリング × Tiered Storage × 90 日超 Athena 移行。4 施策の組み合わせで月額 50% 以上の削減が現実的に達成できる。

落とし穴10選

1. Retention Never Expire 全体デフォルト: 放置すると年率で Storage 費が倍増。LogGroup 作成時の Terraform 必須項目に retention_in_days を設定せよ。
2. Subscription Filter 上限 (2/LogGroup) を把握せずに設計: 3 系統目が必要になった時点で設計変更を強いられる。最初から Kinesis Fan-out で設計しておくと拡張性が高い。
3. Cross-Account Subscription の3点ポリシー漏れ: Destination Policy / IAM Trust / Filter Role のいずれか 1 つでも欠けると無音で失敗する。チェックリストで 3 点必ず確認する。
4. Logs Insights 時間範囲無制限クエリ: 1 週間分の全件スキャンは 1 クエリ数十 USD 超になる。「時間範囲 = コスト」という意識を全員が持つ。
5. Export タスク制限を知らずにバッチ設計: 1 LogGroup 同時 1 タスク制限。大量 Export 時は Firehose 常時転送に変更するのが根本解決。
6. Firehose Error Bucket 未設定: 配信失敗ログがサイレントに消える。本番構築時に必ず設定する。
7. Lambda DEBUG ログを本番環境で全出力: Ingestion 費が急増するだけでなく、機微情報が誤混入するリスクもある。環境変数制御必須。
8. Filter Pattern の大文字小文字区別の誤解: デフォルトは大文字小文字を区別する。ERROR と error は別マッチ。JSON 属性指定を使えば区別の問題を回避できる。
9. Logs Insights の Cross-Log-Group クエリ上限 (50 LogGroup): 横断クエリで 50 グループ超を指定するとエラー。大規模環境では Centralized Logging で集約が必須。
10. S3 Export バケットポリシー忘れ: logs.amazonaws.com からの s3:PutObject 許可がないと Export タスクが作成直後に失敗する。インフラコードにバケットポリシーを必ず含める。

Observability実践 Vol1 — 分散トレース実践編 (X-Ray / Application Signals / ADOT)

Observability本番運用 Vol2 — CloudWatch Logs 深掘り編 (Logs Insights × Lambda統合 × Centralized × コスト最適化) ← 本記事

Observability Vol3 — Application Signals × SLO/SLI × Service Map × CodeGuru Profiler