会社情報
採用情報
お問い合わせ
Article

AWS EventBridgeで構築するイベント駆動アーキテクチャ

高田晃太郎
AWS EventBridgeで構築するイベント駆動アーキテクチャ

イベントサイズ上限256KB、少なくとも一度の配信、数百ミリ秒台の低レイテンシでスケールするというAWS公式ドキュメントに記載の性質を持つEventBridgeは、同期API連携のボトルネックやスパイク耐性の弱さを抱えるシステムに、現実的な打開策をもたらす¹²³⁴。一般に、EventBridgeはマネージドな再試行とバックプレッシャー吸収により、短時間の障害やバーストを受け流しやすい特性がある。また、同期連携中心の設計からの転換は、サービス間の独立性と障害の波及抑制に直結し、DORA指標で語られるデリバリー速度と変更容易性の改善にもつながりやすい。ビジネス的には、新機能の着手からリリースまでのリードタイム短縮や、SaaSとのネイティブ連携による機会損失の低減が期待できる。ここでは中級から上級のCTO・エンジニアリーダーを想定しつつ、初学者でも追えるよう要点を補足しながら、AWS EventBridgeを核にしたイベント駆動アーキテクチャを、設計、実装、運用、ガバナンスの順に深掘りする。

事業価値につなげる設計原則とEventBridgeの特性

イベント駆動の核心は、意図（命令）を直接呼び出すのではなく、事実（何が起きたか）を公表する点にある。サービス間の同期的な結合を断ち、変更の独立性を確保することで、機能の追加やロールバックを個別に行えるようになる。EventBridgeはイベントバス、ルール、ターゲットというシンプルな構成で、プロデューサーとコンシューマーの間に緩衝層を設ける。イベントはJSONで、最大256KBまで許容され、少なくとも一度（at-least-once）配信される¹²。at-least-onceは「重複はありうるが、届かないよりは複数回届くことを優先する」配信モデルであり、重複はアプリ側の冪等化（同じイベントを何度処理しても結果が変わらないようにする設計）で吸収する。順序性が必要な場合はスキーマ側での冪等化や、必要に応じてFIFOなキュー（到着順を保証するキュー）を併用するのが実務的だ⁵。

SNSやSQSと比較したとき、EventBridgeはパターンマッチングによるルーティング⁹や、SaaS/パートナーイベントの取り込み¹、アーカイブとリプレイ⁶、スキーマレジストリ⁷といった開発体験の層が厚い。SQSはバックプレッシャー制御とコンシューマー主導のポーリングに強みがあり、SNSはファンアウトの手軽さが魅力だ。EventBridgeはこれらの要素を包括しつつ、ルールでフィルタリングし、ターゲットごとにリトライやDLQ（Dead-Letter Queue。恒常的に処理できないイベントの待避先）を設定できるため、疎結合な分岐と段階的導入に向いている²。結果として、既存システムに負担をかけずにイベント連携を挿入し、影響範囲を最小化しながら能力を拡張できる⁵。

クロスアカウントやクロスリージョンの連携が必要なエンタープライズでは、イベントバスポリシーで発行元のプリンシパルを制御し、検証済みのスキーマに準拠したイベントのみをルーティングする方針が堅牢だ¹。これにより、プラットフォームチームがガードレールを提供し、各プロダクトチームは自律的にイベントを発行・購読できる。最終的に、データメッシュやドメイン駆動設計の境界と整合したイベント設計が、組織スケールでの開発速度に寄与する。

イベントの粒度と命名規約が変更容易性を左右する

「何が起きたか」を表すドメインイベントは、UIや内部実装の細部から切り離し、ビジネス語彙で命名する。例えばOrderCreatedやInvoiceIssuedのような名前は、下位の集約が追加されても意味を保ちやすい。イベントは不変であり、破壊的変更を避け、必要に応じてバージョニングを行う。こうした基本が、ダイナミックな機能増減の現場で効いてくる。実装者はスキーマレジストリ（イベントの型定義を集中管理する仕組み）を活用し、型の契約を明文化しておくとよい⁷。

実装の中核：CDKとSDKで進める堅牢な基盤づくり

まずは最小構成としてカスタムイベントバス、フィルタルール、Lambdaターゲット、そして障害時のDLQを備えた構成を用意する。インフラはCDKで定義し、アプリケーションはAWS SDKを用いたプロデューサーと、スキーマ検証と冪等化を備えたコンシューマーで実装する。以下では完全なコード例を順に示す。

