From 1489c13725cee090b457ae74d1d2345d6f4875e2 Mon Sep 17 00:00:00 2001 From: Shiladitya Mandal Date: Wed, 27 May 2026 20:08:31 +0000 Subject: [PATCH] Add DynamoDB zero-downtime migration tools General-purpose zero-downtime table migration for DynamoDB using Streams, Export to S3, and conditional writes for conflict resolution. Package: tools/ddb_migration/ Components: - lambda/stream_replay.py: Stream replay with _migration_ts conflict resolution - glue/backfill_job.py: AWS Glue job for large tables (100+ GiB) - scripts/backfill.py: Standalone backfill for smaller tables - scripts/convergence_check.py: Pre-cutover verification - transform.py: Shared item transformation (customize for schema changes) - deploy.sh: One-command infrastructure setup - iam/policies.json: Least-privilege IAM templates (including cross-account) Use cases: - Migration to Global Tables with MRSC - Cross-account table migration - Key schema restructuring - Table consolidation Tested end-to-end with conflict resolution, tombstones, and MRSC strong consistency verification across 3 regions. --- tools/ddb_migration/README.md | 131 +++++++++ tools/ddb_migration/deploy.sh | 147 ++++++++++ tools/ddb_migration/glue/backfill_job.py | 178 +++++++++++ tools/ddb_migration/iam/policies.json | 95 ++++++ tools/ddb_migration/lambda/stream_replay.py | 107 +++++++ tools/ddb_migration/requirements.txt | 1 + tools/ddb_migration/scripts/backfill.py | 277 ++++++++++++++++++ .../scripts/convergence_check.py | 167 +++++++++++ tools/ddb_migration/transform.py | 40 +++ 9 files changed, 1143 insertions(+) create mode 100644 tools/ddb_migration/README.md create mode 100755 tools/ddb_migration/deploy.sh create mode 100644 tools/ddb_migration/glue/backfill_job.py create mode 100644 tools/ddb_migration/iam/policies.json create mode 100644 tools/ddb_migration/lambda/stream_replay.py create mode 100644 tools/ddb_migration/requirements.txt create mode 100644 tools/ddb_migration/scripts/backfill.py create mode 100644 tools/ddb_migration/scripts/convergence_check.py create mode 100644 tools/ddb_migration/transform.py diff --git a/tools/ddb_migration/README.md b/tools/ddb_migration/README.md new file mode 100644 index 00000000..84531a38 --- /dev/null +++ b/tools/ddb_migration/README.md @@ -0,0 +1,131 @@ +# DynamoDB Zero-Downtime Migration + +Zero-downtime migration between DynamoDB tables using DynamoDB Streams, Export to S3, and conditional writes for conflict resolution. + +## When to use this + +- Migrating to Global Tables with Multi-Region Strong Consistency (MRSC) +- Moving a table to a different AWS account +- Restructuring key schema or adding GSIs that require a new table +- Consolidating multiple tables into a single-table design +- Any scenario where you need a new table but can't stop writes + +## How it works + +Three overlapping phases: + +1. **Stream Capture & Bulk Copy** -- Enable streams, export to S3, backfill into target (parallel with stream replay) +2. **Stream Catch-up** -- Lambda replays live changes with conditional writes +3. **Convergence & Switchover** -- Verify consistency, pause briefly, flip routing + +Conflict resolution uses a `_migration_ts` attribute with conditional writes. Backfill writes `_migration_ts = 0` (lowest priority). Stream replay writes `_migration_ts = ApproximateCreationDateTime` (always wins over backfill). Later timestamps always win over earlier ones. + +## Package structure + +``` +ddb_migration/ + lambda/ + stream_replay.py # Lambda handler for DynamoDB Streams -> target table + glue/ + backfill_job.py # AWS Glue job for large tables (100+ GiB) + scripts/ + backfill.py # Standalone backfill script (tables < 100 GiB) + convergence_check.py # Pre-cutover verification + iam/ + policies.json # IAM policy templates for each component + transform.py # Shared item transformation (customize for schema changes) + deploy.sh # One-command infrastructure setup + requirements.txt +``` + +## Quick start + +```bash +# 1. Configure +export SOURCE_TABLE=OrdersV1 +export TARGET_TABLE=OrdersV2 +export PARTITION_KEY=pk +export SORT_KEY=sk +export REGION=us-east-1 + +# 2. Deploy infrastructure (creates target table, Lambda, alarms) +./deploy.sh + +# 3. Export source table +aws dynamodb export-table-to-point-in-time \ + --table-arn arn:aws:dynamodb:$REGION:$(aws sts get-caller-identity --query Account --output text):table/$SOURCE_TABLE \ + --s3-bucket $EXPORT_BUCKET --s3-prefix exports/ \ + --export-format DYNAMODB_JSON --region $REGION + +# 4. Run backfill (while stream replay Lambda is already processing) +export EXPORT_BUCKET=ddb-migration-$(aws sts get-caller-identity --query Account --output text)-$REGION +export EXPORT_PREFIX=exports/ +python scripts/backfill.py + +# 5. Verify convergence +python scripts/convergence_check.py + +# 6. Cutover: flip your application routing to TARGET_TABLE +``` + +## For large tables (100+ GiB) + +Use the Glue job instead of the standalone script: + +```bash +aws glue create-job \ + --name migration-backfill \ + --role AWSGlueServiceRole-Migration \ + --command '{"Name":"pythonshell","ScriptLocation":"s3://bucket/glue/backfill_job.py","PythonVersion":"3.9"}' \ + --default-arguments '{ + "--TARGET_TABLE":"OrdersV2", + "--PARTITION_KEY":"pk", + "--EXPORT_BUCKET":"my-bucket", + "--EXPORT_PREFIX":"exports/", + "--TARGET_REGION":"us-east-1", + "--LAMBDA_FUNCTION":"migration-stream-replay", + "--additional-python-modules":"boto3" + }' \ + --max-capacity 1.0 + +aws glue start-job-run --job-name migration-backfill +``` + +Or use [Bulk Executor](../bulk_executor) for the simple path (maintenance window, no conditional writes needed): +```bash +./bulk load-export --table OrdersV2 --s3-path "s3://bucket/exports/AWSDynamoDB/export-id" +``` + +## Schema changes + +If your target table has a different schema, edit `transform.py`: + +```python +def transform(item, source_event=None): + # Example: rename attribute + item['order_id'] = item.pop('orderId', item.get('order_id')) + # Example: add computed field + item['gsi1pk'] = f"TENANT#{item['tenant_id']}" + return item +``` + +Both the Lambda and backfill import this module, ensuring identical transformations. + +## Cross-account migration + +1. Add a resource-based policy on the target table (see `iam/policies.json` -> `CrossAccountTarget`) +2. Set `TARGET_REGION` and configure the Lambda to assume a role in the target account +3. The S3 export bucket needs a bucket policy allowing the target account to read + +## Prerequisites + +- Python 3.9+ +- boto3 +- Source table with PITR enabled +- AWS CLI v2 + +## Related + +- [Zero-Downtime Migration to MRSC](https://quip-amazon.com/4fhmAPBfxor6) -- MRSC-specific walkthrough +- [Bulk Executor](../bulk_executor) -- for simple-path migrations (maintenance window) +- [AWS Glue DynamoDB export blog](https://aws.amazon.com/blogs/database/filter-transform-and-load-your-dynamodb-table-exports-using-aws-glue/) diff --git a/tools/ddb_migration/deploy.sh b/tools/ddb_migration/deploy.sh new file mode 100755 index 00000000..9d669705 --- /dev/null +++ b/tools/ddb_migration/deploy.sh @@ -0,0 +1,147 @@ +#!/bin/bash +# Deploy infrastructure for DynamoDB zero-downtime migration. +# +# Creates: target table, stream replay Lambda, event source mapping, +# CloudWatch alarms, and optionally a Glue job for large tables. +# +# Usage: +# export SOURCE_TABLE=OrdersV1 +# export TARGET_TABLE=OrdersV2 +# export PARTITION_KEY=pk +# export REGION=us-east-1 +# ./deploy.sh +# +# For cross-account: also set TARGET_ACCOUNT and TARGET_ROLE + +set -euo pipefail + +REGION=${REGION:-us-east-1} +ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text) +SOURCE_TABLE=${SOURCE_TABLE:?SOURCE_TABLE required} +TARGET_TABLE=${TARGET_TABLE:?TARGET_TABLE required} +PARTITION_KEY=${PARTITION_KEY:-pk} +SORT_KEY=${SORT_KEY:-} +LAMBDA_NAME="migration-stream-replay" +EXPORT_BUCKET=${EXPORT_BUCKET:-ddb-migration-${ACCOUNT_ID}-${REGION}} + +echo "=== DynamoDB Zero-Downtime Migration Setup ===" +echo "Source: $SOURCE_TABLE | Target: $TARGET_TABLE | Region: $REGION" +echo "" + +# --- 1. Create target table --- +echo "1. Creating target table..." +KEY_SCHEMA="AttributeName=$PARTITION_KEY,KeyType=HASH" +ATTR_DEFS="AttributeName=$PARTITION_KEY,AttributeType=S" +if [ -n "${SORT_KEY:-}" ]; then + KEY_SCHEMA="$KEY_SCHEMA AttributeName=$SORT_KEY,KeyType=RANGE" + ATTR_DEFS="$ATTR_DEFS AttributeName=$SORT_KEY,AttributeType=S" +fi + +aws dynamodb create-table \ + --table-name "$TARGET_TABLE" \ + --attribute-definitions $ATTR_DEFS \ + --key-schema $KEY_SCHEMA \ + --billing-mode PAY_PER_REQUEST \ + --region "$REGION" > /dev/null 2>&1 || echo " (already exists)" +aws dynamodb wait table-exists --table-name "$TARGET_TABLE" --region "$REGION" +echo " Done." + +# --- 2. Enable streams on source --- +echo "2. Enabling streams on $SOURCE_TABLE..." +aws dynamodb update-table \ + --table-name "$SOURCE_TABLE" \ + --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \ + --region "$REGION" > /dev/null 2>&1 || echo " (already enabled)" +echo " Done." + +# --- 3. Create S3 bucket for export --- +echo "3. Creating export bucket ($EXPORT_BUCKET)..." +if [ "$REGION" = "us-east-1" ]; then + aws s3api create-bucket --bucket "$EXPORT_BUCKET" --region "$REGION" 2>/dev/null || true +else + aws s3api create-bucket --bucket "$EXPORT_BUCKET" --region "$REGION" \ + --create-bucket-configuration LocationConstraint="$REGION" 2>/dev/null || true +fi +echo " Done." + +# --- 4. Deploy Lambda --- +echo "4. Deploying stream replay Lambda..." +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +cd "$SCRIPT_DIR/lambda" +zip -j function.zip stream_replay.py ../transform.py > /dev/null + +aws lambda create-function \ + --function-name "$LAMBDA_NAME" \ + --runtime python3.12 \ + --handler stream_replay.handler \ + --role "arn:aws:iam::${ACCOUNT_ID}:role/migration-stream-replay-role" \ + --environment "Variables={TARGET_TABLE=$TARGET_TABLE,PARTITION_KEY=$PARTITION_KEY,TARGET_REGION=$REGION}" \ + --timeout 300 \ + --memory-size 512 \ + --zip-file fileb://function.zip \ + --region "$REGION" > /dev/null 2>&1 || \ + aws lambda update-function-code \ + --function-name "$LAMBDA_NAME" \ + --zip-file fileb://function.zip \ + --region "$REGION" > /dev/null + +rm -f function.zip +cd "$SCRIPT_DIR" +echo " Done." + +# --- 5. Create event source mapping --- +echo "5. Creating event source mapping..." +STREAM_ARN=$(aws dynamodb describe-table \ + --table-name "$SOURCE_TABLE" \ + --query 'Table.LatestStreamArn' --output text --region "$REGION") + +aws lambda create-event-source-mapping \ + --function-name "$LAMBDA_NAME" \ + --event-source-arn "$STREAM_ARN" \ + --starting-position TRIM_HORIZON \ + --batch-size 100 \ + --maximum-batching-window-in-seconds 5 \ + --bisect-batch-on-function-error \ + --maximum-retry-attempts 3 \ + --function-response-types ReportBatchItemFailures \ + --region "$REGION" > /dev/null 2>&1 || echo " (already exists)" +echo " Done." + +# --- 6. CloudWatch alarms --- +echo "6. Creating iterator age alarms..." +aws cloudwatch put-metric-alarm \ + --alarm-name "Migration-IteratorAge-Warning" \ + --metric-name IteratorAge \ + --namespace AWS/Lambda \ + --dimensions "Name=FunctionName,Value=$LAMBDA_NAME" \ + --statistic Maximum --period 60 --evaluation-periods 5 \ + --threshold 43200000 \ + --comparison-operator GreaterThanThreshold \ + --region "$REGION" 2>/dev/null + +aws cloudwatch put-metric-alarm \ + --alarm-name "Migration-IteratorAge-Critical" \ + --metric-name IteratorAge \ + --namespace AWS/Lambda \ + --dimensions "Name=FunctionName,Value=$LAMBDA_NAME" \ + --statistic Maximum --period 60 --evaluation-periods 5 \ + --threshold 72000000 \ + --comparison-operator GreaterThanThreshold \ + --region "$REGION" 2>/dev/null +echo " Done." + +echo "" +echo "=== Setup complete ===" +echo "" +echo "Next steps:" +echo " 1. Export: aws dynamodb export-table-to-point-in-time \\" +echo " --table-arn arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/$SOURCE_TABLE \\" +echo " --s3-bucket $EXPORT_BUCKET --s3-prefix exports/ \\" +echo " --export-format DYNAMODB_JSON --region $REGION" +echo "" +echo " 2. Backfill: export EXPORT_BUCKET=$EXPORT_BUCKET EXPORT_PREFIX=exports/" +echo " python scripts/backfill.py" +echo "" +echo " 3. Monitor: python scripts/convergence_check.py" +echo "" +echo " 4. Cutover: Flip your application routing to $TARGET_TABLE" diff --git a/tools/ddb_migration/glue/backfill_job.py b/tools/ddb_migration/glue/backfill_job.py new file mode 100644 index 00000000..9ea82aac --- /dev/null +++ b/tools/ddb_migration/glue/backfill_job.py @@ -0,0 +1,178 @@ +""" +AWS Glue job for parallel bulk import during DynamoDB zero-downtime migration. + +Reads DynamoDB export files from S3 and writes to the target table with +_migration_ts conditional writes. Designed to run as a Glue Python Shell job +for massive parallelism on large tables (100+ GiB). + +For tables under 100 GiB, use scripts/backfill.py instead (simpler, no Glue setup). + +Glue job parameters (passed via --additional-python-modules or job args): + --TARGET_TABLE Target DynamoDB table name + --PARTITION_KEY Partition key attribute name + --EXPORT_BUCKET S3 bucket containing the export + --EXPORT_PREFIX S3 prefix for the export + --TARGET_REGION Region of the target table + --MAX_WRITE_RATE Max WCU/s to consume (default: unlimited) + --LAMBDA_FUNCTION Stream replay Lambda name (for circuit breaker) +""" + +import sys +import boto3 +import json +import gzip +import random +import time +from decimal import Decimal +from concurrent.futures import ThreadPoolExecutor, as_completed +from awsglue.utils import getResolvedOptions +from botocore.exceptions import ClientError +from boto3.dynamodb.types import TypeDeserializer + +# Parse Glue job arguments +args = getResolvedOptions(sys.argv, [ + 'TARGET_TABLE', 'PARTITION_KEY', 'EXPORT_BUCKET', 'EXPORT_PREFIX', + 'TARGET_REGION', 'MAX_WRITE_RATE', 'LAMBDA_FUNCTION' +]) + +TARGET_TABLE = args['TARGET_TABLE'] +PARTITION_KEY = args['PARTITION_KEY'] +BUCKET = args['EXPORT_BUCKET'] +PREFIX = args['EXPORT_PREFIX'] +REGION = args.get('TARGET_REGION', 'us-east-1') +MAX_WRITE_RATE = int(args.get('MAX_WRITE_RATE', '0')) # 0 = unlimited +LAMBDA_FUNCTION = args.get('LAMBDA_FUNCTION', 'migration-stream-replay') +ITERATOR_AGE_PAUSE_HOURS = 18 + +s3 = boto3.client('s3', region_name=REGION) +dynamodb = boto3.resource('dynamodb', region_name=REGION) +cloudwatch = boto3.client('cloudwatch', region_name=REGION) +target_table = dynamodb.Table(TARGET_TABLE) +deserializer = TypeDeserializer() + +# Import transform (bundled with the job) +try: + from transform import transform +except ImportError: + def transform(item, source_event=None): + return item + + +def get_data_files(): + """Parse export manifest to get data file S3 keys.""" + paginator = s3.get_paginator('list_objects_v2') + data_files = [] + for page in paginator.paginate(Bucket=BUCKET, Prefix=PREFIX): + for obj in page.get('Contents', []): + if obj['Key'].endswith('.json.gz'): + data_files.append(obj['Key']) + return data_files + + +def should_pause(): + """Circuit breaker: pause if stream replay iterator age is too high.""" + try: + resp = cloudwatch.get_metric_statistics( + Namespace='AWS/Lambda', + MetricName='IteratorAge', + Dimensions=[{'Name': 'FunctionName', 'Value': LAMBDA_FUNCTION}], + StartTime=time.time() - 300, + EndTime=time.time(), + Period=60, + Statistics=['Maximum'] + ) + if resp['Datapoints']: + max_age_h = max(dp['Maximum'] for dp in resp['Datapoints']) / 3_600_000 + if max_age_h > ITERATOR_AGE_PAUSE_HOURS: + print(f"PAUSING: Iterator age {max_age_h:.1f}h > {ITERATOR_AGE_PAUSE_HOURS}h") + return True + except Exception as e: + print(f"Warning: iterator age check failed: {e}") + return False + + +def put_item_conditional(item, max_retries=8): + """PutItem with _migration_ts=0 conditional write and backoff.""" + item = transform(item, source_event=None) + if item is None: + return 'skipped' + + item['_migration_ts'] = 0 + for attempt in range(max_retries): + try: + target_table.put_item( + Item=item, + ConditionExpression='attribute_not_exists(#pk) OR #ts < :ts', + ExpressionAttributeNames={'#pk': PARTITION_KEY, '#ts': '_migration_ts'}, + ExpressionAttributeValues={':ts': 0} + ) + return 'written' + except ClientError as e: + code = e.response['Error']['Code'] + if code == 'ConditionalCheckFailedException': + return 'skipped' + elif code in ('ProvisionedThroughputExceededException', 'ThrottlingException'): + time.sleep(min(2 ** attempt * 0.1, 30) + random.uniform(0, 0.5)) + else: + raise + raise RuntimeError(f"Failed after {max_retries} retries") + + +def process_file(s3_key): + """Process one export data file.""" + obj = s3.get_object(Bucket=BUCKET, Key=s3_key) + content = gzip.decompress(obj['Body'].read()).decode('utf-8') + + items = [] + for line in content.strip().split('\n'): + if line: + record = json.loads(line) + item = {k: deserializer.deserialize(v) for k, v in record['Item'].items()} + items.append(item) + + random.shuffle(items) + written, skipped, errors = 0, 0, 0 + + for i, item in enumerate(items): + if i % 5000 == 0 and i > 0: + while should_pause(): + time.sleep(60) + try: + result = put_item_conditional(item) + if result == 'written': + written += 1 + else: + skipped += 1 + except Exception as e: + errors += 1 + print(f"ERROR: {e}") + + return len(items), written, skipped, errors + + +# Main execution +data_files = get_data_files() +print(f"Found {len(data_files)} data files") + +total_items, total_written, total_skipped, total_errors = 0, 0, 0, 0 +start = time.time() + +with ThreadPoolExecutor(max_workers=16) as executor: + futures = {executor.submit(process_file, f): f for f in data_files} + for future in as_completed(futures): + key = futures[future] + try: + count, written, skipped, errors = future.result() + total_items += count + total_written += written + total_skipped += skipped + total_errors += errors + rate = total_items / (time.time() - start) + print(f"Done {key}: {count} items ({written}w/{skipped}s/{errors}e) " + f"[total: {total_items:,} @ {rate:.0f}/s]") + except Exception as e: + print(f"FAILED {key}: {e}") + +elapsed = time.time() - start +print(f"\nBackfill complete: {total_items:,} items in {elapsed/60:.1f}min " + f"({total_written:,} written, {total_skipped:,} skipped, {total_errors:,} errors)") diff --git a/tools/ddb_migration/iam/policies.json b/tools/ddb_migration/iam/policies.json new file mode 100644 index 00000000..0b70b2f4 --- /dev/null +++ b/tools/ddb_migration/iam/policies.json @@ -0,0 +1,95 @@ +{ + "StreamReplayLambda": { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "ReadSourceStream", + "Effect": "Allow", + "Action": [ + "dynamodb:GetRecords", + "dynamodb:GetShardIterator", + "dynamodb:DescribeStream", + "dynamodb:ListStreams" + ], + "Resource": "arn:aws:dynamodb:REGION:ACCOUNT:table/SOURCE_TABLE/stream/*" + }, + { + "Sid": "WriteTarget", + "Effect": "Allow", + "Action": ["dynamodb:PutItem"], + "Resource": "arn:aws:dynamodb:REGION:ACCOUNT:table/TARGET_TABLE" + }, + { + "Sid": "Logs", + "Effect": "Allow", + "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"], + "Resource": "arn:aws:logs:REGION:ACCOUNT:log-group:/aws/lambda/migration-stream-replay:*" + } + ] + }, + "BackfillScript": { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "ReadExport", + "Effect": "Allow", + "Action": ["s3:GetObject", "s3:ListBucket"], + "Resource": ["arn:aws:s3:::EXPORT_BUCKET", "arn:aws:s3:::EXPORT_BUCKET/*"] + }, + { + "Sid": "WriteTarget", + "Effect": "Allow", + "Action": ["dynamodb:PutItem", "dynamodb:DescribeTable"], + "Resource": "arn:aws:dynamodb:REGION:ACCOUNT:table/TARGET_TABLE" + }, + { + "Sid": "MonitorIteratorAge", + "Effect": "Allow", + "Action": ["cloudwatch:GetMetricStatistics"], + "Resource": "*" + } + ] + }, + "CrossAccountTarget": { + "Description": "Resource-based policy on the TARGET table for cross-account migration", + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "AllowSourceAccountWrite", + "Effect": "Allow", + "Principal": {"AWS": "arn:aws:iam::SOURCE_ACCOUNT:role/migration-stream-replay-role"}, + "Action": ["dynamodb:PutItem", "dynamodb:DeleteItem"], + "Resource": "arn:aws:dynamodb:REGION:TARGET_ACCOUNT:table/TARGET_TABLE" + } + ] + }, + "GlueJob": { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "ReadExport", + "Effect": "Allow", + "Action": ["s3:GetObject", "s3:ListBucket"], + "Resource": ["arn:aws:s3:::EXPORT_BUCKET", "arn:aws:s3:::EXPORT_BUCKET/*"] + }, + { + "Sid": "WriteTarget", + "Effect": "Allow", + "Action": ["dynamodb:PutItem", "dynamodb:DescribeTable"], + "Resource": "arn:aws:dynamodb:REGION:ACCOUNT:table/TARGET_TABLE" + }, + { + "Sid": "MonitorIteratorAge", + "Effect": "Allow", + "Action": ["cloudwatch:GetMetricStatistics"], + "Resource": "*" + }, + { + "Sid": "GlueBaseline", + "Effect": "Allow", + "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"], + "Resource": "arn:aws:logs:REGION:ACCOUNT:log-group:/aws-glue/*" + } + ] + } +} diff --git a/tools/ddb_migration/lambda/stream_replay.py b/tools/ddb_migration/lambda/stream_replay.py new file mode 100644 index 00000000..2b77bdbc --- /dev/null +++ b/tools/ddb_migration/lambda/stream_replay.py @@ -0,0 +1,107 @@ +""" +DynamoDB Stream Replay Lambda for zero-downtime table migration. + +Processes stream records from the source table and writes to the target +using _migration_ts conditional writes for conflict resolution. + +Behaviors: +- INSERT/MODIFY: conditional PutItem with _migration_ts ordering +- REMOVE: writes a tombstone (prevents backfill from re-inserting) +- ConditionalCheckFailedException: safe to skip (newer version exists) +- Other errors: reports as batchItemFailure for Lambda retry + +Environment variables: + TARGET_TABLE - Name of the target table + PARTITION_KEY - Partition key attribute name (default: pk) + TARGET_REGION - Region of target table (default: same as Lambda) + TRANSFORM_MODULE - Optional path to custom transform module + +Deploy with: + --function-response-types ReportBatchItemFailures + --bisect-batch-on-function-error +""" + +import boto3 +import os +import logging +import importlib.util +from decimal import Decimal +from botocore.exceptions import ClientError +from boto3.dynamodb.types import TypeDeserializer + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +TARGET_TABLE_NAME = os.environ['TARGET_TABLE'] +PARTITION_KEY = os.environ.get('PARTITION_KEY', 'pk') +TARGET_REGION = os.environ.get('TARGET_REGION', os.environ.get('AWS_REGION', 'us-east-1')) + +dynamodb = boto3.resource('dynamodb', region_name=TARGET_REGION) +target_table = dynamodb.Table(TARGET_TABLE_NAME) +deserializer = TypeDeserializer() + +# Load transform module +try: + from transform import transform +except ImportError: + def transform(item, source_event=None): + return item + + +def handler(event, context): + """Process DynamoDB Stream records with conditional writes.""" + failed_records = [] + + for record in event['Records']: + try: + process_record(record) + except ClientError as e: + if e.response['Error']['Code'] == 'ConditionalCheckFailedException': + pass # Newer version exists -- safe to skip + else: + logger.error(f"ClientError: {e}") + failed_records.append(record) + except Exception as e: + logger.error(f"Error: {e}", exc_info=True) + failed_records.append(record) + + if failed_records: + return { + 'batchItemFailures': [ + {'itemIdentifier': r['eventID']} for r in failed_records + ] + } + + +def process_record(record): + """Apply a single stream record to the target table.""" + event_name = record['eventName'] + event_ts = Decimal(str(record['dynamodb']['ApproximateCreationDateTime'])) + keys_raw = record['dynamodb']['Keys'] + key = {k: deserializer.deserialize(v) for k, v in keys_raw.items()} + + if event_name in ('INSERT', 'MODIFY'): + new_image_raw = record['dynamodb']['NewImage'] + item = {k: deserializer.deserialize(v) for k, v in new_image_raw.items()} + + # Apply transform + item = transform(item, source_event=event_name) + if item is None: + return # Transform says skip + + item['_migration_ts'] = event_ts + target_table.put_item( + Item=item, + ConditionExpression='attribute_not_exists(#pk) OR #ts < :ts', + ExpressionAttributeNames={'#pk': PARTITION_KEY, '#ts': '_migration_ts'}, + ExpressionAttributeValues={':ts': event_ts} + ) + + elif event_name == 'REMOVE': + tombstone = {**key, '_tombstone': True, '_migration_ts': event_ts} + target_table.put_item( + Item=tombstone, + ConditionExpression='attribute_not_exists(#pk) OR #ts < :ts', + ExpressionAttributeNames={'#pk': PARTITION_KEY, '#ts': '_migration_ts'}, + ExpressionAttributeValues={':ts': event_ts} + ) diff --git a/tools/ddb_migration/requirements.txt b/tools/ddb_migration/requirements.txt new file mode 100644 index 00000000..1d54ad4c --- /dev/null +++ b/tools/ddb_migration/requirements.txt @@ -0,0 +1 @@ +boto3>=1.28.0 diff --git a/tools/ddb_migration/scripts/backfill.py b/tools/ddb_migration/scripts/backfill.py new file mode 100644 index 00000000..adb0372c --- /dev/null +++ b/tools/ddb_migration/scripts/backfill.py @@ -0,0 +1,277 @@ +""" +Parallel bulk import for DynamoDB MRSC migration. + +Reads exported data files from S3, writes items to the MRSC target table +with _migration_ts conditional writes to avoid overwriting stream-replicated data. + +Includes: +- Manifest parsing to discover data files +- Parallel file processing with concurrent.futures +- Conditional writes (_migration_ts = 0, lowest priority) for conflict safety +- Exponential backoff + jitter for throttling +- Circuit breaker that pauses if stream replay iterator age grows too high +- Randomized write order to avoid hot partitions + +Usage: + export SOURCE_TABLE=SourceTable + export TARGET_TABLE=TargetTable-MRSC + export PARTITION_KEY=pk + export REGION=us-east-1 + export EXPORT_BUCKET=my-migration-bucket + export EXPORT_PREFIX=exports/ + export MAX_WORKERS=16 + python backfill.py +""" + +import boto3 +import json +import gzip +import random +import time +import os +import sys +from concurrent.futures import ThreadPoolExecutor, as_completed +from botocore.exceptions import ClientError +from boto3.dynamodb.types import TypeDeserializer + +# --- Configuration --- +BUCKET = os.environ['EXPORT_BUCKET'] +EXPORT_PREFIX = os.environ.get('EXPORT_PREFIX', 'exports/') +TARGET_TABLE_NAME = os.environ['TARGET_TABLE'] +PARTITION_KEY = os.environ.get('PARTITION_KEY', 'pk') +MAX_WORKERS = int(os.environ.get('MAX_WORKERS', '16')) +REGION = os.environ.get('REGION', 'us-east-1') +LAMBDA_FUNCTION_NAME = os.environ.get('LAMBDA_FUNCTION_NAME', 'MRSCStreamReplay') + +# Iterator age threshold (in hours) to pause backfill +ITERATOR_AGE_PAUSE_THRESHOLD_HOURS = 18 + +s3 = boto3.client('s3', region_name=REGION) +dynamodb = boto3.resource('dynamodb', region_name=REGION) +cloudwatch = boto3.client('cloudwatch', region_name=REGION) +target_table = dynamodb.Table(TARGET_TABLE_NAME) +deserializer = TypeDeserializer() + + +def get_data_file_keys_from_manifest(bucket, prefix): + """ + Parse the DynamoDB export manifest to get the list of data file S3 keys. + + The export creates a timestamped subdirectory containing: + - manifest-summary.json (points to the manifest file) + - manifest file (JSONL with one entry per data file) + - data/ directory with .json.gz files + """ + # Find the export directory (DynamoDB creates a timestamped subfolder) + response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, Delimiter='/') + export_dirs = [p['Prefix'] for p in response.get('CommonPrefixes', [])] + + if not export_dirs: + raise ValueError(f"No export directories found under s3://{bucket}/{prefix}") + + # Use the most recent export directory + export_dir = sorted(export_dirs)[-1] + print(f"Using export directory: s3://{bucket}/{export_dir}") + + # Read manifest-summary.json + summary_key = f"{export_dir}manifest-summary.json" + summary = json.loads( + s3.get_object(Bucket=bucket, Key=summary_key)['Body'].read() + ) + print(f"Export status: {summary.get('exportStatus', 'unknown')}") + print(f"Item count: {summary.get('itemCount', 'unknown')}") + + # Read the manifest files list + manifest_key = summary['manifestFilesS3Key'] + manifest_content = s3.get_object(Bucket=bucket, Key=manifest_key)['Body'].read() + + data_files = [] + for line in manifest_content.decode('utf-8').strip().split('\n'): + entry = json.loads(line) + data_files.append(entry['dataFileS3Key']) + + print(f"Found {len(data_files)} data files in export manifest") + return data_files + + +def deserialize_dynamodb_json(dynamo_item): + """Convert DynamoDB JSON format to Python-native types.""" + return {k: deserializer.deserialize(v) for k, v in dynamo_item.items()} + + +def put_item_with_retry(item, max_retries=8): + """ + PutItem with _migration_ts conditional write and exponential backoff. + + Backfill always writes _migration_ts = 0 (lowest priority). Stream replay + writes with ApproximateCreationDateTime (always > 0), so stream data + always wins over backfill data. + + Returns: + 'written' if the item was written + 'skipped' if a newer version already exists (written by stream replay) + """ + item['_migration_ts'] = 0 + for attempt in range(max_retries): + try: + target_table.put_item( + Item=item, + ConditionExpression='attribute_not_exists(#pk) OR #ts < :ts', + ExpressionAttributeNames={'#pk': PARTITION_KEY, '#ts': '_migration_ts'}, + ExpressionAttributeValues={':ts': 0} + ) + return 'written' + except ClientError as e: + code = e.response['Error']['Code'] + if code == 'ConditionalCheckFailedException': + # A newer version exists (stream replay wrote it) — skip safely + return 'skipped' + elif code in ('ProvisionedThroughputExceededException', + 'ThrottlingException'): + wait = min(2 ** attempt * 0.1, 30) + random.uniform(0, 0.5) + time.sleep(wait) + else: + raise + + raise RuntimeError( + f"Failed after {max_retries} retries for item " + f"{item.get(PARTITION_KEY, 'unknown_key')}" + ) + + +def should_pause_backfill(): + """ + Circuit breaker: returns True if stream replay iterator age is dangerously high. + + Checks the IteratorAge CloudWatch metric for the stream replay Lambda. + If the Lambda is more than ITERATOR_AGE_PAUSE_THRESHOLD_HOURS behind, + we pause backfill to free write capacity. + """ + try: + response = cloudwatch.get_metric_statistics( + Namespace='AWS/Lambda', + MetricName='IteratorAge', + Dimensions=[{ + 'Name': 'FunctionName', + 'Value': LAMBDA_FUNCTION_NAME + }], + StartTime=time.time() - 300, + EndTime=time.time(), + Period=60, + Statistics=['Maximum'] + ) + + if response['Datapoints']: + max_age_ms = max(dp['Maximum'] for dp in response['Datapoints']) + max_age_hours = max_age_ms / 3_600_000 + + if max_age_hours > ITERATOR_AGE_PAUSE_THRESHOLD_HOURS: + print(f"⚠️ PAUSING: Stream replay iterator age is {max_age_hours:.1f}h " + f"(threshold: {ITERATOR_AGE_PAUSE_THRESHOLD_HOURS}h)") + return True + elif max_age_hours > 12: + print(f"⚠️ WARNING: Stream replay iterator age is {max_age_hours:.1f}h") + + except Exception as e: + print(f"Warning: could not check iterator age: {e}") + + return False + + +def process_data_file(s3_key): + """Process a single export data file — read items and write to target table.""" + response = s3.get_object(Bucket=BUCKET, Key=s3_key) + + if s3_key.endswith('.gz'): + content = gzip.decompress(response['Body'].read()).decode('utf-8') + else: + content = response['Body'].read().decode('utf-8') + + items = [] + for line in content.strip().split('\n'): + if not line: + continue + record = json.loads(line) + item = deserialize_dynamodb_json(record['Item']) + items.append(item) + + # Randomize to distribute writes across partitions + random.shuffle(items) + + written = 0 + skipped = 0 + errors = 0 + + for i, item in enumerate(items): + # Circuit breaker check every 1000 items + if i % 1000 == 0 and i > 0: + while should_pause_backfill(): + time.sleep(60) + + try: + result = put_item_with_retry(item) + if result == 'written': + written += 1 + else: + skipped += 1 + except Exception as e: + errors += 1 + print(f" ERROR in {s3_key}: {e}") + + return len(items), written, skipped, errors + + +def run_parallel_import(): + """Import all data files in parallel.""" + manifest_files = get_data_file_keys_from_manifest(BUCKET, EXPORT_PREFIX) + + total_items = 0 + total_written = 0 + total_skipped = 0 + total_errors = 0 + failed_files = [] + + start_time = time.time() + + with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + future_to_file = { + executor.submit(process_data_file, f): f + for f in manifest_files + } + + for future in as_completed(future_to_file): + file_key = future_to_file[future] + try: + count, written, skipped, errors = future.result() + total_items += count + total_written += written + total_skipped += skipped + total_errors += errors + elapsed = time.time() - start_time + rate = total_items / elapsed if elapsed > 0 else 0 + print(f"✓ {file_key}: {count} items " + f"({written} written, {skipped} skipped, {errors} errors) " + f"[total: {total_items:,} @ {rate:.0f} items/sec]") + except Exception as e: + print(f"✗ FAILED {file_key}: {e}") + failed_files.append(file_key) + + elapsed = time.time() - start_time + print(f"\n{'='*60}") + print(f"Backfill complete in {elapsed/60:.1f} minutes") + print(f" Total items: {total_items:,}") + print(f" Written: {total_written:,}") + print(f" Skipped: {total_skipped:,} (already existed via stream replay)") + print(f" Errors: {total_errors:,}") + print(f" Failed files: {len(failed_files)}") + print(f" Avg rate: {total_items/elapsed:.0f} items/sec") + + if failed_files: + print(f"\nFAILED files — retry these manually:") + for f in failed_files: + print(f" {f}") + sys.exit(1) + + +if __name__ == '__main__': + run_parallel_import() diff --git a/tools/ddb_migration/scripts/convergence_check.py b/tools/ddb_migration/scripts/convergence_check.py new file mode 100644 index 00000000..87edb55a --- /dev/null +++ b/tools/ddb_migration/scripts/convergence_check.py @@ -0,0 +1,167 @@ +""" +Pre-cutover convergence verification for MRSC migration. + +Checks that the migration is safe to cut over: +1. Stream replay iterator age is near zero (Lambda is caught up) +2. DLQ is empty (no failed records) +3. Item counts are within tolerance + +Usage: + export TARGET_TABLE=TargetTable-MRSC + export SOURCE_TABLE=SourceTable + export REGION=us-east-1 + export DLQ_URL=https://sqs.us-east-1.amazonaws.com/123456789012/migration-dlq + export LAMBDA_FUNCTION_NAME=MRSCStreamReplay + python convergence_check.py +""" + +import boto3 +import time +import sys +import os + +REGION = os.environ.get('REGION', 'us-east-1') +SOURCE_TABLE = os.environ['SOURCE_TABLE'] +TARGET_TABLE = os.environ['TARGET_TABLE'] +DLQ_URL = os.environ['DLQ_URL'] +LAMBDA_FUNCTION_NAME = os.environ.get('LAMBDA_FUNCTION_NAME', 'MRSCStreamReplay') + +cloudwatch = boto3.client('cloudwatch', region_name=REGION) +dynamodb = boto3.client('dynamodb', region_name=REGION) +sqs = boto3.client('sqs', region_name=REGION) + + +def check_iterator_age(max_wait_seconds=600): + """ + Wait for stream replay Lambda to be fully caught up. + + Iterator age = how far behind the Lambda is from the tip of the stream. + An age of 0 means events are being processed in real time. + + Returns True if caught up within timeout. + """ + print("Checking stream replay iterator age...") + start = time.time() + + while time.time() - start < max_wait_seconds: + response = cloudwatch.get_metric_statistics( + Namespace='AWS/Lambda', + MetricName='IteratorAge', + Dimensions=[{ + 'Name': 'FunctionName', + 'Value': LAMBDA_FUNCTION_NAME + }], + StartTime=time.time() - 120, + EndTime=time.time(), + Period=60, + Statistics=['Maximum'] + ) + + if response['Datapoints']: + max_age_ms = max(dp['Maximum'] for dp in response['Datapoints']) + print(f" Iterator age: {max_age_ms/1000:.1f}s") + + if max_age_ms < 1000: # < 1 second + print(" ✓ Stream replay is caught up (processing in real time)") + return True + else: + print(" No datapoints yet, waiting...") + + time.sleep(10) + + print(" ✗ Stream replay did not converge within timeout") + return False + + +def check_dlq_empty(): + """Verify no unprocessed records in the dead-letter queue.""" + print("Checking DLQ...") + attrs = sqs.get_queue_attributes( + QueueUrl=DLQ_URL, + AttributeNames=['ApproximateNumberOfMessages', 'ApproximateNumberOfMessagesNotVisible'] + ) + visible = int(attrs['Attributes']['ApproximateNumberOfMessages']) + in_flight = int(attrs['Attributes']['ApproximateNumberOfMessagesNotVisible']) + total = visible + in_flight + + if total == 0: + print(" ✓ DLQ is empty — no failed records") + return True + else: + print(f" ✗ DLQ has {total} messages ({visible} visible, {in_flight} in-flight)") + print(" Investigate these records before proceeding with cutover") + return False + + +def check_item_counts(): + """ + Compare approximate item counts between source and target. + + Note: DynamoDB ItemCount updates approximately every 6 hours. + This is a sanity check, not a precise validation. + """ + print("Checking item counts (approximate)...") + source = dynamodb.describe_table(TableName=SOURCE_TABLE) + target = dynamodb.describe_table(TableName=TARGET_TABLE) + + src_count = source['Table']['ItemCount'] + tgt_count = target['Table']['ItemCount'] + + print(f" Source: ~{src_count:,} items") + print(f" Target: ~{tgt_count:,} items") + + if src_count == 0: + print(" ⚠️ Source count is 0 — ItemCount may not be populated yet") + return True + + diff_pct = abs(src_count - tgt_count) / src_count * 100 + + if diff_pct < 5: + print(f" ✓ Counts within {diff_pct:.1f}% (ItemCount updates every ~6h)") + return True + else: + print(f" ⚠️ Counts differ by {diff_pct:.1f}% — may be stale, or backfill incomplete") + print(" Consider running a precise count via Scan if concerned") + return True # Don't block on approximate metric + + +def main(): + print("=" * 60) + print("MRSC Migration — Pre-Cutover Convergence Check") + print("=" * 60) + print() + + results = [] + + results.append(('Iterator Age', check_iterator_age())) + print() + results.append(('DLQ Empty', check_dlq_empty())) + print() + results.append(('Item Counts', check_item_counts())) + print() + + print("=" * 60) + print("Results:") + all_passed = True + for name, passed in results: + status = "✓ PASS" if passed else "✗ FAIL" + print(f" {status}: {name}") + if not passed: + all_passed = False + + print() + if all_passed: + print("✓ All checks passed — safe to proceed with cutover") + print() + print("Next steps:") + print(" 1. Pause writes briefly (~5 seconds)") + print(" 2. Wait for in-flight stream records to drain") + print(" 3. Flip your feature flag to route traffic to TargetTable-MRSC") + print(" 4. Resume writes") + else: + print("✗ Some checks failed — do NOT proceed with cutover") + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/tools/ddb_migration/transform.py b/tools/ddb_migration/transform.py new file mode 100644 index 00000000..500d36d2 --- /dev/null +++ b/tools/ddb_migration/transform.py @@ -0,0 +1,40 @@ +""" +Shared item transformation for DynamoDB zero-downtime migration. + +Both the backfill script and stream replay Lambda import this module +to ensure identical transformation logic. If your migration requires +schema changes, modify the transform() function here. +""" + +from decimal import Decimal + + +def transform(item, source_event=None): + """ + Transform an item before writing to the target table. + + Override this function for schema migrations. Both the backfill + and stream replay call this with every item before writing. + + Args: + item: Dict of the DynamoDB item (Python-native types) + source_event: For stream replay, the event name ('INSERT', 'MODIFY', 'REMOVE'). + For backfill, None. + + Returns: + Transformed item dict, or None to skip this item. + + Example - rename an attribute: + item['order_id'] = item.pop('orderId', item.get('order_id')) + return item + + Example - add a computed field: + item['search_key'] = f"{item['tenant']}#{item['created_at']}" + return item + + Example - filter out items: + if item.get('status') == 'DELETED': + return None + return item + """ + return item