diff --git a/eventbridge-firehose-opensearch-cdk/README.md b/eventbridge-firehose-opensearch-cdk/README.md new file mode 100644 index 000000000..6de2b6216 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/README.md @@ -0,0 +1,222 @@ +# Amazon EventBridge to Amazon OpenSearch via Amazon Data Firehose + +This pattern gives you **full-text search over every event on an EventBridge bus**, usually within about 60 seconds of emission. A catch-all rule captures all events on a custom bus and streams them to an OpenSearch domain through Amazon Data Firehose, where OpenSearch Dashboards makes them searchable, filterable, and chartable. + +CloudWatch metrics tell you *how many* events flowed. This tells you *what was in them*. + +![Architecture](architecture.png) + +``` +Any event producer + │ + ▼ +EventBridge custom bus + │ Rule: matches ALL events + ▼ +Amazon Data Firehose ──────► Transform Lambda (flattens the envelope) + │ 60s / 1MB buffer + ├──────► OpenSearch domain index: events-YYYY-MM-DD + └──────► S3 bucket all documents + delivery failures +``` + +Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/ + +Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example. + +## Why this pattern + +In an event-driven architecture you eventually need answers to questions metrics cannot give you: + +- What events flowed through the bus in the last five minutes? +- Which producer emitted the most events today? +- Show me every event related to claim `CLM-12345`. +- Is event volume anomalous compared to yesterday? + +Because the rule is a catch-all, you get this for every event on the bus without touching any producer. + +## How it works + +1. A producer calls `PutEvents` on the `event-monitor-bus` custom bus. +2. A catch-all rule matches the event. The pattern is `{"source": [{"prefix": ""}]}` — every EventBridge event carries a `source`, so an empty prefix matches all of them. An entirely empty pattern is rejected by EventBridge. +3. The rule's IAM role calls `firehose:PutRecord` on the delivery stream. +4. Firehose buffers for **60 seconds or 1 MB**, whichever comes first. This is the minimum Firehose allows and it defines the latency of the pattern. +5. A Lambda transform flattens each EventBridge envelope (see below). +6. Firehose signs its request with SigV4 using its delivery role and indexes each document into `events-YYYY-MM-DD`. +7. Every document is also written to the S3 bucket under `events/`. Records that fail to transform or deliver land under `errors/`. + +### What the transform does + +EventBridge delivers a nested envelope. Two things make that awkward to query: + +- `detail-type` contains a hyphen, so it needs escaping in DQL and Lucene queries. +- Business fields sit one level down under `detail`, so every filter reads `detail.claimId`. + +[`src/transform/handler.py`](src/transform/handler.py) renames `detail-type` to `detail_type` and promotes the `detail` keys to the top level: + +```jsonc +// in +{ "source": "agent.claims", "detail-type": "ClaimApproved", "time": "2026-08-17T10:00:00Z", + "detail": { "claimId": "CLM-001", "decision": "approved" } } + +// out +{ "source": "agent.claims", "detail_type": "ClaimApproved", "time": "2026-08-17T10:00:00Z", + "claimId": "CLM-001", "decision": "approved" } +``` + +Envelope fields win on collision: a payload carrying its own `source` key is indexed as `detail_source` rather than masking the real event source. Records that cannot be parsed are returned as `ProcessingFailed`, which routes that one record to the S3 error prefix and lets the rest of the batch through. + +Disable the transform with `-c enableTransform=false` to index the raw envelope instead. + +### Authentication between components + +| Hop | Mechanism | +|---|---| +| Rule → Firehose | Rule target IAM role with `firehose:PutRecord`, `firehose:PutRecordBatch`, scoped to the stream | +| Firehose → OpenSearch | SigV4 with the delivery role, granted on **both** sides: an identity policy on the role and a domain access policy naming it as principal | +| Firehose → Lambda / S3 / Logs | Same delivery role | +| Operator → Dashboards | Anonymous, restricted to the CIDR passed as `dashboardAccessIp` | + +Two details are easy to get wrong here. A managed domain authorizes every request against its own access policy, so an identity policy alone is not enough. And Firehose needs `es:DescribeDomain`, `es:DescribeDomainConfig`, and `es:DescribeDomains` to resolve the domain endpoint before it can deliver anything — `grantIndexWrite()` does not include those. + +## Prerequisites + +- [AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) with sufficient permissions +- [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cli.html) installed and configured +- [Node.js 20+](https://nodejs.org/en/download/) and npm +- [AWS CDK CLI](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) (`npm i -g aws-cdk`), bootstrapped in the target account and Region +- Your public IP, to reach OpenSearch Dashboards: `curl -s https://checkip.amazonaws.com` + +## Deployment + +```bash +git clone https://github.com/aws-samples/serverless-patterns +cd serverless-patterns/eventbridge-firehose-opensearch-cdk/cdk +npm install +cdk deploy -c dashboardAccessIp=$(curl -s https://checkip.amazonaws.com)/32 +``` + +Creating the OpenSearch domain takes 10 to 20 minutes; the rest of the stack is quick. + +Omit `dashboardAccessIp` and the domain stays closed to everything except Firehose. Delivery still works, but you will not be able to open Dashboards. + +Context values: + +| Value | Default | Effect | +|---|---|---| +| `dashboardAccessIp` | none | CIDR(s) granted Dashboards access, e.g. `1.2.3.4/32`. Comma-separate for several: `1.2.3.4/32,5.6.7.8/32` | +| `enableTransform` | `true` | Set `false` to index the raw EventBridge envelope | + +Changing the allowed CIDRs later is cheap. It updates only the domain access policy, so the redeploy takes seconds rather than rebuilding the domain. + +## Testing + +Emit a test event onto the bus: + +```bash +aws events put-events --entries '[{ + "Source": "demo.test", + "DetailType": "TestEvent", + "Detail": "{\"message\": \"Hello OpenSearch\", \"claimId\": \"CLM-001\"}", + "EventBusName": "event-monitor-bus" +}]' +``` + +A successful call returns `"FailedEntryCount": 0`. Wait 60 to 90 seconds for the Firehose buffer to flush. + +Then query the domain directly from the allowlisted IP, using `DomainEndpoint` from the stack outputs: + +```bash +ENDPOINT="" +curl -s "https://$ENDPOINT/events-*/_search?pretty" -H 'Content-Type: application/json' \ + -d '{"query": {"match": {"claimId": "CLM-001"}}}' +``` + +You should see the flattened document, with `detail_type` set to `TestEvent` and `claimId` at the top level. + +Confirm the backup copy reached S3: + +```bash +aws s3 ls "s3:///events/" --recursive +``` + +### If nothing arrives + +Delivery failures are invisible unless you look for them, which is why the stack creates a log group for them. Check `FirehoseLogGroup` from the stack outputs: + +```bash +aws logs tail "" --since 15m +``` + +| Symptom | Likely cause | +|---|---| +| `403` `User: anonymous is not authorized` on your own queries | The address you are calling from is not covered by `dashboardAccessIp` | +| `AccessDeniedException` on the domain | Domain access policy missing the Firehose role, or the role lacks the `es:Describe*` actions | +| Records in S3 under `errors/` but nothing in OpenSearch | Mapping conflict, usually a field indexed as two different types across events | +| Nothing anywhere, `FailedEntryCount: 0` on `put-events` | Rule not matching, or the rule's target role lacks `firehose:PutRecord` | +| Transform errors | Check the `event-monitor-transform` Lambda log group | + +## Setting up OpenSearch Dashboards + +1. Open `DashboardsUrl` from the stack outputs. +2. **Dashboards Management → Index patterns → Create index pattern**. +3. Index pattern name: `events-*`. Time field: `time`. +4. Go to **Discover** to browse events, or **Visualize** to chart them. Useful starting points: event count by `source` as a pie chart, events over time as a line chart, `detail_type` breakdown as a bar chart. + +### Optional: apply an index template + +Without a template, OpenSearch infers mappings dynamically. That works, with two rough edges: string fields become `text` with a `.keyword` subfield, so aggregations need `source.keyword` rather than `source`; and new indices default to one replica, which leaves a single-node cluster permanently yellow because the replica shard can never be assigned. + +Applying a template fixes both. Paste this into **Dev Tools** in Dashboards before sending events: + +```json +PUT _index_template/events +{ + "index_patterns": ["events-*"], + "template": { + "settings": { + "number_of_shards": 1, + "number_of_replicas": 0 + }, + "mappings": { + "properties": { + "id": { "type": "keyword" }, + "source": { "type": "keyword" }, + "detail_type": { "type": "keyword" }, + "time": { "type": "date" }, + "account": { "type": "keyword" }, + "region": { "type": "keyword" }, + "claimId": { "type": "keyword" }, + "status": { "type": "keyword" }, + "agentId": { "type": "keyword" } + } + } + } +} +``` + +The field names here match the flattened output of the transform. If you deploy with `-c enableTransform=false`, map `detail-type` and `detail.*` instead. + +## Cleanup + +```bash +cd cdk +cdk destroy +``` + +The domain, the S3 bucket and its contents, and all log groups are removed. Deleting the domain takes several minutes. + +## Cost considerations + +For a low-volume demo the domain dominates the bill: + +- OpenSearch `t3.small.search`, 1 node: ~$26/month, plus ~$1.60/month for 20 GB gp3 +- Data Firehose: $0.029 per GB ingested +- Lambda transform, EventBridge, and S3: negligible at demo volume + +Destroy the stack when you are done. For production, size the domain to your retention and query load, move to Multi-AZ with a dedicated master, and consider UltraWarm for older indices. + +--- + +Copyright 2026 Amazon.com, Inc. or its affiliates. All Rights Reserved. + +SPDX-License-Identifier: MIT-0 diff --git a/eventbridge-firehose-opensearch-cdk/architecture.png b/eventbridge-firehose-opensearch-cdk/architecture.png new file mode 100644 index 000000000..8d5aa0f16 Binary files /dev/null and b/eventbridge-firehose-opensearch-cdk/architecture.png differ diff --git a/eventbridge-firehose-opensearch-cdk/cdk/.gitignore b/eventbridge-firehose-opensearch-cdk/cdk/.gitignore new file mode 100644 index 000000000..459d58545 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/.gitignore @@ -0,0 +1,7 @@ +node_modules +cdk.out +*.js +!jest.config.js +*.d.ts +.cdk.staging +*.tsbuildinfo diff --git a/eventbridge-firehose-opensearch-cdk/cdk/bin/app.ts b/eventbridge-firehose-opensearch-cdk/cdk/bin/app.ts new file mode 100644 index 000000000..053dc7431 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/bin/app.ts @@ -0,0 +1,14 @@ +#!/usr/bin/env node +import * as cdk from 'aws-cdk-lib'; +import { EventBridgeOpenSearchStack } from '../lib/eventbridge-opensearch-stack'; + +const app = new cdk.App(); + +new EventBridgeOpenSearchStack(app, 'EventMonitorStack', { + description: + 'ServerlessLand pattern: stream all EventBridge events to OpenSearch via Amazon Data Firehose for near real-time monitoring', + env: { + account: process.env.CDK_DEFAULT_ACCOUNT, + region: process.env.CDK_DEFAULT_REGION, + }, +}); diff --git a/eventbridge-firehose-opensearch-cdk/cdk/cdk.json b/eventbridge-firehose-opensearch-cdk/cdk/cdk.json new file mode 100644 index 000000000..020a631ce --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/cdk.json @@ -0,0 +1,21 @@ +{ + "app": "npx ts-node --prefer-ts-exts bin/app.ts", + "watch": { + "include": ["**"], + "exclude": [ + "README.md", + "cdk*.json", + "**/*.d.ts", + "**/*.js", + "tsconfig.json", + "package*.json", + "node_modules", + "cdk.out" + ] + }, + "context": { + "@aws-cdk/aws-iam:minimizePolicies": true, + "@aws-cdk/core:checkSecretUsage": true, + "@aws-cdk/aws-iam:standardizedServicePrincipals": true + } +} diff --git a/eventbridge-firehose-opensearch-cdk/cdk/lib/eventbridge-opensearch-stack.ts b/eventbridge-firehose-opensearch-cdk/cdk/lib/eventbridge-opensearch-stack.ts new file mode 100644 index 000000000..6dac56b38 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/lib/eventbridge-opensearch-stack.ts @@ -0,0 +1,427 @@ +import * as path from 'path'; +import * as cdk from 'aws-cdk-lib'; +import * as ec2 from 'aws-cdk-lib/aws-ec2'; +import * as events from 'aws-cdk-lib/aws-events'; +import * as targets from 'aws-cdk-lib/aws-events-targets'; +import * as iam from 'aws-cdk-lib/aws-iam'; +import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose'; +import * as lambda from 'aws-cdk-lib/aws-lambda'; +import * as logs from 'aws-cdk-lib/aws-logs'; +import * as opensearch from 'aws-cdk-lib/aws-opensearchservice'; +import * as s3 from 'aws-cdk-lib/aws-s3'; +import { Construct } from 'constructs'; + +/** + * EventBridge -> Amazon Data Firehose -> OpenSearch (Event Monitor pattern) + * + * A catch-all EventBridge rule captures every event on a custom bus and + * streams it to an OpenSearch domain through Amazon Data Firehose, giving + * full-text search over event payloads within about 60 seconds. + * + * Flow: + * Any producer -> EventBridge custom bus + * -> Rule (matches ALL events) + * -> Firehose delivery stream + * -> transform Lambda (flattens the EventBridge envelope) + * -> OpenSearch domain, daily-rotated index + * -> S3 bucket (backup + transform/delivery failures) + * + * Auth model: + * - Rule -> Firehose: the rule target's IAM role holds firehose:PutRecord + * and firehose:PutRecordBatch, scoped to this delivery stream. + * - Firehose -> OpenSearch: SigV4 with the Firehose delivery role. Access + * is granted on both sides -- an identity policy on the role AND a + * domain access policy naming the role as principal. A managed domain + * authorizes every request against its access policy, so the identity + * policy alone is not enough. + * - Firehose -> Lambda / S3 / CloudWatch Logs: same delivery role. + * - Operator -> Dashboards: browsers cannot sign requests with SigV4, so + * Dashboards access is granted to an optional IP CIDR instead. Omit the + * `dashboardAccessIp` context value and the domain stays closed to + * everything except Firehose. + * + * Context values: + * -c dashboardAccessIp=1.2.3.4/32 grant Dashboards access to a CIDR + * -c enableTransform=false index the raw EventBridge envelope + */ +export class EventBridgeOpenSearchStack extends cdk.Stack { + constructor(scope: Construct, id: string, props?: cdk.StackProps) { + super(scope, id, props); + + // Accepts one CIDR or a comma-separated list, so several operator + // addresses can be allowed without widening the range. + const dashboardAccessIpRaw = this.node.tryGetContext('dashboardAccessIp') as string | undefined; + const dashboardAccessIps = dashboardAccessIpRaw + ? dashboardAccessIpRaw + .split(',') + .map((cidr) => cidr.trim()) + .filter((cidr) => cidr.length > 0) + : []; + + // Transform is on by default; it is what makes the data pleasant to query. + const enableTransform = this.node.tryGetContext('enableTransform') !== 'false'; + + const domainName = 'event-monitor'; + const indexName = 'events'; + + // The domain ARN is built by hand rather than read off the Domain + // construct. The domain's own access policy has to name the Firehose + // role, and the Firehose role's policy has to name the domain -- going + // through the construct in both directions would be a circular + // reference. The name is fixed, so the ARN is knowable up front. + const domainArn = this.formatArn({ + service: 'es', + resource: 'domain', + resourceName: domainName, + }); + + // ------------------------------------------------------------------- + // 1. EventBridge custom bus + // ------------------------------------------------------------------- + const eventBus = new events.EventBus(this, 'EventMonitorBus', { + eventBusName: 'event-monitor-bus', + }); + + // ------------------------------------------------------------------- + // 2. S3 backup bucket + // + // Firehose requires an S3 configuration on the OpenSearch destination + // even when backing up only failures, so this bucket is not optional. + // ------------------------------------------------------------------- + const backupBucket = new s3.Bucket(this, 'BackupBucket', { + encryption: s3.BucketEncryption.S3_MANAGED, + blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL, + enforceSSL: true, + lifecycleRules: [ + { + id: 'archive-then-expire', + transitions: [ + { + storageClass: s3.StorageClass.GLACIER, + transitionAfter: cdk.Duration.days(30), + }, + ], + expiration: cdk.Duration.days(365), + }, + ], + // Demo pattern: leave nothing behind on `cdk destroy`. + removalPolicy: cdk.RemovalPolicy.DESTROY, + autoDeleteObjects: true, + }); + + // ------------------------------------------------------------------- + // 3. Firehose delivery role + // + // Created before the domain so the domain access policy can name it. + // Its permissions are attached further down, once the resources it + // needs to reach actually exist. + // ------------------------------------------------------------------- + const firehoseRole = new iam.Role(this, 'FirehoseDeliveryRole', { + assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'), + description: 'Lets Firehose deliver EventBridge events to OpenSearch and back them up to S3', + }); + + // ------------------------------------------------------------------- + // 4. OpenSearch domain + // ------------------------------------------------------------------- + const domainAccessPolicies = [ + // Firehose delivery role: write documents and read index metadata. + new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + principals: [new iam.ArnPrincipal(firehoseRole.roleArn)], + actions: ['es:ESHttpPost', 'es:ESHttpPut', 'es:ESHttpGet', 'es:ESHttpHead'], + resources: [domainArn, `${domainArn}/*`], + }), + ]; + + if (dashboardAccessIps.length > 0) { + // Dashboards runs in a browser, which cannot SigV4-sign requests. + // Anonymous access narrowed to known CIDRs is the standard way in when + // fine-grained access control is off. + domainAccessPolicies.push( + new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + principals: [new iam.AnyPrincipal()], + actions: ['es:ESHttp*'], + resources: [domainArn, `${domainArn}/*`], + conditions: { + IpAddress: { 'aws:SourceIp': dashboardAccessIps }, + }, + }), + ); + } + + const domain = new opensearch.Domain(this, 'EventMonitorDomain', { + domainName, + version: opensearch.EngineVersion.OPENSEARCH_2_19, + capacity: { + dataNodes: 1, + dataNodeInstanceType: 't3.small.search', + // t3 instance types cannot run Multi-AZ with standby, and CDK + // turns it on by default. Leaving this unset fails synthesis. + multiAzWithStandbyEnabled: false, + }, + ebs: { + volumeSize: 20, + volumeType: ec2.EbsDeviceVolumeType.GP3, + }, + // Single node, so no zone awareness and no dedicated master. + zoneAwareness: { enabled: false }, + enforceHttps: true, + nodeToNodeEncryption: true, + encryptionAtRest: { enabled: true }, + accessPolicies: domainAccessPolicies, + logging: { + appLogEnabled: true, + appLogGroup: new logs.LogGroup(this, 'DomainAppLogs', { + retention: logs.RetentionDays.ONE_WEEK, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }), + }, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }); + + // ------------------------------------------------------------------- + // 5. Transform Lambda (flattens the EventBridge envelope) + // ------------------------------------------------------------------- + let transformFn: lambda.Function | undefined; + + if (enableTransform) { + transformFn = new lambda.Function(this, 'TransformFunction', { + functionName: 'event-monitor-transform', + runtime: lambda.Runtime.PYTHON_3_12, + handler: 'handler.handler', + code: lambda.Code.fromAsset(path.join(__dirname, '..', '..', 'src', 'transform')), + // Firehose gives a transform 60s before it counts as a failure. + timeout: cdk.Duration.seconds(60), + memorySize: 256, + description: 'Flattens the EventBridge envelope before documents are indexed in OpenSearch', + logGroup: new logs.LogGroup(this, 'TransformLogs', { + retention: logs.RetentionDays.ONE_WEEK, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }), + }); + } + + // ------------------------------------------------------------------- + // 6. Firehose error logging + // + // Delivery failures surface nowhere else -- without this, a rejected + // document is invisible. + // ------------------------------------------------------------------- + const firehoseLogGroup = new logs.LogGroup(this, 'FirehoseLogs', { + retention: logs.RetentionDays.ONE_WEEK, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }); + + const openSearchLogStream = new logs.LogStream(this, 'OpenSearchDeliveryLogStream', { + logGroup: firehoseLogGroup, + logStreamName: 'OpenSearchDelivery', + removalPolicy: cdk.RemovalPolicy.DESTROY, + }); + + const backupLogStream = new logs.LogStream(this, 'BackupDeliveryLogStream', { + logGroup: firehoseLogGroup, + logStreamName: 'BackupDelivery', + removalPolicy: cdk.RemovalPolicy.DESTROY, + }); + + // ------------------------------------------------------------------- + // 7. Firehose delivery role permissions + // ------------------------------------------------------------------- + // es:ESHttp* writes the documents. The Describe* actions are separate + // and easy to miss: Firehose calls them to resolve the domain endpoint + // before it can deliver anything, and `grantIndexWrite` does not + // include them. + firehoseRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'OpenSearchDescribe', + actions: ['es:DescribeDomain', 'es:DescribeDomainConfig', 'es:DescribeDomains'], + resources: [domainArn], + }), + ); + + firehoseRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'OpenSearchWrite', + actions: ['es:ESHttpPost', 'es:ESHttpPut', 'es:ESHttpGet'], + resources: [domainArn, `${domainArn}/*`], + }), + ); + + backupBucket.grantWrite(firehoseRole); + firehoseRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'S3BackupRead', + actions: ['s3:GetBucketLocation', 's3:ListBucket', 's3:ListBucketMultipartUploads'], + resources: [backupBucket.bucketArn], + }), + ); + + firehoseRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'FirehoseErrorLogging', + actions: ['logs:PutLogEvents'], + resources: [firehoseLogGroup.logGroupArn, `${firehoseLogGroup.logGroupArn}:*`], + }), + ); + + if (transformFn) { + firehoseRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'InvokeTransform', + actions: ['lambda:InvokeFunction', 'lambda:GetFunctionConfiguration'], + resources: [transformFn.functionArn, `${transformFn.functionArn}:*`], + }), + ); + } + + // ------------------------------------------------------------------- + // 8. Firehose delivery stream + // + // L1 by design: the L2 DeliveryStream only accepts destinations that + // implement IDestination, and the only one shipped is S3. OpenSearch + // has to be configured through CfnDeliveryStream. + // ------------------------------------------------------------------- + const deliveryStreamName = 'event-monitor-stream'; + + const cfnStream = new firehose.CfnDeliveryStream(this, 'EventDeliveryStream', { + deliveryStreamName, + deliveryStreamType: 'DirectPut', + amazonopensearchserviceDestinationConfiguration: { + // Combined with OneDay rotation this yields events-YYYY-MM-DD. + indexName, + indexRotationPeriod: 'OneDay', + // domainArn and clusterEndpoint are mutually exclusive. + domainArn: domain.domainArn, + roleArn: firehoseRole.roleArn, + // 60s / 1MB is the floor Firehose allows, and it sets the + // end-to-end latency of this pattern. + bufferingHints: { + intervalInSeconds: 60, + sizeInMBs: 1, + }, + retryOptions: { + durationInSeconds: 300, + }, + // AllDocuments keeps a durable copy of everything indexed, which + // makes the S3 bucket an audit trail rather than just a dead + // letter destination. + s3BackupMode: 'AllDocuments', + s3Configuration: { + bucketArn: backupBucket.bucketArn, + roleArn: firehoseRole.roleArn, + prefix: 'events/', + errorOutputPrefix: 'errors/', + bufferingHints: { + intervalInSeconds: 300, + sizeInMBs: 5, + }, + compressionFormat: 'GZIP', + cloudWatchLoggingOptions: { + enabled: true, + logGroupName: firehoseLogGroup.logGroupName, + logStreamName: backupLogStream.logStreamName, + }, + }, + cloudWatchLoggingOptions: { + enabled: true, + logGroupName: firehoseLogGroup.logGroupName, + logStreamName: openSearchLogStream.logStreamName, + }, + processingConfiguration: transformFn + ? { + enabled: true, + processors: [ + { + type: 'Lambda', + parameters: [ + { parameterName: 'LambdaArn', parameterValue: transformFn.functionArn }, + { parameterName: 'RoleArn', parameterValue: firehoseRole.roleArn }, + // Lambda processor buffer must stay within 0.2-3 MB. + { parameterName: 'BufferSizeInMBs', parameterValue: '1' }, + { parameterName: 'BufferIntervalInSeconds', parameterValue: '60' }, + { parameterName: 'NumberOfRetries', parameterValue: '3' }, + ], + }, + ], + } + : undefined, + }, + }); + + // The stream is only usable once the role's policies are attached and + // the domain access policy has been applied. Neither shows up as a + // CloudFormation reference, so the ordering has to be explicit. + cfnStream.node.addDependency(firehoseRole); + cfnStream.node.addDependency(domain); + + // ------------------------------------------------------------------- + // 9. Catch-all EventBridge rule + // ------------------------------------------------------------------- + // Wrapping the L1 stream as an L2 lets the event target construct + // build the rule's IAM role for us. + const deliveryStream = firehose.DeliveryStream.fromDeliveryStreamArn( + this, + 'ImportedDeliveryStream', + cfnStream.attrArn, + ); + + const rule = new events.Rule(this, 'CatchAllRule', { + eventBus, + ruleName: 'event-monitor-catch-all', + description: 'Captures every event on the bus and streams it to OpenSearch via Data Firehose', + // Every event carries a source, so an empty prefix matches all of + // them. An empty event pattern is rejected by EventBridge. + eventPattern: { + source: events.Match.prefix(''), + }, + }); + + rule.addTarget(new targets.FirehoseDeliveryStream(deliveryStream)); + + // Imported constructs carry no dependency edge of their own. + rule.node.addDependency(cfnStream); + + // ------------------------------------------------------------------- + // Outputs + // ------------------------------------------------------------------- + new cdk.CfnOutput(this, 'DashboardsUrl', { + value: `https://${domain.domainEndpoint}/_dashboards/`, + description: + dashboardAccessIps.length > 0 + ? `OpenSearch Dashboards URL (reachable from: ${dashboardAccessIps.join(', ')})` + : 'OpenSearch Dashboards URL (no public access granted -- redeploy with -c dashboardAccessIp=YOUR_IP/32)', + }); + + new cdk.CfnOutput(this, 'DomainEndpoint', { + value: domain.domainEndpoint, + description: 'OpenSearch domain endpoint', + }); + + new cdk.CfnOutput(this, 'EventBusName', { + value: eventBus.eventBusName, + description: 'EventBridge custom bus being monitored', + }); + + new cdk.CfnOutput(this, 'DeliveryStreamName', { + value: deliveryStreamName, + description: 'Firehose delivery stream carrying events to OpenSearch', + }); + + new cdk.CfnOutput(this, 'BackupBucketName', { + value: backupBucket.bucketName, + description: 'S3 bucket holding the event backup and any delivery failures', + }); + + new cdk.CfnOutput(this, 'FirehoseLogGroup', { + value: firehoseLogGroup.logGroupName, + description: 'CloudWatch log group for Firehose delivery errors', + }); + + new cdk.CfnOutput(this, 'IndexPattern', { + value: `${indexName}-*`, + description: 'Index pattern to create in OpenSearch Dashboards (time field: time)', + }); + } +} diff --git a/eventbridge-firehose-opensearch-cdk/cdk/package.json b/eventbridge-firehose-opensearch-cdk/cdk/package.json new file mode 100644 index 000000000..cf7492011 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/package.json @@ -0,0 +1,25 @@ +{ + "name": "eventbridge-firehose-opensearch-cdk", + "version": "1.0.0", + "description": "Stream all EventBridge events to OpenSearch via Amazon Data Firehose for near real-time monitoring", + "bin": { + "app": "bin/app.js" + }, + "scripts": { + "build": "tsc", + "watch": "tsc -w", + "cdk": "cdk", + "deploy": "cdk deploy", + "destroy": "cdk destroy" + }, + "devDependencies": { + "@types/node": "20.14.9", + "aws-cdk": "2.1136.0", + "ts-node": "10.9.2", + "typescript": "5.5.3" + }, + "dependencies": { + "aws-cdk-lib": "2.264.0", + "constructs": "10.8.1" + } +} diff --git a/eventbridge-firehose-opensearch-cdk/cdk/tsconfig.json b/eventbridge-firehose-opensearch-cdk/cdk/tsconfig.json new file mode 100644 index 000000000..b1eaa510e --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/cdk/tsconfig.json @@ -0,0 +1,24 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "lib": ["es2022"], + "declaration": true, + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "noImplicitThis": true, + "alwaysStrict": true, + "noUnusedLocals": false, + "noUnusedParameters": false, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": false, + "inlineSourceMap": true, + "inlineSources": true, + "experimentalDecorators": true, + "strictPropertyInitialization": false, + "typeRoots": ["./node_modules/@types"] + }, + "exclude": ["node_modules", "cdk.out"] +} diff --git a/eventbridge-firehose-opensearch-cdk/eventbridge-firehose-opensearch-cdk.json b/eventbridge-firehose-opensearch-cdk/eventbridge-firehose-opensearch-cdk.json new file mode 100644 index 000000000..f9979627a --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/eventbridge-firehose-opensearch-cdk.json @@ -0,0 +1,128 @@ +{ + "title": "Amazon EventBridge to Amazon OpenSearch via Amazon Data Firehose", + "description": "Monitor every event flowing through an EventBridge bus in near real time by streaming it to OpenSearch through Amazon Data Firehose, with daily index rotation, payload flattening, and an S3 backup of all documents.", + "language": "TypeScript", + "level": "200", + "framework": "CDK", + "patternArch": { + "icon1": { + "x": 10, + "y": 50, + "service": "eventbridge", + "label": "Amazon EventBridge" + }, + "icon2": { + "x": 40, + "y": 50, + "service": "kinesis-firehose", + "label": "Amazon Data Firehose" + }, + "icon3": { + "x": 40, + "y": 15, + "service": "lambda", + "label": "Transform Lambda" + }, + "icon4": { + "x": 75, + "y": 50, + "service": "opensearch", + "label": "Amazon OpenSearch" + }, + "icon5": { + "x": 75, + "y": 15, + "service": "s3", + "label": "Amazon S3" + }, + "line1": { + "from": "icon1", + "to": "icon2", + "label": "All events" + }, + "line2": { + "from": "icon2", + "to": "icon3", + "label": "Flatten" + }, + "line3": { + "from": "icon2", + "to": "icon4", + "label": "Index" + }, + "line4": { + "from": "icon2", + "to": "icon5", + "label": "Backup" + } + }, + "introBox": { + "headline": "How it works", + "text": [ + "This pattern gives you full-text search over every event on an Amazon EventBridge bus, usually within about 60 seconds of the event being emitted. CloudWatch metrics tell you how many events flowed; this tells you what was in them.", + "A catch-all EventBridge rule matches every event on a custom bus and sends it to an Amazon Data Firehose delivery stream. The rule pattern matches on a source prefix of an empty string, because every EventBridge event carries a source and an entirely empty pattern is rejected.", + "Firehose buffers for 60 seconds or 1 MB, whichever comes first. That buffer is the floor Firehose allows and it sets the end-to-end latency of the pattern.", + "Before indexing, a Lambda transform flattens the EventBridge envelope: detail-type is renamed to detail_type so it needs no escaping in queries, and the fields inside detail are promoted to the top level so a dashboard can filter on claimId rather than detail.claimId. Envelope fields win on collision, so a business payload carrying its own source key is indexed as detail_source instead of masking the real event source.", + "Documents land in a daily-rotated index, events-YYYY-MM-DD, and every document is also written to an S3 bucket, which makes that bucket an audit trail rather than only a dead letter destination.", + "Firehose authenticates to OpenSearch with SigV4 using its delivery role. A managed domain authorizes each request against its own access policy, so the role is granted access on both sides: an identity policy on the role and a domain access policy naming that role as principal." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/eventbridge-firehose-opensearch-cdk", + "templateURL": "serverless-patterns/eventbridge-firehose-opensearch-cdk", + "projectFolder": "eventbridge-firehose-opensearch-cdk", + "templateFile": "cdk/lib/eventbridge-opensearch-stack.ts" + } + }, + "resources": { + "bullets": [ + { + "text": "Loading streaming data into Amazon OpenSearch Service with Amazon Data Firehose", + "link": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/integrations-fh.html" + }, + { + "text": "Amazon Data Firehose data transformation with AWS Lambda", + "link": "https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html" + }, + { + "text": "Amazon EventBridge event patterns", + "link": "https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html" + }, + { + "text": "Identity and Access Management in Amazon OpenSearch Service", + "link": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html" + }, + { + "text": "Audit AWS service events with Amazon EventBridge and Amazon Data Firehose", + "link": "https://aws.amazon.com/blogs/big-data/audit-aws-service-events-with-amazon-eventbridge-and-amazon-kinesis-data-firehose/" + } + ] + }, + "deploy": { + "text": [ + "cd cdk", + "npm install", + "cdk deploy -c dashboardAccessIp=YOUR_IP/32" + ] + }, + "testing": { + "headline": "Testing", + "text": [ + "See the GitHub repo README.md for detailed testing instructions.", + "Emit a test event onto the bus: aws events put-events --entries '[{\"Source\":\"demo.test\",\"DetailType\":\"TestEvent\",\"Detail\":\"{\\\"message\\\":\\\"Hello OpenSearch\\\",\\\"claimId\\\":\\\"CLM-001\\\"}\",\"EventBusName\":\"event-monitor-bus\"}]'", + "Wait 60 to 90 seconds for the Firehose buffer to flush, then open the DashboardsUrl from the stack outputs and create an index pattern of events-* with time as the time field." + ] + }, + "cleanup": { + "headline": "Cleanup", + "text": ["cd cdk", "cdk destroy"] + }, + "authors": [ + { + "name": "Antoine Boucherie", + "bio": "Principal Solutions Architect, AWS Global Financial Services", + "linkedin": "antoineboucherie" + } + ] +} diff --git a/eventbridge-firehose-opensearch-cdk/src/transform/handler.py b/eventbridge-firehose-opensearch-cdk/src/transform/handler.py new file mode 100644 index 000000000..d112260e4 --- /dev/null +++ b/eventbridge-firehose-opensearch-cdk/src/transform/handler.py @@ -0,0 +1,119 @@ +""" +Amazon Data Firehose transform: flatten the EventBridge envelope. + +EventBridge delivers events to Firehose in their raw envelope form: + + { + "version": "0", + "id": "e7c9...", + "detail-type": "ClaimApproved", + "source": "agent.claims-processor", + "account": "111122223333", + "time": "2026-08-17T10:00:00Z", + "region": "us-east-1", + "resources": [], + "detail": { "claimId": "CLM-001", "decision": "approved" } + } + +Two things make that awkward to query in OpenSearch: + +1. ``detail-type`` contains a hyphen, so it needs escaping in DQL/Lucene + queries and cannot be referenced directly in some aggregations. +2. Business fields are nested one level down under ``detail``, so every + dashboard filter has to be written as ``detail.claimId`` instead of + ``claimId``. + +This transform renames ``detail-type`` to ``detail_type`` and promotes the +``detail`` keys to the top level, so a search for ``claimId: "CLM-001"`` +works directly. + +Envelope fields win on collision: if a payload contains its own ``source`` +key it is indexed as ``detail_source`` rather than overwriting the +EventBridge envelope value. Without this guard a business payload could +silently mask the real event source. +""" + +import base64 +import json +import logging + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +# Envelope keys that a business payload must never overwrite. +RESERVED_KEYS = frozenset( + { + "source", + "detail_type", + "time", + "account", + "region", + "id", + "resources", + "version", + } +) + + +def flatten(payload: dict) -> dict: + """Flatten one EventBridge envelope into a single-level document.""" + flat = { + "id": payload.get("id"), + "source": payload.get("source"), + "detail_type": payload.get("detail-type"), + "time": payload.get("time"), + "account": payload.get("account"), + "region": payload.get("region"), + "resources": payload.get("resources", []), + } + + detail = payload.get("detail") + if isinstance(detail, dict): + for key, value in detail.items(): + # Prefix rather than overwrite so envelope metadata stays truthful. + flat[f"detail_{key}" if key in RESERVED_KEYS else key] = value + elif detail is not None: + # Non-object detail (string, list, number) still needs to be indexed. + flat["detail"] = detail + + # Drop keys the producer never set so OpenSearch does not index nulls. + return {k: v for k, v in flat.items() if v is not None} + + +def handler(event, context): + output = [] + + for record in event["records"]: + record_id = record["recordId"] + try: + raw = base64.b64decode(record["data"]) + payload = json.loads(raw) + + if not isinstance(payload, dict): + raise ValueError(f"expected a JSON object, got {type(payload).__name__}") + + document = json.dumps(flatten(payload)) + "\n" + + output.append( + { + "recordId": record_id, + "result": "Ok", + "data": base64.b64encode(document.encode("utf-8")).decode("utf-8"), + } + ) + except Exception as exc: + # ProcessingFailed routes just this record to the S3 error prefix + # and lets the rest of the batch through. + logger.warning("Record %s failed to transform: %s", record_id, exc) + output.append( + { + "recordId": record_id, + "result": "ProcessingFailed", + "data": record["data"], + } + ) + + ok = sum(1 for r in output if r["result"] == "Ok") + logger.info("Transformed %d/%d records", ok, len(output)) + + return {"records": output}