CDK（TypeScript）：イベントバス、ルール、ターゲット、DLQ

import * as cdk from 'aws-cdk-lib';
import { Stack, StackProps, Duration } from 'aws-cdk-lib';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as events from 'aws-cdk-lib/aws-events';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as iam from 'aws-cdk-lib/aws-iam';

export class EdaStack extends Stack {
  constructor(scope: cdk.App, id: string, props?: StackProps) {
    super(scope, id, props);

    const bus = new events.EventBus(this, 'AppBus', {
      eventBusName: 'app-bus',
    });

    const dlq = new sqs.Queue(this, 'ConsumerDlq', {
      queueName: 'consumer-dlq',
      retentionPeriod: Duration.days(14),
    });

    const fn = new lambda.Function(this, 'ConsumerFn', {
      runtime: lambda.Runtime.NODEJS_18_X,
      code: lambda.Code.fromInline(`exports.handler = async (e) => { console.log(JSON.stringify(e)); }`),
      handler: 'index.handler',
      timeout: Duration.seconds(10),
    });

    const rule = new events.Rule(this, 'OrderEventsRule', {
      eventBus: bus,
      eventPattern: {
        source: ['com.example.orders'],
        detailType: ['OrderCreated'],
      },
    });

    rule.addTarget(new targets.LambdaFunction(fn, {
      deadLetterQueue: dlq,
      maxEventAge: Duration.hours(2),
      retryPolicy: { maxRetryAttempts: 185, maximumEventAge: Duration.hours(24) },
    }));

    const producerRole = new iam.Role(this, 'ProducerRole', {
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
    });

    bus.grantPutEventsTo(producerRole);
  }
}

この定義により、指定したソースとディテールタイプだけをLambdaへ配信し、失敗時はDLQに退避できる。再試行は指数バックオフで行われ、最大24時間の再試行ウィンドウを活用できる設計だ²。プロデューサーへはbus.grantPutEventsToで最小権限を付与し、バス外への誤送信を防ぐ。

イベントプロデューサー（Node.js, AWS SDK v3）：PutEventsの完全例

import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge';

const client = new EventBridgeClient({ region: process.env.AWS_REGION });

export async function publishOrderCreated(order: { id: string; userId: string; total: number }) {
  const detail = {
    orderId: order.id,
    userId: order.userId,
    total: order.total,
    occurredAt: new Date().toISOString(),
  };

  const cmd = new PutEventsCommand({
    Entries: [{
      EventBusName: 'app-bus',
      Source: 'com.example.orders',
      DetailType: 'OrderCreated',
      Detail: JSON.stringify(detail),
    }],
  });

  const res = await client.send(cmd);
  if ((res.FailedEntryCount ?? 0) > 0) {
    const err = res.Entries?.find(e => e.ErrorCode);
    throw new Error(`PutEvents failed: ${err?.ErrorCode} - ${err?.ErrorMessage}`);
  }
  return res.Entries?.[0]?.EventId;
}

プロデューサー側では失敗数を必ず確認し、部分的な不達を検知したらログとメトリクスを計上する。イベントIDは監査や相関のキーとして保存しておくと、後段のトレースに役立つ。なお、PutEventsは1リクエストで最大10件のイベントをバッチ送信できる⁸。

コンシューマー（Python Lambda）：スキーマ検証と冪等化

import json
import os
import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timezone

DDB_TABLE = os.environ.get('IDEMPOTENCY_TABLE')
ddb = boto3.resource('dynamodb').Table(DDB_TABLE)

class ValidationError(Exception):
    pass

def validate(payload: dict) -> None:
    required = ['orderId', 'userId', 'total', 'occurredAt']
    for key in required:
        if key not in payload:
            raise ValidationError(f'missing: {key}')
    if not isinstance(payload['total'], (int, float)):
        raise ValidationError('total must be number')

def is_duplicate(event_id: str) -> bool:
    try:
        res = ddb.put_item(
            Item={'pk': event_id, 'ts': datetime.now(timezone.utc).isoformat()},
            ConditionExpression='attribute_not_exists(pk)'
        )
        return False
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            return True
        raise

def handler(event, context):
    records = event.get('Records') or []
    # EventBridge invokes Lambda with 'detail' at top-level for each event
    detail = event.get('detail')
    if detail:
        records = [event]

    for r in records:
        eid = r.get('id') or r.get('detail', {}).get('eventId')
        payload = r.get('detail') or {}
        try:
            validate(payload)
            if eid and is_duplicate(eid):
                print(f'duplicate event: {eid}')
                continue
            process(payload)
        except ValidationError as ve:
            print(f'bad event: {ve} - {json.dumps(r)[:512]}')
        except Exception as ex:
            print(f'unexpected error: {ex}')
            raise


def process(payload: dict) -> None:
    print('processing', json.dumps(payload))

この例ではEventBridgeのイベントIDを冪等性キーに用い、DynamoDBで重複を排除している。スキーマの必須項目検証を行い、想定外のデータは例外としてDLQに流す設計が運用では安全だ。イベントごとのトレースIDをCloudWatch Logsとメトリクスに記録すれば、遅延や再試行の可視化が容易になる⁴。

Terraform：ルールとターゲットのIaC定義

provider "aws" {
  region = var.region
}

resource "aws_cloudwatch_event_bus" "app" {
  name = "app-bus"
}

resource "aws_cloudwatch_event_rule" "orders" {
  name        = "orders-created"
  event_bus_name = aws_cloudwatch_event_bus.app.name
  event_pattern = jsonencode({
    source = ["com.example.orders"],
    "detail-type" = ["OrderCreated"]
  })
}

resource "aws_sqs_queue" "dlq" {
  name = "consumer-dlq"
}

resource "aws_cloudwatch_event_target" "lambda" {
  rule      = aws_cloudwatch_event_rule.orders.name
  event_bus_name = aws_cloudwatch_event_bus.app.name
  arn       = aws_lambda_function.consumer.arn
  dead_letter_config {
    arn = aws_sqs_queue.dlq.arn
  }
}

CDKとTerraformを併用する場合でも、イベントバス名とルール名の命名規則を統一し、監査や運用での検索性を高めるのがよい。変更は宣言的に反映されるため、ステージングでの検証から本番へのプロモーションまで同じコードで管理できる。

AWS CLI：アーカイブとリプレイで不具合を再現

aws events create-archive \
  --archive-name app-archive \
  --event-source-arn arn:aws:events:ap-northeast-1:111111111111:event-bus/app-bus \
  --retention-days 7 \
  --description "short-term replay for debugging"

aws events start-replay \
  --replay-name rc-bugfix-replay \
  --description "replay last hour" \
  --event-start-time "$(date -u -d '-60 minutes' +%Y-%m-%dT%H:%M:%SZ)" \
  --event-end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
  --destination file://<(cat <<JSON
{"Arn": "arn:aws:events:ap-northeast-1:111111111111:event-bus/app-bus"}
JSON
)

アーカイブは短期保持でも価値が高い。発生条件の特定が難しい不具合に対し、当該時間帯のイベントをそのまま再現できるため、暫定対応から恒久対応への移行を加速する⁶。セキュリティ上はPIIの混入に注意し、必要に応じてトークナイズやマスキングを組み合わせる。

可観測性とパフォーマンス：SLO設計と実測で運用を固める

イベント駆動では「配信と処理の総遅延」をSLOの主指標に置くのが実践的だ。EventBridge単体は低レイテンシで動くが、ターゲット側のLambdaやStep Functionsの起動時間、外部I/Oの遅延が優位になる場面が多い。したがって、イベント投入から最終状態の到達までを一意に相関し、p50（中央値）とp99（99パーセンタイル）の分布を継続計測する⁴。一般的なワークロードでは、p50は数百ミリ秒、p99は数秒以内に収まることを目安にしつつ、ターゲットのコールドスタートや外部依存の劣化を見越してSLOを設計すると現実的だ。なお、EventBridgeのスループット期待値やスケーリングの観点は公式情報を併せて参照するとよい³。

CloudWatch Logs Insights：遅延の分布を継続監視

fields @timestamp, detail.occurredAt as occurredAt, @ingestionTime as ing
| filter detailType = 'OrderCreated'
| parse occurredAt "*" as occ
| stats 
  pct((ing - toMillis(occ)), 50) as p50_ms, 
  pct((ing - toMillis(occ)), 99) as p99_ms,
  count(*) as cnt
  by bin(1m)

イベントの発生時刻をdetailに含めていれば、取り込み時刻との差分で配信レイテンシを近似できる。正確なエンドツーエンドの遅延は、最後に状態が確定するポイントでメトリクスを発火し、イベントIDで相関させて求める。SLO違反が連続した場合は、ターゲットのスロットリングや外部依存の劣化を疑い、同時実行数の上限やリトライ設定を見直す⁴。

コストとROI：疎結合が生む独立デプロイの価値

EventBridgeの課金はイベントの発行やマッチに応じた従量であり、価格はリージョンや利用状況に依存する¹。同期APIのスケールアウトや待機時間のコストと比べると、事業スループットの改善に対して費用対効果を見込みやすい。費用の最適化は、不要な発行の抑制、detailのミニマム化、イベントの正規化で重複を避けることから始めるとよい。マイクロサービスが独立にデプロイ可能になることで、合意待ちや調整コストが減り、路線変更時の廃棄コストも小さくなる。これらは数値化しづらいが、四半期単位でのリードタイム中央値や変更失敗率を追うと、意思決定の材料になる。

スケールとガバナンス：マルチアカウントとセキュリティの要点

エンタープライズでは、組織単位のアカウント分離とネットワーク境界を保ちながら、イベントを横断的に流したい要求が強い。EventBridgeのイベントバスポリシーを用いれば、発行元のAWSアカウント、特定のロール、あるいはSaaSパートナーからのイベントのみ受け入れる制御が可能になる¹。IAMは最小権限を徹底し、プロデューサーにはevents:PutEvents、コンシューマーには必要なターゲットの実行権限のみを付与する。スキーマレジストリを必須化し、破壊的変更は新しいdetail-typeやバージョンを導入して段階的に移行する⁷。

フォールトトレランスでは、ルールごとのDLQと再試行ポリシーが中心になる。恒常的な失敗はDLQに集約され、オペレーションチームはDLQのデキューと是正を標準手順として運用できる。スパイク対策としては、ターゲット側のスロットリング上限と同時実行数を先に定め、EventBridgeの再試行を吸収層として用いると、スローダウンしつつも処理落ちを防げる²。可用性設計の観点では、クリティカルなドメインに対してはイベントの重要度に応じた冪等処理とオフセット管理を強化し、データ損失のリスクを下げる。

Step FunctionsやPipesとの使い分け

一連の業務を明示的にオーケストレーションしたい場合はStep Functionsをターゲットとして組み込み、ガードレールの下で分岐や待機を表現すると読みやすい。イベントの中継や変換が主眼ならEventBridge Pipesが適しており、SQSやKinesisなどのプル型ソースからターゲットへ、フィルタやEnrichmentを挟んで流せる¹。バスのルーティングは「何が起きたか」の共有と扇状の配信に向き、Pipesは「どこからどこへ」のストリーム処理に向く。両者を併用して、複雑さを適切な層に押し込めるのが設計の勘所だ。

例：Step Functionsをターゲットにして業務を起動

import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as events from 'aws-cdk-lib/aws-events';

const machine = new sfn.StateMachine(this, 'OrderFlow', {
  definitionBody: sfn.DefinitionBody.fromChainable(
    new sfn.Pass(this, 'Start')
  )
});

const bus = new events.EventBus(this, 'AppBus', { eventBusName: 'app-bus' });

new events.Rule(this, 'KickoffRule', {
  eventBus: bus,
  eventPattern: { source: ['com.example.orders'], detailType: ['OrderCreated'] }
}).addTarget(new targets.SfnStateMachine(machine, {
  input: events.RuleTargetInput.fromEventPath('$.detail')
}));

このように、イベントのdetailだけを入力に渡せば、ワークフローはイベントペイロードを前提として進行できる。状態機械側のタイムアウトやエラーハンドリングを整備すれば、複数の外部依存を伴う業務でも安定的に運用できる。

品質を保つ運用作法：スキーマ、バージョニング、リリース戦略

本番運用で最も効いてくるのは、スキーマとバージョニングの規律だ。互換性を破らない変更はフィールドの追加と既定値で吸収し、破壊的変更多は新しいdetail-typeまたはversionフィールドを導入して移行期間を設ける。プロデューサーは過去互換のイベントを二重発行し、コンシューマーは新旧を並行受信する、いわゆるストラングラー的な戦略が安全である⁷。これにより、各サービスが独立にデプロイでき、長期の調整やビッグバン移行を避けられる。

テスト戦略とステージングでの検証

契約テストの導入は、イベント駆動でも大きな効果を発揮する。スキーマレジストリに対してコンシューマーテストを先に走らせ、プロデューサーの変更が安全かを検証する⁷。ステージングではアーカイブとリプレイを活用して実データの匿名化サンプルで流量と遅延を測り、キャパシティ計画に反映する⁶。障害対応手順はDLQの監視とエラー種別ごとのオペレーションに落とし込み、再送の前提条件を明文化しておくと、オンコール時の判断が早くなる。

参考となるエンドツーエンドのイベント例

{
  "id": "a1b2c3d4-...",
  "source": "com.example.orders",
  "detail-type": "OrderCreated",
  "time": "2025-08-30T12:34:56Z",
  "region": "ap-northeast-1",
  "resources": [],
  "detail": {
    "orderId": "ORD-12345",
    "userId": "USR-999",
    "total": 12800,
    "occurredAt": "2025-08-30T12:34:55Z",
    "version": 1
  }
}

この形であれば、監査に必要なメタデータを保持しつつ、detailはビジネスドメインに閉じた表現を維持できる。イベントを将来にわたって資産として扱う姿勢が品質を支える。

デプロイパイプラインへの統合

CIではスキーマの互換性チェック、IaCの差分検証、SDKレベルのプロデューサーテストを必ず回す。CDでは環境ごとにイベントバス名を切り替え、影響範囲を限定した段階的リリースを実施する。監視アラートは配信失敗、DLQの増加、遅延のp99悪化を基準に設定しておくと、運用の見落としが減る。イベント数の急増は事業の成長と障害の兆候の両面がありうるため、ビジネスメトリクスと合わせて解釈するのが現実的だ。

まとめ：小さく始めて、確実に価値へつなげる

EventBridgeを据えたイベント駆動アーキテクチャは、同期連携の脆さを克服し、チームの自律性を高める現実解だ。まずは単一ドメインのイベントから始め、バス、ルール、ターゲット、DLQ、そしてアーカイブといった最小要素で運用を回し、レイテンシの分布と失敗率を継続的に観測する。スキーマとバージョンの規律を守り、プロデューサーとコンシューマーを独立に進化させれば、全体を止めずに価値を積み上げられる。いま直面する同期APIのつらさが一つでもあるなら、最も痛みの強い箇所を一件だけイベント化し、計測とリプレイ可能な運用を試してみてほしい。小さな成功が、アーキテクチャ全体の転換点になる。

参考文献

  1. Amazon EventBridge FAQs. https://aws.amazon.com/eventbridge/faqs/
  2. Amazon EventBridge delivery behavior (at-least-once, retries, DLQ). https://docs.aws.amazon.com/eventbridge/latest/ref/event-delivery-level.html
  3. Amazon EventBridge FAQs – Throughput expectations. https://aws.amazon.com/eventbridge/faqs/#:~:text=What%20throughput%20can%20I%20expect,from%20EventBridge
  4. Monitoring best practices for event delivery with Amazon EventBridge. https://aws.amazon.com/blogs/compute/monitoring-best-practices-for-event-delivery-with-amazon-eventbridge/
  5. Decision guide: When to use Amazon SNS, SQS, or EventBridge. https://docs.aws.amazon.com/decision-guides/latest/sns-or-sqs-or-eventbridge/sns-or-sqs-or-eventbridge.html
  6. Archiving and replaying events with Amazon EventBridge. https://aws.amazon.com/blogs/compute/archiving-and-replaying-events-with-amazon-eventbridge/
  7. Introducing Amazon EventBridge schema registry and discovery. https://aws.amazon.com/blogs/compute/introducing-amazon-eventbridge-schema-registry-and-discovery-in-preview/
  8. Amazon EventBridge API Reference – PutEvents. https://docs.aws.amazon.com/eventbridge/latest/APIReference/API_PutEvents.html
  9. Decision guide (message filtering and pattern-based routing). https://docs.aws.amazon.com/decision-guides/latest/sns-or-sqs-or-eventbridge/sns-or-sqs-or-eventbridge.html#:~:text=Message%20filtering%20%20,based%20filtering