From b8ccb24ee083262503ac3e80e1ceaf2b6e126802 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:32:24 +0530 Subject: [PATCH 01/21] start with creating a migration config to be used for migration s3 between accounts --- hq_s3_migration/config.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) create mode 100644 hq_s3_migration/config.py diff --git a/hq_s3_migration/config.py b/hq_s3_migration/config.py new file mode 100644 index 0000000000..eef14481a0 --- /dev/null +++ b/hq_s3_migration/config.py @@ -0,0 +1,26 @@ +from dataclasses import dataclass + +@dataclass +class MigrationConfig: + """Configuration for S3 cross-account migration.""" + source_profile: str + dest_profile: str + source_bucket: str + dest_bucket: str + source_account_id: str + dest_account_id: str + region: str + replication_role_name: str + datasync_role_name: str + enable_rtc: bool = True # S3 Replication Time Control (15-min SLA) + enable_delete_replication: bool = True + + +ACCOUNT_IDS = { + 'production': '051428382917', + 'staging': '737236193635', + 'backup-production': '213307118311', + 'dimagi':'437781348816' +} + +ACCOUNT_NAMES = {v: k for k, v in ACCOUNT_IDS.items()} From 25216ed9986ce681cf0a5d728bc6d5aaf70aa740 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:33:28 +0530 Subject: [PATCH 02/21] orchestrator to call aws services for both source and destination accounts --- hq_s3_migration/orchestrator.py | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 hq_s3_migration/orchestrator.py diff --git a/hq_s3_migration/orchestrator.py b/hq_s3_migration/orchestrator.py new file mode 100644 index 0000000000..729dc5f6b5 --- /dev/null +++ b/hq_s3_migration/orchestrator.py @@ -0,0 +1,29 @@ +import boto3 + +from .config import MigrationConfig + + +class S3MigrationContext: + """Thin container for boto3 sessions and clients used across migration phases.""" + + def __init__(self, config: MigrationConfig): + self.config = config + + # Initialize boto3 sessions + self.source_session = boto3.Session( + profile_name=config.source_profile, + region_name=config.region + ) + self.dest_session = boto3.Session( + profile_name=config.dest_profile, + region_name=config.region + ) + + # Initialize clients + self.source_s3 = self.source_session.client('s3') + self.dest_s3 = self.dest_session.client('s3') + self.source_iam = self.source_session.client('iam') + self.dest_iam = self.dest_session.client('iam') + self.source_datasync = self.source_session.client('datasync') + self.source_sts = self.source_session.client('sts') + self.dest_sts = self.dest_session.client('sts') From 80dc5376fd44006969f13950c33793efcdedaa7e Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:34:06 +0530 Subject: [PATCH 03/21] put __init__ --- hq_s3_migration/__init__.py | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 hq_s3_migration/__init__.py diff --git a/hq_s3_migration/__init__.py b/hq_s3_migration/__init__.py new file mode 100644 index 0000000000..e11d478ddf --- /dev/null +++ b/hq_s3_migration/__init__.py @@ -0,0 +1,5 @@ +"""S3 Cross-Account Migration Tool — hybrid DataSync + Live Replication.""" + +from .orchestrator import S3MigrationContext + +__all__ = ["S3MigrationContext"] From cbc5fca127b998589ab53b1ce07fd6029d2cb904 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:36:01 +0530 Subject: [PATCH 04/21] add different roles and policies that needs to be added to the source and destination envs --- hq_s3_migration/policies/__init__.py | 20 ++++++++ .../policies/datasync_bucket_stmt.json | 25 ++++++++++ hq_s3_migration/policies/datasync_role.json | 46 +++++++++++++++++++ hq_s3_migration/policies/datasync_trust.json | 16 +++++++ .../policies/destination_bucket.json | 33 +++++++++++++ .../policies/replication_role.json | 43 +++++++++++++++++ .../policies/replication_trust.json | 12 +++++ 7 files changed, 195 insertions(+) create mode 100644 hq_s3_migration/policies/__init__.py create mode 100644 hq_s3_migration/policies/datasync_bucket_stmt.json create mode 100644 hq_s3_migration/policies/datasync_role.json create mode 100644 hq_s3_migration/policies/datasync_trust.json create mode 100644 hq_s3_migration/policies/destination_bucket.json create mode 100644 hq_s3_migration/policies/replication_role.json create mode 100644 hq_s3_migration/policies/replication_trust.json diff --git a/hq_s3_migration/policies/__init__.py b/hq_s3_migration/policies/__init__.py new file mode 100644 index 0000000000..993e0444d0 --- /dev/null +++ b/hq_s3_migration/policies/__init__.py @@ -0,0 +1,20 @@ +import json +from pathlib import Path + +_POLICIES_DIR = Path(__file__).parent + + +def render_policy(template_name: str, **kwargs) -> dict: + """Load a JSON policy template and substitute {placeholder} variables. + + Args: + template_name: Filename (e.g. "replication_role.json") inside policies/. + **kwargs: Placeholder values to substitute. + + Returns: + Parsed dict with all placeholders replaced. + """ + text = (_POLICIES_DIR / template_name).read_text() + for key, value in kwargs.items(): + text = text.replace(f"{{{key}}}", value) + return json.loads(text) diff --git a/hq_s3_migration/policies/datasync_bucket_stmt.json b/hq_s3_migration/policies/datasync_bucket_stmt.json new file mode 100644 index 0000000000..9ac8f8282e --- /dev/null +++ b/hq_s3_migration/policies/datasync_bucket_stmt.json @@ -0,0 +1,25 @@ +{ + "Sid": "AllowDataSyncAccess", + "Effect": "Allow", + "Principal": { + "AWS": "{datasync_role_arn}" + }, + "Action": [ + "s3:GetBucketLocation", + "s3:ListBucket", + "s3:ListBucketMultipartUploads", + "s3:GetObject", + "s3:GetObjectTagging", + "s3:GetObjectVersion", + "s3:GetObjectVersionTagging", + "s3:PutObject", + "s3:PutObjectTagging", + "s3:DeleteObject", + "s3:AbortMultipartUpload", + "s3:ListMultipartUploadParts" + ], + "Resource": [ + "arn:aws:s3:::{dest_bucket}", + "arn:aws:s3:::{dest_bucket}/*" + ] +} diff --git a/hq_s3_migration/policies/datasync_role.json b/hq_s3_migration/policies/datasync_role.json new file mode 100644 index 0000000000..0a0e1b6cb9 --- /dev/null +++ b/hq_s3_migration/policies/datasync_role.json @@ -0,0 +1,46 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "ReadSource", + "Effect": "Allow", + "Action": [ + "s3:GetBucketLocation", + "s3:ListBucket", + "s3:ListBucketMultipartUploads", + "s3:GetObject", + "s3:GetObjectTagging", + "s3:GetObjectVersion", + "s3:GetObjectVersionTagging", + "s3:ListMultipartUploadParts", + "s3:AbortMultipartUpload" + ], + "Resource": [ + "arn:aws:s3:::{source_bucket}", + "arn:aws:s3:::{source_bucket}/*" + ] + }, + { + "Sid": "WriteDestination", + "Effect": "Allow", + "Action": [ + "s3:GetBucketLocation", + "s3:ListBucket", + "s3:ListBucketMultipartUploads", + "s3:GetObject", + "s3:GetObjectTagging", + "s3:GetObjectVersion", + "s3:GetObjectVersionTagging", + "s3:PutObject", + "s3:PutObjectTagging", + "s3:DeleteObject", + "s3:ListMultipartUploadParts", + "s3:AbortMultipartUpload" + ], + "Resource": [ + "arn:aws:s3:::{dest_bucket}", + "arn:aws:s3:::{dest_bucket}/*" + ] + } + ] +} diff --git a/hq_s3_migration/policies/datasync_trust.json b/hq_s3_migration/policies/datasync_trust.json new file mode 100644 index 0000000000..4594a9368d --- /dev/null +++ b/hq_s3_migration/policies/datasync_trust.json @@ -0,0 +1,16 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": { + "Service": [ + "datasync.amazonaws.com" + ] + }, + "Action": [ + "sts:AssumeRole" + ] + } + ] +} diff --git a/hq_s3_migration/policies/destination_bucket.json b/hq_s3_migration/policies/destination_bucket.json new file mode 100644 index 0000000000..dfc7649cb2 --- /dev/null +++ b/hq_s3_migration/policies/destination_bucket.json @@ -0,0 +1,33 @@ +{ + "Version": "2012-10-17", + "Id": "PolicyForCrossAccountReplication", + "Statement": [ + { + "Sid": "AllowReplicationFromSource", + "Effect": "Allow", + "Principal": { + "AWS": "{role_arn}" + }, + "Action": [ + "s3:List*", + "s3:GetBucketVersioning", + "s3:PutBucketVersioning", + "s3:ReplicateDelete", + "s3:ReplicateObject" + ], + "Resource": [ + "arn:aws:s3:::{dest_bucket}", + "arn:aws:s3:::{dest_bucket}/*" + ] + }, + { + "Sid": "AllowBucketOwnerOverride", + "Effect": "Allow", + "Principal": { + "AWS": "{role_arn}" + }, + "Action": "s3:ObjectOwnerOverrideToBucketOwner", + "Resource": "arn:aws:s3:::{dest_bucket}/*" + } + ] +} diff --git a/hq_s3_migration/policies/replication_role.json b/hq_s3_migration/policies/replication_role.json new file mode 100644 index 0000000000..3054dcb225 --- /dev/null +++ b/hq_s3_migration/policies/replication_role.json @@ -0,0 +1,43 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "GetSourceBucketConfiguration", + "Effect": "Allow", + "Action": [ + "s3:InitiateReplication", + "s3:ListBucket", + "s3:GetBucketLocation", + "s3:GetBucketAcl", + "s3:GetReplicationConfiguration", + "s3:GetObjectVersionForReplication", + "s3:GetObjectVersionAcl", + "s3:GetObjectVersionTagging", + "s3:PutInventoryConfiguration" + ], + "Resource": [ + "arn:aws:s3:::{source_bucket}", + "arn:aws:s3:::{source_bucket}/*" + ] + }, + { + "Sid": "ReplicateToDestinationBucket", + "Effect": "Allow", + "Action": [ + "s3:List*", + "s3:GetObject", + "s3:PutObject", + "s3:PutObjectAcl", + "s3:PutObjectTagging", + "s3:ReplicateObject", + "s3:ReplicateDelete", + "s3:ReplicateTags", + "s3:ObjectOwnerOverrideToBucketOwner" + ], + "Resource": [ + "arn:aws:s3:::{dest_bucket}", + "arn:aws:s3:::{dest_bucket}/*" + ] + } + ] +} diff --git a/hq_s3_migration/policies/replication_trust.json b/hq_s3_migration/policies/replication_trust.json new file mode 100644 index 0000000000..873e3abe14 --- /dev/null +++ b/hq_s3_migration/policies/replication_trust.json @@ -0,0 +1,12 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": { + "Service": "s3.amazonaws.com" + }, + "Action": "sts:AssumeRole" + } + ] +} From 78301587b59db78c4cec0154b71a7ed194f3c9aa Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:36:45 +0530 Subject: [PATCH 05/21] add iam util to view and create policies on aws accounts --- hq_s3_migration/iam.py | 227 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 227 insertions(+) create mode 100644 hq_s3_migration/iam.py diff --git a/hq_s3_migration/iam.py b/hq_s3_migration/iam.py new file mode 100644 index 0000000000..98d5d5f6ad --- /dev/null +++ b/hq_s3_migration/iam.py @@ -0,0 +1,227 @@ +import json +import time +from typing import Optional + +from botocore.exceptions import ClientError + +from .orchestrator import S3MigrationContext +from .policies import render_policy + + +def print_iam_policies(ctx: S3MigrationContext): + """Print all IAM policies for manual setup.""" + cfg = ctx.config + + role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.replication_role_name}" + datasync_role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.datasync_role_name}" + + print("\n" + "=" * 60) + print("IAM POLICIES FOR MANUAL SETUP") + print("=" * 60) + + print("\n" + "-" * 40) + print(f"1. REPLICATION ROLE TRUST POLICY (Source Account: {cfg.source_account_id})") + print(f" Role Name: {cfg.replication_role_name}") + print("-" * 40) + print(json.dumps(render_policy("replication_trust.json"), indent=2)) + + print("\n" + "-" * 40) + print(f"2. REPLICATION ROLE POLICY (Source Account: {cfg.source_account_id})") + print("-" * 40) + print(json.dumps( + render_policy("replication_role.json", + source_bucket=cfg.source_bucket, + dest_bucket=cfg.dest_bucket), + indent=2)) + + print("\n" + "-" * 40) + print(f"3. DATASYNC ROLE TRUST POLICY (Source Account: {cfg.source_account_id})") + print(f" Role Name: {cfg.datasync_role_name}") + print("-" * 40) + print(json.dumps(render_policy("datasync_trust.json"), indent=2)) + + print("\n" + "-" * 40) + print(f"4. DATASYNC ROLE POLICY (Source Account: {cfg.source_account_id})") + print("-" * 40) + print(json.dumps( + render_policy("datasync_role.json", + source_bucket=cfg.source_bucket, + dest_bucket=cfg.dest_bucket), + indent=2)) + + print("\n" + "-" * 40) + print(f"5. DESTINATION BUCKET POLICY (Dest Account: {cfg.dest_account_id})") + print(f" Bucket: {cfg.dest_bucket}") + print("-" * 40) + dest_policy = render_policy("destination_bucket.json", + role_arn=role_arn, + dest_bucket=cfg.dest_bucket) + datasync_stmt = render_policy("datasync_bucket_stmt.json", + datasync_role_arn=datasync_role_arn, + dest_bucket=cfg.dest_bucket) + dest_policy['Statement'].append(datasync_stmt) + print(json.dumps(dest_policy, indent=2)) + + +def create_replication_role(ctx: S3MigrationContext) -> Optional[str]: + """Create IAM role for S3 replication in source account.""" + cfg = ctx.config + print(f"\nCreating replication role '{cfg.replication_role_name}'...") + + trust_policy = render_policy("replication_trust.json") + role_policy = render_policy("replication_role.json", + source_bucket=cfg.source_bucket, + dest_bucket=cfg.dest_bucket) + + try: + response = ctx.source_iam.create_role( + RoleName=cfg.replication_role_name, + AssumeRolePolicyDocument=json.dumps(trust_policy), + Description="Role for S3 cross-account replication" + ) + role_arn = response['Role']['Arn'] + print(f" Created role: {role_arn}") + + ctx.source_iam.put_role_policy( + RoleName=cfg.replication_role_name, + PolicyName="S3ReplicationPolicy", + PolicyDocument=json.dumps(role_policy) + ) + print(f" Attached replication policy") + + print(f" Waiting for role propagation...") + time.sleep(10) + + return role_arn + except ClientError as e: + if e.response['Error']['Code'] == 'EntityAlreadyExists': + print(f" Role already exists") + response = ctx.source_iam.get_role(RoleName=cfg.replication_role_name) + return response['Role']['Arn'] + print(f" ERROR: {e}") + return None + + +def create_datasync_role(ctx: S3MigrationContext) -> Optional[str]: + """Create IAM role for DataSync in source account.""" + cfg = ctx.config + print(f"\nCreating DataSync role '{cfg.datasync_role_name}'...") + + trust_policy = render_policy("datasync_trust.json") + role_policy = render_policy("datasync_role.json", + source_bucket=cfg.source_bucket, + dest_bucket=cfg.dest_bucket) + + try: + response = ctx.source_iam.create_role( + RoleName=cfg.datasync_role_name, + AssumeRolePolicyDocument=json.dumps(trust_policy), + Description="Role for DataSync S3 to S3 transfer" + ) + role_arn = response['Role']['Arn'] + print(f" Created role: {role_arn}") + + ctx.source_iam.put_role_policy( + RoleName=cfg.datasync_role_name, + PolicyName="DataSyncS3Policy", + PolicyDocument=json.dumps(role_policy) + ) + print(f" Attached DataSync policy") + + print(f" Waiting for role propagation...") + time.sleep(10) + + return role_arn + except ClientError as e: + if e.response['Error']['Code'] == 'EntityAlreadyExists': + print(f" Role already exists, updating policy...") + ctx.source_iam.put_role_policy( + RoleName=cfg.datasync_role_name, + PolicyName="DataSyncS3Policy", + PolicyDocument=json.dumps(role_policy) + ) + response = ctx.source_iam.get_role(RoleName=cfg.datasync_role_name) + return response['Role']['Arn'] + print(f" ERROR: {e}") + return None + + + +def apply_destination_bucket_policy(ctx: S3MigrationContext) -> bool: + """Apply bucket policy to destination bucket, with option to merge or replace.""" + cfg = ctx.config + print(f"\nApplying bucket policy to destination bucket...") + + role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.replication_role_name}" + datasync_role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.datasync_role_name}" + + try: + # Check for existing policy + existing_policy = None + try: + existing = ctx.dest_s3.get_bucket_policy(Bucket=cfg.dest_bucket) + existing_policy = json.loads(existing['Policy']) + except ClientError as e: + if e.response['Error']['Code'] != 'NoSuchBucketPolicy': + raise + + # Build the statements we need + dest_policy = render_policy("destination_bucket.json", + role_arn=role_arn, + dest_bucket=cfg.dest_bucket) + datasync_stmt = render_policy("datasync_bucket_stmt.json", + datasync_role_arn=datasync_role_arn, + dest_bucket=cfg.dest_bucket) + new_statements = dest_policy['Statement'] + new_statements.append(datasync_stmt) + + if existing_policy: + existing_statements = existing_policy.get('Statement', []) + print(f"\n EXISTING bucket policy ({len(existing_statements)} statements):") + print(json.dumps(existing_policy, indent=2)) + + print(f"\n NEW statements to add ({len(new_statements)} statements):") + print(json.dumps(new_statements, indent=2)) + + print(f"\n Options:") + print(f" 1. APPEND - Add new statements to existing policy") + print(f" 2. REPLACE - Replace entire policy with new statements only") + print(f" 3. SKIP - Do not modify bucket policy") + choice = input("\n Choose (1/2/3): ").strip() + + if choice == '1': + new_sids = {s['Sid'] for s in new_statements if 'Sid' in s} + merged = [s for s in existing_statements if s.get('Sid') not in new_sids] + merged.extend(new_statements) + existing_policy['Statement'] = merged + final_policy = existing_policy + elif choice == '2': + final_policy = render_policy("destination_bucket.json", + role_arn=role_arn, + dest_bucket=cfg.dest_bucket) + final_policy['Statement'].append( + render_policy("datasync_bucket_stmt.json", + datasync_role_arn=datasync_role_arn, + dest_bucket=cfg.dest_bucket)) + else: + print(f" Skipped bucket policy update.") + return False + else: + print(f"\n No existing bucket policy found.") + final_policy = render_policy("destination_bucket.json", + role_arn=role_arn, + dest_bucket=cfg.dest_bucket) + final_policy['Statement'].append( + render_policy("datasync_bucket_stmt.json", + datasync_role_arn=datasync_role_arn, + dest_bucket=cfg.dest_bucket)) + + ctx.dest_s3.put_bucket_policy( + Bucket=cfg.dest_bucket, + Policy=json.dumps(final_policy) + ) + print(f" Applied bucket policy to '{cfg.dest_bucket}'") + return True + except ClientError as e: + print(f" ERROR: {e}") + return False From 832630da710cfa8100ce8967bd92bf54559765df Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:37:08 +0530 Subject: [PATCH 06/21] add utils to start replication --- hq_s3_migration/replication.py | 103 +++++++++++++++++++++++++++++++++ 1 file changed, 103 insertions(+) create mode 100644 hq_s3_migration/replication.py diff --git a/hq_s3_migration/replication.py b/hq_s3_migration/replication.py new file mode 100644 index 0000000000..406f7f629f --- /dev/null +++ b/hq_s3_migration/replication.py @@ -0,0 +1,103 @@ +from botocore.exceptions import ClientError + +from .orchestrator import S3MigrationContext + + +def enable_live_replication(ctx: S3MigrationContext) -> bool: + """Enable S3 live replication from source to destination.""" + cfg = ctx.config + + print("\n" + "=" * 60) + print("PHASE 2: Enabling S3 Live Replication") + print("=" * 60) + + role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.replication_role_name}" + + rule = { + 'ID': 'CrossAccountReplication', + 'Status': 'Enabled', + 'Priority': 1, + 'Filter': {}, + 'Destination': { + 'Bucket': f'arn:aws:s3:::{cfg.dest_bucket}', + 'Account': cfg.dest_account_id, + 'AccessControlTranslation': { + 'Owner': 'Destination' + } + }, + 'DeleteMarkerReplication': { + 'Status': 'Enabled' if cfg.enable_delete_replication else 'Disabled' + } + } + + if cfg.enable_rtc: + rule['Destination']['ReplicationTime'] = { + 'Status': 'Enabled', + 'Time': {'Minutes': 15} + } + rule['Destination']['Metrics'] = { + 'Status': 'Enabled', + 'EventThreshold': {'Minutes': 15} + } + + replication_config = { + 'Role': role_arn, + 'Rules': [rule] + } + + print(f"\nConfiguring replication rule...") + print(f" Source: {cfg.source_bucket}") + print(f" Destination: {cfg.dest_bucket}") + print(f" Delete replication: {'Enabled' if cfg.enable_delete_replication else 'Disabled'}") + print(f" RTC (15-min SLA): {'Enabled' if cfg.enable_rtc else 'Disabled'}") + + try: + ctx.source_s3.put_bucket_replication( + Bucket=cfg.source_bucket, + ReplicationConfiguration=replication_config + ) + print(f"\n SUCCESS: Live replication enabled") + print(f" All NEW objects will now replicate automatically") + return True + except ClientError as e: + print(f"\n ERROR: {e}") + return False + + +def get_replication_status(ctx: S3MigrationContext) -> dict: + """Get current replication configuration and status.""" + print("\nChecking replication status...") + + result = { + 'configured': False, + 'rules': [], + 'metrics': None + } + + try: + config = ctx.source_s3.get_bucket_replication(Bucket=ctx.config.source_bucket) + result['configured'] = True + result['rules'] = config['ReplicationConfiguration']['Rules'] + print(f" Replication is configured") + for rule in result['rules']: + print(f" Rule '{rule['ID']}': {rule['Status']}") + except ClientError as e: + if 'ReplicationConfigurationNotFoundError' in str(e): + print(f" Replication is NOT configured") + else: + print(f" ERROR: {e}") + + return result + + +def disable_replication(ctx: S3MigrationContext) -> bool: + """Disable S3 replication rule.""" + print("\nDisabling replication...") + + try: + ctx.source_s3.delete_bucket_replication(Bucket=ctx.config.source_bucket) + print(f" Replication disabled") + return True + except ClientError as e: + print(f" ERROR: {e}") + return False From 06bc72927455ad41e2fd7c9b77ba79182f639bea Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:52:37 +0530 Subject: [PATCH 07/21] add util for datasync --- hq_s3_migration/datasync.py | 188 ++++++++++++++++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 hq_s3_migration/datasync.py diff --git a/hq_s3_migration/datasync.py b/hq_s3_migration/datasync.py new file mode 100644 index 0000000000..c1c373cf31 --- /dev/null +++ b/hq_s3_migration/datasync.py @@ -0,0 +1,188 @@ +import time +from datetime import datetime, timezone +from typing import Optional + +from botocore.exceptions import ClientError + +from .orchestrator import S3MigrationContext + + +def create_datasync_source_location(ctx: S3MigrationContext) -> Optional[str]: + """Create DataSync source location (S3 bucket).""" + cfg = ctx.config + print(f"\nCreating DataSync source location...") + + datasync_role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.datasync_role_name}" + + try: + response = ctx.source_datasync.create_location_s3( + S3BucketArn=f"arn:aws:s3:::{cfg.source_bucket}", + S3Config={ + 'BucketAccessRoleArn': datasync_role_arn + } + ) + location_arn = response['LocationArn'] + print(f" Created source location: {location_arn}") + return location_arn + except ClientError as e: + print(f" ERROR: {e}") + return None + + +def create_datasync_destination_location(ctx: S3MigrationContext) -> Optional[str]: + """Create DataSync destination location (S3 bucket in different account).""" + cfg = ctx.config + print(f"\nCreating DataSync destination location...") + + datasync_role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.datasync_role_name}" + + try: + response = ctx.source_datasync.create_location_s3( + S3BucketArn=f"arn:aws:s3:::{cfg.dest_bucket}", + S3Config={ + 'BucketAccessRoleArn': datasync_role_arn + } + ) + location_arn = response['LocationArn'] + print(f" Created destination location: {location_arn}") + return location_arn + except ClientError as e: + print(f" ERROR: {e}") + return None + + +def create_datasync_task(ctx: S3MigrationContext, + source_location_arn: str, + dest_location_arn: str) -> Optional[str]: + """Create DataSync task for S3 to S3 transfer.""" + cfg = ctx.config + print(f"\nCreating DataSync task...") + + try: + response = ctx.source_datasync.create_task( + SourceLocationArn=source_location_arn, + DestinationLocationArn=dest_location_arn, + Name=f"s3-migration-{cfg.source_bucket}-to-{cfg.dest_bucket}", + TaskMode='ENHANCED', + Options={ + 'VerifyMode': 'ONLY_FILES_TRANSFERRED', + 'OverwriteMode': 'ALWAYS', + 'PreserveDeletedFiles': 'REMOVE', + 'PreserveDevices': 'NONE', + 'PosixPermissions': 'NONE', + 'Uid': 'NONE', + 'Gid': 'NONE', + 'TaskQueueing': 'ENABLED', + 'TransferMode': 'CHANGED', + 'ObjectTags': 'PRESERVE', + 'LogLevel': 'BASIC', + }, + ) + task_arn = response['TaskArn'] + print(f" Created task: {task_arn}") + return task_arn + except ClientError as e: + print(f" ERROR: {e}") + return None + + +def start_datasync_task(ctx: S3MigrationContext, task_arn: str) -> Optional[str]: + """Start a DataSync task execution.""" + print(f"\nStarting DataSync task execution...") + + try: + response = ctx.source_datasync.start_task_execution( + TaskArn=task_arn + ) + execution_arn = response['TaskExecutionArn'] + print(f" Started execution: {execution_arn}") + return execution_arn + except ClientError as e: + print(f" ERROR: {e}") + return None + + +def get_datasync_task_status(ctx: S3MigrationContext, execution_arn: str) -> dict: + """Get status of DataSync task execution.""" + try: + response = ctx.source_datasync.describe_task_execution( + TaskExecutionArn=execution_arn + ) + + status = { + 'status': response['Status'], + 'bytes_transferred': response.get('BytesTransferred', 0), + 'bytes_written': response.get('BytesWritten', 0), + 'files_transferred': response.get('FilesTransferred', 0), + 'estimated_bytes': response.get('EstimatedBytesToTransfer', 0), + 'estimated_files': response.get('EstimatedFilesToTransfer', 0), + } + + if status['estimated_bytes'] > 0: + status['progress_pct'] = (status['bytes_transferred'] / status['estimated_bytes']) * 100 + else: + status['progress_pct'] = 0 + + return status + except ClientError as e: + return {'status': 'ERROR', 'error': str(e)} + + +def monitor_datasync_task(ctx: S3MigrationContext, execution_arn: str, interval: int = 60): + """Monitor DataSync task until completion.""" + print(f"\nMonitoring DataSync task execution...") + print(f" Execution ARN: {execution_arn}") + print(f" Checking every {interval} seconds (Ctrl+C to stop monitoring)...") + print() + + status = None + try: + while True: + status = get_datasync_task_status(ctx, execution_arn) + + timestamp = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') + print(f"[{timestamp}] Status: {status['status']}") + + if status['status'] == 'ERROR' and 'error' in status: + print(f" API error: {status['error']}, will retry...") + time.sleep(interval) + continue + + if status['status'] not in ['QUEUED', 'LAUNCHING', 'PREPARING', 'TRANSFERRING', 'VERIFYING']: + print(f"\nTask completed with status: {status['status']}") + if status['status'] == 'SUCCESS': + print(f" Files transferred: {status['files_transferred']:,}") + print(f" Bytes transferred: {status['bytes_transferred']:,}") + break + + if status['estimated_bytes'] > 0: + print(f" Progress: {status['progress_pct']:.1f}%") + print(f" Transferred: {status['bytes_transferred']:,} / {status['estimated_bytes']:,} bytes") + print(f" Files: {status['files_transferred']:,} / {status['estimated_files']:,}") + + time.sleep(interval) + except KeyboardInterrupt: + print(f"\n\nMonitoring stopped by user. Task is still running.") + print(f" Resume monitoring with the same command and profile arguments.") + + return status + + +def list_datasync_tasks(ctx: S3MigrationContext) -> list: + """List all DataSync tasks.""" + print("\nListing DataSync tasks...") + + try: + tasks = [] + paginator = ctx.source_datasync.get_paginator('list_tasks') + for page in paginator.paginate(): + tasks.extend(page.get('Tasks', [])) + + for task in tasks: + print(f" {task['Name']}: {task['Status']}") + print(f" ARN: {task['TaskArn']}") + + return tasks + except ClientError as e: + print(f" ERROR: {e}") + return [] From 3a4b6b2143d0163e6a20243859029ac2b9fef410 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:53:13 +0530 Subject: [PATCH 08/21] add validation checks after sync --- hq_s3_migration/validation.py | 258 ++++++++++++++++++++++++++++++++++ 1 file changed, 258 insertions(+) create mode 100644 hq_s3_migration/validation.py diff --git a/hq_s3_migration/validation.py b/hq_s3_migration/validation.py new file mode 100644 index 0000000000..5463b3c120 --- /dev/null +++ b/hq_s3_migration/validation.py @@ -0,0 +1,258 @@ +import random +import string +from datetime import datetime, timedelta, timezone + +from botocore.exceptions import ClientError + +from .orchestrator import S3MigrationContext +from .replication import get_replication_status + + +def format_size(size_bytes: int) -> str: + """Format bytes to human readable string.""" + for unit in ['B', 'KB', 'MB', 'GB', 'TB', 'PB']: + if size_bytes < 1024.0: + return f"{size_bytes:.2f} {unit}" + size_bytes /= 1024.0 + return f"{size_bytes:.2f} EB" + + +def get_bucket_stats(ctx: S3MigrationContext, s3_client, bucket_name: str) -> dict: + """Get object count and total size using CloudWatch metrics (scales to 1B+ objects).""" + print(f"\nGetting stats for bucket '{bucket_name}'...") + + session = ctx.source_session if s3_client is ctx.source_s3 else ctx.dest_session + cloudwatch = session.client('cloudwatch') + + now = datetime.now(timezone.utc) + metrics_result = {} + + try: + for metric_name, stat_key in [('NumberOfObjects', 'object_count'), ('BucketSizeBytes', 'total_size')]: + response = cloudwatch.get_metric_statistics( + Namespace='AWS/S3', + MetricName=metric_name, + Dimensions=[ + {'Name': 'BucketName', 'Value': bucket_name}, + {'Name': 'StorageType', 'Value': 'AllStorageTypes' if metric_name == 'NumberOfObjects' else 'StandardStorage'} + ], + StartTime=now - timedelta(days=3), + EndTime=now, + Period=86400, + Statistics=['Average'] + ) + if response['Datapoints']: + latest = max(response['Datapoints'], key=lambda x: x['Timestamp']) + metrics_result[stat_key] = int(latest['Average']) + + if 'object_count' in metrics_result: + total_objects = metrics_result['object_count'] + total_size = metrics_result.get('total_size', 0) + print(f" Total objects: {total_objects:,} (from CloudWatch metrics)") + print(f" Total size: {format_size(total_size)}") + print(f" Note: CloudWatch S3 metrics may be up to 24h delayed") + return {'object_count': total_objects, 'total_size': total_size} + else: + print(f" WARNING: No CloudWatch metrics available for this bucket") + print(f" Ensure S3 request metrics are enabled or wait for daily storage metrics") + return {'object_count': 0, 'total_size': 0, 'error': 'no_metrics'} + except ClientError as e: + print(f" ERROR: {e}") + return {'object_count': 0, 'total_size': 0, 'error': str(e)} + + +def verify_sample(ctx: S3MigrationContext, sample_size: int, mismatches: list) -> bool: + """Verify a sample of objects between source and destination. + + Uses random prefixes to sample from across the entire key space, + avoiding bias toward lexicographically early keys. + """ + cfg = ctx.config + sample_keys = [] + prefixes_tried = set() + prefix_chars = string.ascii_lowercase + string.digits + + try: + while len(sample_keys) < sample_size * 3 and len(prefixes_tried) < len(prefix_chars): + prefix = random.choice(prefix_chars) + if prefix in prefixes_tried: + continue + prefixes_tried.add(prefix) + + response = ctx.source_s3.list_objects_v2( + Bucket=cfg.source_bucket, + Prefix=prefix, + MaxKeys=sample_size + ) + if 'Contents' in response: + for obj in response['Contents']: + sample_keys.append(obj['Key']) + except ClientError: + return False + + if not sample_keys: + print(" No objects to sample") + return True + + sample = random.sample(sample_keys, min(sample_size, len(sample_keys))) + + verified = 0 + for key in sample: + try: + source_obj = ctx.source_s3.head_object(Bucket=cfg.source_bucket, Key=key) + dest_obj = ctx.dest_s3.head_object(Bucket=cfg.dest_bucket, Key=key) + + if source_obj['ETag'] == dest_obj['ETag']: + verified += 1 + elif source_obj['ContentLength'] == dest_obj['ContentLength']: + verified += 1 + else: + mismatches.append({ + 'key': key, + 'reason': 'Size mismatch', + 'source_size': source_obj['ContentLength'], + 'dest_size': dest_obj['ContentLength'], + 'source_etag': source_obj['ETag'], + 'dest_etag': dest_obj['ETag'] + }) + except ClientError as e: + mismatches.append({ + 'key': key, + 'reason': str(e) + }) + + print(f" Verified: {verified}/{sample_size}") + if mismatches: + print(f" Mismatches: {len(mismatches)}") + for m in mismatches[:5]: + print(f" - {m['key']}: {m['reason']}") + + return len(mismatches) == 0 + + +def validate_migration(ctx: S3MigrationContext, sample_size: int = 100) -> dict: + """Validate migration by comparing buckets.""" + cfg = ctx.config + + print("\n" + "=" * 60) + print("PHASE 4: Validating Migration") + print("=" * 60) + + result = { + 'source_stats': None, + 'dest_stats': None, + 'sample_verified': False, + 'mismatches': [] + } + + print("\nComparing bucket statistics...") + result['source_stats'] = get_bucket_stats(ctx, ctx.source_s3, cfg.source_bucket) + result['dest_stats'] = get_bucket_stats(ctx, ctx.dest_s3, cfg.dest_bucket) + + source_count = result['source_stats']['object_count'] + dest_count = result['dest_stats']['object_count'] + + print(f"\nComparison:") + print(f" Source objects: {source_count:,}") + print(f" Destination objects: {dest_count:,}") + print(f" Difference: {abs(source_count - dest_count):,}") + + if sample_size > 0: + print(f"\nVerifying {sample_size} random objects...") + result['sample_verified'] = verify_sample(ctx, sample_size, result['mismatches']) + + return result + + +def check_replication_lag(ctx: S3MigrationContext) -> dict: + """Check replication lag using CloudWatch metrics.""" + cfg = ctx.config + print("\nChecking replication lag...") + + cloudwatch = ctx.source_session.client('cloudwatch') + + try: + response = cloudwatch.get_metric_statistics( + Namespace='AWS/S3', + MetricName='ReplicationLatency', + Dimensions=[ + {'Name': 'SourceBucket', 'Value': cfg.source_bucket}, + {'Name': 'DestinationBucket', 'Value': cfg.dest_bucket}, + {'Name': 'RuleId', 'Value': 'CrossAccountReplication'} + ], + StartTime=datetime.now(timezone.utc) - timedelta(hours=1), + EndTime=datetime.now(timezone.utc), + Period=300, + Statistics=['Average', 'Maximum'] + ) + + if response['Datapoints']: + latest = max(response['Datapoints'], key=lambda x: x['Timestamp']) + print(f" Average latency: {latest.get('Average', 'N/A')} seconds") + print(f" Maximum latency: {latest.get('Maximum', 'N/A')} seconds") + return { + 'avg_latency': latest.get('Average'), + 'max_latency': latest.get('Maximum'), + 'timestamp': latest['Timestamp'] + } + else: + print(" No replication metrics available") + return {'error': 'No metrics available'} + except ClientError as e: + print(f" ERROR: {e}") + return {'error': str(e)} + + +def cutover_checklist(ctx: S3MigrationContext) -> dict: + """Run pre-cutover checklist.""" + cfg = ctx.config + + print("\n" + "=" * 60) + print("PHASE 5: Pre-Cutover Checklist") + print("=" * 60) + + checks = { + 'replication_active': False, + 'replication_lag_ok': False, + 'object_counts_match': False, + 'sample_verified': False + } + + print("\n1. Checking replication is active...") + rep_status = get_replication_status(ctx) + checks['replication_active'] = rep_status['configured'] + + print("\n2. Checking replication lag...") + lag = check_replication_lag(ctx) + if 'avg_latency' in lag: + checks['replication_lag_ok'] = lag['avg_latency'] < 300 + else: + checks['replication_lag_ok'] = True + + print("\n3. Comparing object counts...") + source_stats = get_bucket_stats(ctx, ctx.source_s3, cfg.source_bucket) + dest_stats = get_bucket_stats(ctx, ctx.dest_s3, cfg.dest_bucket) + + diff = abs(source_stats['object_count'] - dest_stats['object_count']) + diff_pct = (diff / max(source_stats['object_count'], 1)) * 100 + checks['object_counts_match'] = diff_pct < 1 + + print("\n4. Verifying sample objects...") + mismatches = [] + checks['sample_verified'] = verify_sample(ctx, 50, mismatches) + + print("\n" + "-" * 40) + print("Cutover Checklist Results:") + all_passed = True + for check, passed in checks.items(): + status = "PASS" if passed else "FAIL" + print(f" {check}: {status}") + if not passed: + all_passed = False + + if all_passed: + print("\nAll checks passed. Ready for cutover.") + else: + print("\nSome checks failed. Review before proceeding with cutover.") + + return checks From 04540e9e22980430dea9cdcb80438a83a2438896 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:53:29 +0530 Subject: [PATCH 09/21] wrap everything up in cli --- hq_s3_migration/cli.py | 331 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 331 insertions(+) create mode 100644 hq_s3_migration/cli.py diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py new file mode 100644 index 0000000000..f3b402d9c1 --- /dev/null +++ b/hq_s3_migration/cli.py @@ -0,0 +1,331 @@ +#!/usr/bin/env python3 +""" +S3 Cross-Account Migration Tool + +Implements hybrid approach for large-scale S3 migrations: +- DataSync for bulk transfer of existing data +- S3 Live Replication for ongoing sync + +Designed for 204 TB / 1B+ objects with zero-downtime cutover. +""" + +import argparse +import sys + +from botocore.exceptions import ClientError + +from .config import ACCOUNT_IDS, ACCOUNT_NAMES, MigrationConfig +from .datasync import (create_datasync_destination_location, + create_datasync_source_location, create_datasync_task, + list_datasync_tasks, monitor_datasync_task, + start_datasync_task) +from .iam import (apply_destination_bucket_policy, create_datasync_role, + create_replication_role, print_iam_policies) +from .orchestrator import S3MigrationContext +from .replication import enable_live_replication, get_replication_status +from .validation import cutover_checklist, validate_migration + + +def _check_prerequisites(ctx: S3MigrationContext) -> dict: + """Check all prerequisites for migration.""" + cfg = ctx.config + + print("\n" + "=" * 60) + print("PHASE 1: Checking Prerequisites") + print("=" * 60) + + results = { + 'source_bucket_exists': False, + 'dest_bucket_exists': False, + 'source_versioning': False, + 'dest_versioning': False, + 'source_account_verified': False, + 'dest_account_verified': False, + } + + print("\nVerifying AWS account access...") + try: + source_identity = ctx.source_sts.get_caller_identity() + results['source_account_verified'] = True + actual_source_account = source_identity['Account'] + print(f" Source account: {actual_source_account}") + if actual_source_account != cfg.source_account_id: + print(f" WARNING: Configured source account ID ({cfg.source_account_id}) " + f"doesn't match actual ({actual_source_account})") + except ClientError as e: + print(f" ERROR: Cannot access source account: {e}") + + try: + dest_identity = ctx.dest_sts.get_caller_identity() + results['dest_account_verified'] = True + actual_dest_account = dest_identity['Account'] + print(f" Destination account: {actual_dest_account}") + if actual_dest_account != cfg.dest_account_id: + print(f" WARNING: Configured dest account ID ({cfg.dest_account_id}) " + f"doesn't match actual ({actual_dest_account})") + except ClientError as e: + print(f" ERROR: Cannot access destination account: {e}") + + print(f"\nChecking source bucket '{cfg.source_bucket}'...") + try: + ctx.source_s3.head_bucket(Bucket=cfg.source_bucket) + results['source_bucket_exists'] = True + print(f" Bucket exists") + + versioning = ctx.source_s3.get_bucket_versioning(Bucket=cfg.source_bucket) + status = versioning.get('Status', 'Disabled') + results['source_versioning'] = status == 'Enabled' + print(f" Versioning: {status}") + except ClientError as e: + print(f" ERROR: {e}") + + print(f"\nChecking destination bucket '{cfg.dest_bucket}'...") + try: + ctx.dest_s3.head_bucket(Bucket=cfg.dest_bucket) + results['dest_bucket_exists'] = True + print(f" Bucket exists") + + versioning = ctx.dest_s3.get_bucket_versioning(Bucket=cfg.dest_bucket) + status = versioning.get('Status', 'Disabled') + results['dest_versioning'] = status == 'Enabled' + print(f" Versioning: {status}") + except ClientError as e: + if e.response['Error']['Code'] == '404': + print(f" Bucket does not exist (will be created)") + else: + print(f" ERROR: {e}") + + print("\n" + "-" * 40) + print("Prerequisites Summary:") + for key, value in results.items(): + status = "OK" if value else "MISSING" + print(f" {key}: {status}") + + return results + + +def _create_destination_bucket(ctx: S3MigrationContext) -> bool: + """Create destination bucket with versioning enabled.""" + cfg = ctx.config + print(f"\nCreating destination bucket '{cfg.dest_bucket}'...") + + try: + if cfg.region == 'us-east-1': + ctx.dest_s3.create_bucket(Bucket=cfg.dest_bucket) + else: + print(f" This tool is designed to create buckets in the Staging and Production accounts.") + return False + print(f" Created bucket") + + ctx.dest_s3.put_bucket_versioning( + Bucket=cfg.dest_bucket, + VersioningConfiguration={'Status': 'Enabled'} + ) + print(f" Enabled versioning") + + return True + except ClientError as e: + if e.response['Error']['Code'] == 'BucketAlreadyOwnedByYou': + print(f" Bucket already exists") + ctx.dest_s3.put_bucket_versioning( + Bucket=cfg.dest_bucket, + VersioningConfiguration={'Status': 'Enabled'} + ) + print(f" Enabled versioning") + return True + print(f" ERROR: {e}") + return False + + +def _enable_source_versioning(ctx: S3MigrationContext) -> bool: + """Enable versioning on source bucket if not already enabled.""" + cfg = ctx.config + print(f"\nEnabling versioning on source bucket...") + + try: + ctx.source_s3.put_bucket_versioning( + Bucket=cfg.source_bucket, + VersioningConfiguration={'Status': 'Enabled'} + ) + print(f" Versioning enabled on '{cfg.source_bucket}'") + return True + except ClientError as e: + print(f" ERROR: {e}") + return False + + +def _resolve_account_id(value): + """Resolve account name alias or validate 12-digit account ID.""" + if value in ACCOUNT_IDS: + return ACCOUNT_IDS[value] + if value.isdigit() and len(value) == 12: + return value + aliases = ', '.join(ACCOUNT_IDS.keys()) + raise argparse.ArgumentTypeError( + f"Must be a 12-digit account ID or one of: {aliases}. Got: '{value}'" + ) + + +def main(): + parser = argparse.ArgumentParser( + description='S3 Cross-Account Migration Tool (Hybrid: DataSync + Live Replication)', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Commands: + prepare Check prerequisites and create destination bucket + setup-iam Print IAM policies for manual setup or create them + enable-replication Enable S3 live replication + create-datasync Create DataSync task + start-datasync Start DataSync task execution + monitor-datasync Monitor DataSync task execution + validate Validate migration status + cutover-check Run pre-cutover checklist + status Show current migration status + +Examples: + # Check prerequisites + python -m hq_s3_migration prepare --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + + # Print IAM policies + python -m hq_s3_migration setup-iam --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + + # Enable live replication + python -m hq_s3_migration enable-replication --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + + # Create and start DataSync task + python -m hq_s3_migration create-datasync --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + """ + ) + + parser.add_argument('command', choices=[ + 'prepare', 'setup-iam', 'enable-replication', 'create-datasync', + 'start-datasync', 'monitor-datasync', 'validate', 'cutover-check', + 'status' + ], help='Command to execute') + + parser.add_argument('--source-profile', default='StagingAdminAccess', + help='AWS profile for source account') + parser.add_argument('--dest-profile', default='BackupAdminAccess', + help='AWS profile for destination account') + parser.add_argument('--source-bucket', default='ap-source-for-replication', + help='Source bucket name') + parser.add_argument('--dest-bucket', default='ap-destination-for-replication', + help='Destination bucket name') + parser.add_argument('--source-account', required=True, type=_resolve_account_id, + help=f'Source AWS account ID or alias ({", ".join(ACCOUNT_IDS.keys())})') + parser.add_argument('--dest-account', required=True, type=_resolve_account_id, + help=f'Destination AWS account ID or alias ({", ".join(ACCOUNT_IDS.keys())})') + parser.add_argument('--region', default='us-east-1', + help='AWS region') + parser.add_argument('--disable-rtc', action='store_true', + help='Disable S3 Replication Time Control (enabled by default)') + parser.add_argument('--create-iam', action='store_true', + help='Create IAM roles (for setup-iam command)') + parser.add_argument('--task-arn', + help='DataSync task ARN (for start-datasync)') + parser.add_argument('--execution-arn', + help='DataSync execution ARN (for monitor-datasync)') + + args = parser.parse_args() + + # Append environment name to role names for uniqueness across accounts + env_name = ACCOUNT_NAMES.get(args.source_account, args.source_account) + replication_role = f"s3-cross-account-replication-role-{env_name}" + datasync_role = f"datasync-s3-access-role-{env_name}" + + config = MigrationConfig( + source_profile=args.source_profile, + dest_profile=args.dest_profile, + source_bucket=args.source_bucket, + dest_bucket=args.dest_bucket, + source_account_id=args.source_account, + dest_account_id=args.dest_account, + region=args.region, + enable_rtc=not args.disable_rtc, + replication_role_name=replication_role, + datasync_role_name=datasync_role, + ) + + ctx = S3MigrationContext(config) + + if args.command == 'prepare': + results = _check_prerequisites(ctx) + if not results['source_bucket_exists']: + print("\nERROR: Source bucket does not exist. Aborting.") + sys.exit(1) + if not results['source_account_verified'] or not results['dest_account_verified']: + print("\nERROR: Cannot verify AWS account access. Aborting.") + sys.exit(1) + if not results['dest_bucket_exists']: + _create_destination_bucket(ctx) + else: + if not results['dest_versioning']: + ctx.dest_s3.put_bucket_versioning( + Bucket=config.dest_bucket, + VersioningConfiguration={'Status': 'Enabled'} + ) + print(f"\n Enabled versioning on existing destination bucket") + _enable_source_versioning(ctx) + + elif args.command == 'setup-iam': + print_iam_policies(ctx) + if args.create_iam: + print("\n" + "=" * 60) + print("Creating IAM Resources") + print("=" * 60) + create_replication_role(ctx) + create_datasync_role(ctx) + apply_destination_bucket_policy(ctx) + + elif args.command == 'enable-replication': + enable_live_replication(ctx) + + elif args.command == 'create-datasync': + source_loc = create_datasync_source_location(ctx) + if source_loc: + dest_loc = create_datasync_destination_location(ctx) + if dest_loc: + task_arn = create_datasync_task(ctx, source_loc, dest_loc) + if task_arn: + print(f"\nDataSync task created successfully!") + print(f"Task ARN: {task_arn}") + print(f"\nTo start the task, run:") + print(f" python -m hq_s3_migration.cli start-datasync --task-arn '{task_arn}' \\") + print(f" --source-profile {args.source_profile} --dest-profile {args.dest_profile} \\") + print(f" --source-account {args.source_account} --dest-account {args.dest_account}") + + elif args.command == 'start-datasync': + if not args.task_arn: + print("ERROR: --task-arn is required for start-datasync") + sys.exit(1) + execution_arn = start_datasync_task(ctx, args.task_arn) + if execution_arn: + print(f"\nTo monitor the task, run:") + print(f" python -m hq_s3_migration.cli monitor-datasync --execution-arn '{execution_arn}' \\") + print(f" --source-profile {args.source_profile} --dest-profile {args.dest_profile} \\") + print(f" --source-account {args.source_account} --dest-account {args.dest_account}") + + elif args.command == 'monitor-datasync': + if not args.execution_arn: + print("ERROR: --execution-arn is required for monitor-datasync") + sys.exit(1) + monitor_datasync_task(ctx, args.execution_arn) + + elif args.command == 'validate': + validate_migration(ctx) + + elif args.command == 'cutover-check': + cutover_checklist(ctx) + + elif args.command == 'status': + _check_prerequisites(ctx) + get_replication_status(ctx) + list_datasync_tasks(ctx) + + +if __name__ == '__main__': + main() From 7e239a1a524701321975f2f136641553a83ed5da Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 4 Mar 2026 22:53:50 +0530 Subject: [PATCH 10/21] calling module calls cli --- hq_s3_migration/__main__.py | 3 +++ 1 file changed, 3 insertions(+) create mode 100644 hq_s3_migration/__main__.py diff --git a/hq_s3_migration/__main__.py b/hq_s3_migration/__main__.py new file mode 100644 index 0000000000..4e28416e10 --- /dev/null +++ b/hq_s3_migration/__main__.py @@ -0,0 +1,3 @@ +from .cli import main + +main() From 4fa5a6f1c5b1d13749fade905a051451c455d736 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Fri, 6 Mar 2026 16:47:02 +0530 Subject: [PATCH 11/21] instead of calling head, rely on different apis which are not added to block list --- hq_s3_migration/cli.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index f3b402d9c1..3644b8b03c 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -68,11 +68,10 @@ def _check_prerequisites(ctx: S3MigrationContext) -> dict: print(f"\nChecking source bucket '{cfg.source_bucket}'...") try: - ctx.source_s3.head_bucket(Bucket=cfg.source_bucket) + versioning = ctx.source_s3.get_bucket_versioning(Bucket=cfg.source_bucket) results['source_bucket_exists'] = True print(f" Bucket exists") - versioning = ctx.source_s3.get_bucket_versioning(Bucket=cfg.source_bucket) status = versioning.get('Status', 'Disabled') results['source_versioning'] = status == 'Enabled' print(f" Versioning: {status}") @@ -81,16 +80,15 @@ def _check_prerequisites(ctx: S3MigrationContext) -> dict: print(f"\nChecking destination bucket '{cfg.dest_bucket}'...") try: - ctx.dest_s3.head_bucket(Bucket=cfg.dest_bucket) + versioning = ctx.dest_s3.get_bucket_versioning(Bucket=cfg.dest_bucket) results['dest_bucket_exists'] = True print(f" Bucket exists") - versioning = ctx.dest_s3.get_bucket_versioning(Bucket=cfg.dest_bucket) status = versioning.get('Status', 'Disabled') results['dest_versioning'] = status == 'Enabled' print(f" Versioning: {status}") except ClientError as e: - if e.response['Error']['Code'] == '404': + if e.response['Error']['Code'] in ('404', 'NoSuchBucket'): print(f" Bucket does not exist (will be created)") else: print(f" ERROR: {e}") From cd0d812773794d36fc61ab7cfcc3a3c25aa9651c Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Fri, 6 Mar 2026 16:49:32 +0530 Subject: [PATCH 12/21] during datasync and live replication set the buckets to be of intelligent tiering --- hq_s3_migration/datasync.py | 1 + hq_s3_migration/replication.py | 1 + 2 files changed, 2 insertions(+) diff --git a/hq_s3_migration/datasync.py b/hq_s3_migration/datasync.py index c1c373cf31..5266cb064d 100644 --- a/hq_s3_migration/datasync.py +++ b/hq_s3_migration/datasync.py @@ -39,6 +39,7 @@ def create_datasync_destination_location(ctx: S3MigrationContext) -> Optional[st try: response = ctx.source_datasync.create_location_s3( S3BucketArn=f"arn:aws:s3:::{cfg.dest_bucket}", + S3StorageClass='INTELLIGENT_TIERING', S3Config={ 'BucketAccessRoleArn': datasync_role_arn } diff --git a/hq_s3_migration/replication.py b/hq_s3_migration/replication.py index 406f7f629f..94879c21ef 100644 --- a/hq_s3_migration/replication.py +++ b/hq_s3_migration/replication.py @@ -21,6 +21,7 @@ def enable_live_replication(ctx: S3MigrationContext) -> bool: 'Destination': { 'Bucket': f'arn:aws:s3:::{cfg.dest_bucket}', 'Account': cfg.dest_account_id, + 'StorageClass': 'INTELLIGENT_TIERING', 'AccessControlTranslation': { 'Owner': 'Destination' } From 681233ce1383eb369dbb3f489def704081c434a8 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Fri, 6 Mar 2026 17:23:09 +0530 Subject: [PATCH 13/21] add Environment tag in the destination s3 bucket --- hq_s3_migration/cli.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index 3644b8b03c..d65afb9211 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -121,6 +121,17 @@ def _create_destination_bucket(ctx: S3MigrationContext) -> bool: ) print(f" Enabled versioning") + env_name = ACCOUNT_NAMES.get(cfg.dest_account_id, 'unknown') + ctx.dest_s3.put_bucket_tagging( + Bucket=cfg.dest_bucket, + Tagging={ + 'TagSet': [ + {'Key': 'Environment', 'Value': env_name}, + ] + } + ) + print(f" Tagged with Environment={env_name}") + return True except ClientError as e: if e.response['Error']['Code'] == 'BucketAlreadyOwnedByYou': From ecfc8a072b8b68b35ab02bea4aff3ca0bdeea4f6 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Wed, 17 Jun 2026 17:25:28 +0530 Subject: [PATCH 14/21] have source and destination roles both in the rolename --- hq_s3_migration/cli.py | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index d65afb9211..d03c61d116 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -15,12 +15,20 @@ from botocore.exceptions import ClientError from .config import ACCOUNT_IDS, ACCOUNT_NAMES, MigrationConfig -from .datasync import (create_datasync_destination_location, - create_datasync_source_location, create_datasync_task, - list_datasync_tasks, monitor_datasync_task, - start_datasync_task) -from .iam import (apply_destination_bucket_policy, create_datasync_role, - create_replication_role, print_iam_policies) +from .datasync import ( + create_datasync_destination_location, + create_datasync_source_location, + create_datasync_task, + list_datasync_tasks, + monitor_datasync_task, + start_datasync_task, +) +from .iam import ( + apply_destination_bucket_policy, + create_datasync_role, + create_replication_role, + print_iam_policies, +) from .orchestrator import S3MigrationContext from .replication import enable_live_replication, get_replication_status from .validation import cutover_checklist, validate_migration @@ -242,9 +250,12 @@ def main(): args = parser.parse_args() # Append environment name to role names for uniqueness across accounts - env_name = ACCOUNT_NAMES.get(args.source_account, args.source_account) - replication_role = f"s3-cross-account-replication-role-{env_name}" - datasync_role = f"datasync-s3-access-role-{env_name}" + source_env_name = ACCOUNT_NAMES.get(args.source_account, args.source_account) + dest_env_name = ACCOUNT_NAMES.get(args.dest_account, args.dest_account) + print(f"Source environment name: {source_env_name}") + print(f"Destination environment name: {dest_env_name}") + replication_role = f"s3-cross-account-replication-role-for-{source_env_name}-to-{dest_env_name}" + datasync_role = f"datasync-s3-access-role-for-{source_env_name}-to-{dest_env_name}" config = MigrationConfig( source_profile=args.source_profile, From 7e5d73ad4437bce0b209851d62e94481e6c9f90f Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 08:36:22 +0530 Subject: [PATCH 15/21] add utils and role json for configuring error events to sqs for live replication We will be creating 2 queues, one specifically for passing on errored events and another for threshold events, then the bucket is configured to send these events to SQS where they reside for 14 days. For now we have kept the retention for 14 days which is maximum for SQS. And there is a helper that polls sqs to get all the events and write to files in batches of 10000 items at a time --- hq_s3_migration/monitoring.py | 235 ++++++++++++++++++ .../policies/sqs_queue_access.json | 22 ++ 2 files changed, 257 insertions(+) create mode 100644 hq_s3_migration/monitoring.py create mode 100644 hq_s3_migration/policies/sqs_queue_access.json diff --git a/hq_s3_migration/monitoring.py b/hq_s3_migration/monitoring.py new file mode 100644 index 0000000000..4a3d849888 --- /dev/null +++ b/hq_s3_migration/monitoring.py @@ -0,0 +1,235 @@ +import json + +from botocore.exceptions import ClientError + +from .orchestrator import S3MigrationContext +from .policies import render_policy + +RETENTION_SECONDS = "1209600" # 14 days, SQS max + +FAILURE_EVENTS = ["s3:Replication:OperationFailedReplication"] +THRESHOLD_EVENTS = [ + "s3:Replication:OperationMissedThreshold", + "s3:Replication:OperationReplicatedAfterThreshold", + "s3:Replication:OperationNotTracked", +] + + +def _ensure_queue(ctx, queue_name): + """Create (or look up) a queue; return its URL.""" + try: + resp = ctx.source_sqs.create_queue(QueueName=queue_name) + return resp["QueueUrl"] + except ClientError as e: + if e.response["Error"]["Code"] in ("QueueNameExists", + "QueueAlreadyExists", + "AWS.SimpleQueueService.QueueNameExists"): + return ctx.source_sqs.get_queue_url(QueueName=queue_name)["QueueUrl"] + raise + + +def _queue_arn(ctx, queue_url): + resp = ctx.source_sqs.get_queue_attributes( + QueueUrl=queue_url, AttributeNames=["QueueArn"] + ) + return resp["Attributes"]["QueueArn"] + + +def create_monitoring_queues(ctx: S3MigrationContext) -> dict: + """Create the failure + threshold SQS queues with retention and access policy.""" + cfg = ctx.config + print("\n" + "=" * 60) + print("Creating Replication Monitoring Queues") + print("=" * 60) + + result = {} + for key, name in (("failures", cfg.failures_queue_name), + ("threshold", cfg.threshold_queue_name)): + print(f"\nQueue '{name}'...") + try: + url = _ensure_queue(ctx, name) + arn = _queue_arn(ctx, url) + + ctx.source_sqs.set_queue_attributes( + QueueUrl=url, + Attributes={"MessageRetentionPeriod": RETENTION_SECONDS}, + ) + policy = render_policy( + "sqs_queue_access.json", + queue_arn=arn, + source_bucket=cfg.source_bucket, + source_account_id=cfg.source_account_id, + ) + ctx.source_sqs.set_queue_attributes( + QueueUrl=url, + Attributes={"Policy": json.dumps(policy)}, + ) + print(f" URL: {url}") + print(f" ARN: {arn}") + print(f" Retention: 14 days; access policy applied") + result[key] = {"url": url, "arn": arn} + except ClientError as e: + print(f" ERROR: {e}") + result[key] = None + + return result + + +def configure_bucket_notifications(ctx: S3MigrationContext, + failure_arn: str, + threshold_arn: str) -> bool: + """Merge our replication-event queue notifications into the source bucket config.""" + cfg = ctx.config + print(f"\nConfiguring bucket notifications on '{cfg.source_bucket}'...") + + try: + existing = ctx.source_s3.get_bucket_notification_configuration( + Bucket=cfg.source_bucket + ) + except ClientError as e: + print(f" ERROR reading existing notifications: {e}") + return False + + notif = {k: v for k, v in existing.items() if k != "ResponseMetadata"} + + ours = { + "replication-failures": { + "Id": "replication-failures", + "QueueArn": failure_arn, + "Events": FAILURE_EVENTS, + }, + "replication-threshold": { + "Id": "replication-threshold", + "QueueArn": threshold_arn, + "Events": THRESHOLD_EVENTS, + }, + } + + queue_cfgs = [ + c for c in notif.get("QueueConfigurations", []) + if c.get("Id") not in ours + ] + queue_cfgs.extend(ours.values()) + notif["QueueConfigurations"] = queue_cfgs + + try: + ctx.source_s3.put_bucket_notification_configuration( + Bucket=cfg.source_bucket, + NotificationConfiguration=notif, + ) + print(f" Applied notifications (failures + threshold queues)") + return True + except ClientError as e: + print(f" ERROR: {e}") + return False + + +def _queue_exists(ctx, queue_name) -> bool: + try: + ctx.source_sqs.get_queue_url(QueueName=queue_name) + return True + except ClientError as e: + if "NonExistentQueue" in e.response["Error"]["Code"]: + return False + raise + + +def get_monitoring_status(ctx: S3MigrationContext) -> dict: + """Print and return monitoring queue + notification status.""" + cfg = ctx.config + print("\nChecking replication monitoring status...") + + status = { + "failures_exists": _queue_exists(ctx, cfg.failures_queue_name), + "threshold_exists": _queue_exists(ctx, cfg.threshold_queue_name), + "notifications": [], + } + + try: + notif = ctx.source_s3.get_bucket_notification_configuration( + Bucket=cfg.source_bucket + ) + status["notifications"] = notif.get("QueueConfigurations", []) + except ClientError as e: + print(f" ERROR reading notifications: {e}") + + print(f" Failures queue: {'OK' if status['failures_exists'] else 'MISSING'}") + print(f" Threshold queue: {'OK' if status['threshold_exists'] else 'MISSING'}") + print(f" Bucket queue notifications: {len(status['notifications'])}") + return status + + +FILE_ROTATION = 10000 + + +def drain_queue_to_jsonl(ctx: S3MigrationContext, + queue_url: str, + output_prefix: str, + delete: bool = False, + max_messages=None) -> tuple: + """Drain queue messages into JSONL files, 10,000 entries per file. + + Each line is the raw message body (for S3 event notifications, a JSON object). + Returns (file_paths, total_count). With delete=False (peek) messages are not + removed and full coverage is NOT guaranteed (visibility timeout + duplicates). + """ + if not delete: + print(" WARNING: peek mode (no delete) does not guarantee full coverage " + "and may return duplicates.") + + file_paths = [] + total = 0 + file_index = 0 + current_file = None + lines_in_file = 0 + + def open_new_file(): + nonlocal file_index, current_file, lines_in_file + if current_file: + current_file.close() + file_index += 1 + path = f"{output_prefix}-{file_index:04d}.jsonl" + file_paths.append(path) + current_file = open(path, "w") + lines_in_file = 0 + + open_new_file() + + try: + while True: + if max_messages is not None and total >= max_messages: + break + resp = ctx.source_sqs.receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=10, + WaitTimeSeconds=10, + ) + messages = resp.get("Messages", []) + if not messages: + break + + handles = [] + for msg in messages: + if lines_in_file >= FILE_ROTATION: + open_new_file() + current_file.write(msg.get("Body", "") + "\n") + lines_in_file += 1 + total += 1 + handles.append(msg["ReceiptHandle"]) + if max_messages is not None and total >= max_messages: + break + + current_file.flush() + + if delete and handles: + ctx.source_sqs.delete_message_batch( + QueueUrl=queue_url, + Entries=[{"Id": str(i), "ReceiptHandle": h} + for i, h in enumerate(handles)], + ) + print(f" Drained {total} messages into {len(file_paths)} file(s)") + finally: + if current_file: + current_file.close() + + return file_paths, total diff --git a/hq_s3_migration/policies/sqs_queue_access.json b/hq_s3_migration/policies/sqs_queue_access.json new file mode 100644 index 0000000000..a637081a55 --- /dev/null +++ b/hq_s3_migration/policies/sqs_queue_access.json @@ -0,0 +1,22 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "AllowS3Notifications", + "Effect": "Allow", + "Principal": { + "Service": "s3.amazonaws.com" + }, + "Action": "sqs:SendMessage", + "Resource": "{queue_arn}", + "Condition": { + "ArnLike": { + "aws:SourceArn": "arn:aws:s3:::{source_bucket}" + }, + "StringEquals": { + "aws:SourceAccount": "{source_account_id}" + } + } + } + ] +} From b06d007a4057188aa870446d9fc1a32cf7753a1d Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 08:41:22 +0530 Subject: [PATCH 16/21] add a helper to get resource names which appends the env details at the end --- hq_s3_migration/config.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/hq_s3_migration/config.py b/hq_s3_migration/config.py index eef14481a0..1440921d8b 100644 --- a/hq_s3_migration/config.py +++ b/hq_s3_migration/config.py @@ -1,5 +1,19 @@ from dataclasses import dataclass + +def derive_resource_names(source_env: str, dest_env: str) -> dict: + """Derive all per-migration resource names from env-name aliases.""" + suffix = f"{source_env}-to-{dest_env}" + return { + "replication_role_name": f"s3-cross-account-replication-role-for-{suffix}", + "datasync_role_name": f"datasync-s3-access-role-for-{suffix}", + "report_bucket_name": f"s3-migration-{suffix}-datasync-run-reports", + "report_role_name": f"datasync-report-access-role-for-{suffix}", + "failures_queue_name": f"s3-replication-failures-{suffix}", + "threshold_queue_name": f"s3-replication-threshold-{suffix}", + } + + @dataclass class MigrationConfig: """Configuration for S3 cross-account migration.""" From 307660e8329b35335b80a4102066965f463d3d0e Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 08:46:29 +0530 Subject: [PATCH 17/21] hookup the added utils to the cli for live replication monitoring --- hq_s3_migration/cli.py | 57 ++++++++++++++++--- hq_s3_migration/orchestrator.py | 1 + .../policies/datasync_report_role.json | 17 ++++++ 3 files changed, 68 insertions(+), 7 deletions(-) create mode 100644 hq_s3_migration/policies/datasync_report_role.json diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index d03c61d116..c8d32433a1 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -14,7 +14,7 @@ from botocore.exceptions import ClientError -from .config import ACCOUNT_IDS, ACCOUNT_NAMES, MigrationConfig +from .config import ACCOUNT_IDS, ACCOUNT_NAMES, MigrationConfig, derive_resource_names from .datasync import ( create_datasync_destination_location, create_datasync_source_location, @@ -27,8 +27,16 @@ apply_destination_bucket_policy, create_datasync_role, create_replication_role, + create_report_bucket, + create_report_role, print_iam_policies, ) +from .monitoring import ( + configure_bucket_notifications, + create_monitoring_queues, + drain_queue_to_jsonl, + get_monitoring_status, +) from .orchestrator import S3MigrationContext from .replication import enable_live_replication, get_replication_status from .validation import cutover_checklist, validate_migration @@ -221,7 +229,7 @@ def main(): parser.add_argument('command', choices=[ 'prepare', 'setup-iam', 'enable-replication', 'create-datasync', 'start-datasync', 'monitor-datasync', 'validate', 'cutover-check', - 'status' + 'status', 'setup-monitoring', 'drain-queue' ], help='Command to execute') parser.add_argument('--source-profile', default='StagingAdminAccess', @@ -246,16 +254,24 @@ def main(): help='DataSync task ARN (for start-datasync)') parser.add_argument('--execution-arn', help='DataSync execution ARN (for monitor-datasync)') + parser.add_argument('--create-monitoring', action='store_true', + help='Create monitoring queues and bucket notifications (for setup-monitoring)') + parser.add_argument('--queue', default='failures', + help="Queue to drain: 'failures', 'threshold', or an explicit queue URL") + parser.add_argument('--output-prefix', default='replication-events', + help='Output filename prefix for drain-queue JSONL files') + parser.add_argument('--max-messages', type=int, default=None, + help='Optional cap on messages to drain') + parser.add_argument('--delete', action='store_true', + help='Delete messages after writing (destructive drain; default peek)') args = parser.parse_args() - # Append environment name to role names for uniqueness across accounts source_env_name = ACCOUNT_NAMES.get(args.source_account, args.source_account) dest_env_name = ACCOUNT_NAMES.get(args.dest_account, args.dest_account) print(f"Source environment name: {source_env_name}") print(f"Destination environment name: {dest_env_name}") - replication_role = f"s3-cross-account-replication-role-for-{source_env_name}-to-{dest_env_name}" - datasync_role = f"datasync-s3-access-role-for-{source_env_name}-to-{dest_env_name}" + names = derive_resource_names(source_env_name, dest_env_name) config = MigrationConfig( source_profile=args.source_profile, @@ -266,8 +282,9 @@ def main(): dest_account_id=args.dest_account, region=args.region, enable_rtc=not args.disable_rtc, - replication_role_name=replication_role, - datasync_role_name=datasync_role, + replication_role_name=names["replication_role_name"], + failures_queue_name=names["failures_queue_name"], + threshold_queue_name=names["threshold_queue_name"], ) ctx = S3MigrationContext(config) @@ -344,8 +361,34 @@ def main(): elif args.command == 'status': _check_prerequisites(ctx) get_replication_status(ctx) + get_monitoring_status(ctx) list_datasync_tasks(ctx) + elif args.command == 'setup-monitoring': + if args.create_monitoring: + queues = create_monitoring_queues(ctx) + if queues.get('failures') and queues.get('threshold'): + configure_bucket_notifications( + ctx, queues['failures']['arn'], queues['threshold']['arn']) + else: + print("\nERROR: queue creation failed; skipping notification config.") + else: + get_monitoring_status(ctx) + + elif args.command == 'drain-queue': + if args.queue == 'failures': + queue_url = ctx.source_sqs.get_queue_url( + QueueName=config.failures_queue_name)['QueueUrl'] + elif args.queue == 'threshold': + queue_url = ctx.source_sqs.get_queue_url( + QueueName=config.threshold_queue_name)['QueueUrl'] + else: + queue_url = args.queue + paths, total = drain_queue_to_jsonl( + ctx, queue_url, args.output_prefix, + delete=args.delete, max_messages=args.max_messages) + print(f"\nDrained {total} messages into {len(paths)} file(s)") + if __name__ == '__main__': main() diff --git a/hq_s3_migration/orchestrator.py b/hq_s3_migration/orchestrator.py index 729dc5f6b5..3ce42786a5 100644 --- a/hq_s3_migration/orchestrator.py +++ b/hq_s3_migration/orchestrator.py @@ -25,5 +25,6 @@ def __init__(self, config: MigrationConfig): self.source_iam = self.source_session.client('iam') self.dest_iam = self.dest_session.client('iam') self.source_datasync = self.source_session.client('datasync') + self.source_sqs = self.source_session.client('sqs') self.source_sts = self.source_session.client('sts') self.dest_sts = self.dest_session.client('sts') diff --git a/hq_s3_migration/policies/datasync_report_role.json b/hq_s3_migration/policies/datasync_report_role.json new file mode 100644 index 0000000000..5484656a76 --- /dev/null +++ b/hq_s3_migration/policies/datasync_report_role.json @@ -0,0 +1,17 @@ +{ + "Version": "2012-10-17", + "Statement": [ + { + "Action": [ + "s3:PutObject" + ], + "Effect": "Allow", + "Resource": "arn:aws:s3:::{report_bucket}/*", + "Condition": { + "StringEquals": { + "s3:ResourceAccount": "{source_account_id}" + } + } + } + ] +} From fad1e36dc81232b3a6ae514211cd65b61bf11f84 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 08:49:14 +0530 Subject: [PATCH 18/21] configure datasync to have a reporting bucket where all the details will be pushed --- hq_s3_migration/cli.py | 5 +++ hq_s3_migration/config.py | 4 +++ hq_s3_migration/datasync.py | 13 +++++++ hq_s3_migration/iam.py | 67 +++++++++++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+) diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index c8d32433a1..91fd2a088b 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -283,6 +283,9 @@ def main(): region=args.region, enable_rtc=not args.disable_rtc, replication_role_name=names["replication_role_name"], + datasync_role_name=names["datasync_role_name"], + report_bucket_name=names["report_bucket_name"], + report_role_name=names["report_role_name"], failures_queue_name=names["failures_queue_name"], threshold_queue_name=names["threshold_queue_name"], ) @@ -317,6 +320,8 @@ def main(): create_replication_role(ctx) create_datasync_role(ctx) apply_destination_bucket_policy(ctx) + create_report_bucket(ctx) + create_report_role(ctx) elif args.command == 'enable-replication': enable_live_replication(ctx) diff --git a/hq_s3_migration/config.py b/hq_s3_migration/config.py index 1440921d8b..c7e822c9d9 100644 --- a/hq_s3_migration/config.py +++ b/hq_s3_migration/config.py @@ -26,6 +26,10 @@ class MigrationConfig: region: str replication_role_name: str datasync_role_name: str + report_bucket_name: str + report_role_name: str + failures_queue_name: str + threshold_queue_name: str enable_rtc: bool = True # S3 Replication Time Control (15-min SLA) enable_delete_replication: bool = True diff --git a/hq_s3_migration/datasync.py b/hq_s3_migration/datasync.py index 5266cb064d..a8a774b6b7 100644 --- a/hq_s3_migration/datasync.py +++ b/hq_s3_migration/datasync.py @@ -59,6 +59,8 @@ def create_datasync_task(ctx: S3MigrationContext, cfg = ctx.config print(f"\nCreating DataSync task...") + report_role_arn = f"arn:aws:iam::{cfg.source_account_id}:role/{cfg.report_role_name}" + try: response = ctx.source_datasync.create_task( SourceLocationArn=source_location_arn, @@ -78,6 +80,17 @@ def create_datasync_task(ctx: S3MigrationContext, 'ObjectTags': 'PRESERVE', 'LogLevel': 'BASIC', }, + TaskReportConfig={ + 'Destination': { + 'S3': { + 'S3BucketArn': f'arn:aws:s3:::{cfg.report_bucket_name}', + 'BucketAccessRoleArn': report_role_arn, + } + }, + 'OutputType': 'STANDARD', + 'ReportLevel': 'SUCCESSES_AND_ERRORS', + 'ObjectVersionIds': 'INCLUDE', + }, ) task_arn = response['TaskArn'] print(f" Created task: {task_arn}") diff --git a/hq_s3_migration/iam.py b/hq_s3_migration/iam.py index 98d5d5f6ad..26119f4a14 100644 --- a/hq_s3_migration/iam.py +++ b/hq_s3_migration/iam.py @@ -62,6 +62,17 @@ def print_iam_policies(ctx: S3MigrationContext): dest_policy['Statement'].append(datasync_stmt) print(json.dumps(dest_policy, indent=2)) + print("\n" + "-" * 40) + print(f"6. DATASYNC REPORT ROLE POLICY (Source Account: {cfg.source_account_id})") + print(f" Role Name: {cfg.report_role_name}") + print(f" Report Bucket: {cfg.report_bucket_name}") + print("-" * 40) + print(json.dumps( + render_policy("datasync_report_role.json", + report_bucket=cfg.report_bucket_name, + source_account_id=cfg.source_account_id), + indent=2)) + def create_replication_role(ctx: S3MigrationContext) -> Optional[str]: """Create IAM role for S3 replication in source account.""" @@ -225,3 +236,59 @@ def apply_destination_bucket_policy(ctx: S3MigrationContext) -> bool: except ClientError as e: print(f" ERROR: {e}") return False + + +def create_report_role(ctx: S3MigrationContext) -> Optional[str]: + """Create IAM role allowing DataSync to write task reports to the report bucket.""" + cfg = ctx.config + print(f"\nCreating report role '{cfg.report_role_name}'...") + + trust_policy = render_policy("datasync_trust.json") + role_policy = render_policy("datasync_report_role.json", + report_bucket=cfg.report_bucket_name, + source_account_id=cfg.source_account_id) + + try: + response = ctx.source_iam.create_role( + RoleName=cfg.report_role_name, + AssumeRolePolicyDocument=json.dumps(trust_policy), + Description="Role for DataSync task report writes", + ) + role_arn = response["Role"]["Arn"] + print(f" Created role: {role_arn}") + ctx.source_iam.put_role_policy( + RoleName=cfg.report_role_name, + PolicyName="DataSyncReportPolicy", + PolicyDocument=json.dumps(role_policy), + ) + print(f" Attached report policy") + print(f" Waiting for role propagation...") + time.sleep(10) + return role_arn + except ClientError as e: + if e.response["Error"]["Code"] == "EntityAlreadyExists": + print(f" Role already exists, updating policy...") + ctx.source_iam.put_role_policy( + RoleName=cfg.report_role_name, + PolicyName="DataSyncReportPolicy", + PolicyDocument=json.dumps(role_policy), + ) + return ctx.source_iam.get_role(RoleName=cfg.report_role_name)["Role"]["Arn"] + print(f" ERROR: {e}") + return None + + +def create_report_bucket(ctx: S3MigrationContext) -> bool: + """Create the DataSync report bucket in the source account.""" + cfg = ctx.config + print(f"\nCreating report bucket '{cfg.report_bucket_name}'...") + try: + ctx.source_s3.create_bucket(Bucket=cfg.report_bucket_name) + print(f" Created bucket") + return True + except ClientError as e: + if e.response["Error"]["Code"] in ("BucketAlreadyOwnedByYou", "BucketAlreadyExists"): + print(f" Bucket already exists") + return True + print(f" ERROR: {e}") + return False From 73e49a29a031c50cd197ffc177a6d6e835bd5752 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 15:40:50 +0530 Subject: [PATCH 19/21] shorten role name - aws allows only 64 charachters --- hq_s3_migration/config.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/hq_s3_migration/config.py b/hq_s3_migration/config.py index c7e822c9d9..c6f04cda8f 100644 --- a/hq_s3_migration/config.py +++ b/hq_s3_migration/config.py @@ -5,10 +5,10 @@ def derive_resource_names(source_env: str, dest_env: str) -> dict: """Derive all per-migration resource names from env-name aliases.""" suffix = f"{source_env}-to-{dest_env}" return { - "replication_role_name": f"s3-cross-account-replication-role-for-{suffix}", - "datasync_role_name": f"datasync-s3-access-role-for-{suffix}", + "replication_role_name": f"s3-replication-role-{suffix}", + "datasync_role_name": f"datasync-s3-role-{suffix}", "report_bucket_name": f"s3-migration-{suffix}-datasync-run-reports", - "report_role_name": f"datasync-report-access-role-for-{suffix}", + "report_role_name": f"datasync-report-role-{suffix}", "failures_queue_name": f"s3-replication-failures-{suffix}", "threshold_queue_name": f"s3-replication-threshold-{suffix}", } From 5b6d935e6c5755093b1560dde41b1fe1396281ff Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Mon, 22 Jun 2026 23:54:27 +0530 Subject: [PATCH 20/21] add ability to print queue errors --- hq_s3_migration/cli.py | 33 ++++++++++++++++++++--- hq_s3_migration/monitoring.py | 51 +++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/hq_s3_migration/cli.py b/hq_s3_migration/cli.py index 91fd2a088b..32688e857d 100644 --- a/hq_s3_migration/cli.py +++ b/hq_s3_migration/cli.py @@ -36,6 +36,7 @@ create_monitoring_queues, drain_queue_to_jsonl, get_monitoring_status, + print_queue_messages, ) from .orchestrator import S3MigrationContext from .replication import enable_live_replication, get_replication_status @@ -206,6 +207,8 @@ def main(): validate Validate migration status cutover-check Run pre-cutover checklist status Show current migration status + setup-monitoring Create/inspect SQS replication-event monitoring + drain-queue Drain a monitoring queue to JSONL files Examples: # Check prerequisites @@ -223,6 +226,21 @@ def main(): # Create and start DataSync task python -m hq_s3_migration create-datasync --source-bucket my-source --dest-bucket my-dest \\ --source-account 111111111111 --dest-account 222222222222 + + # Create SQS replication-event monitoring (queues + bucket notifications) + python -m hq_s3_migration setup-monitoring --create-monitoring \\ + --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + + # Inspect monitoring status (omit --create-monitoring) + python -m hq_s3_migration setup-monitoring --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 + + # Drain the failures queue to JSONL files (delete as you go) + python -m hq_s3_migration drain-queue --queue failures --delete \\ + --output-prefix replication-failures \\ + --source-bucket my-source --dest-bucket my-dest \\ + --source-account 111111111111 --dest-account 222222222222 """ ) @@ -264,6 +282,8 @@ def main(): help='Optional cap on messages to drain') parser.add_argument('--delete', action='store_true', help='Delete messages after writing (destructive drain; default peek)') + parser.add_argument('--print', dest='print_console', action='store_true', + help='Print message bodies to the console instead of writing JSONL files (for drain-queue)') args = parser.parse_args() @@ -389,10 +409,15 @@ def main(): QueueName=config.threshold_queue_name)['QueueUrl'] else: queue_url = args.queue - paths, total = drain_queue_to_jsonl( - ctx, queue_url, args.output_prefix, - delete=args.delete, max_messages=args.max_messages) - print(f"\nDrained {total} messages into {len(paths)} file(s)") + if args.print_console: + print_queue_messages( + ctx, queue_url, + delete=args.delete, max_messages=args.max_messages) + else: + paths, total = drain_queue_to_jsonl( + ctx, queue_url, args.output_prefix, + delete=args.delete, max_messages=args.max_messages) + print(f"\nDrained {total} messages into {len(paths)} file(s)") if __name__ == '__main__': diff --git a/hq_s3_migration/monitoring.py b/hq_s3_migration/monitoring.py index 4a3d849888..f651f04be4 100644 --- a/hq_s3_migration/monitoring.py +++ b/hq_s3_migration/monitoring.py @@ -159,6 +159,57 @@ def get_monitoring_status(ctx: S3MigrationContext) -> dict: return status +def print_queue_messages(ctx: S3MigrationContext, + queue_url: str, + delete: bool = False, + max_messages=None) -> int: + """Print queue message bodies to the console. Returns count printed. + + With delete=False (default) this is a non-destructive peek; full coverage is + NOT guaranteed (visibility timeout + duplicates). + """ + if not delete: + print(" NOTE: peek mode (no delete) may show duplicates and miss some " + "in-flight messages.") + + total = 0 + while True: + if max_messages is not None and total >= max_messages: + break + resp = ctx.source_sqs.receive_message( + QueueUrl=queue_url, + MaxNumberOfMessages=10, + WaitTimeSeconds=10, + ) + messages = resp.get("Messages", []) + if not messages: + break + + handles = [] + for msg in messages: + total += 1 + body = msg.get("Body", "") + print("\n" + "-" * 60) + print(f"Message {total}:") + try: + print(json.dumps(json.loads(body), indent=2)) + except (ValueError, TypeError): + print(body) + handles.append(msg["ReceiptHandle"]) + if max_messages is not None and total >= max_messages: + break + + if delete and handles: + ctx.source_sqs.delete_message_batch( + QueueUrl=queue_url, + Entries=[{"Id": str(i), "ReceiptHandle": h} + for i, h in enumerate(handles)], + ) + + print(f"\n Printed {total} message(s)") + return total + + FILE_ROTATION = 10000 From b3c44c4d06e8851098defbcebf44ef918bf32f28 Mon Sep 17 00:00:00 2001 From: Amit Phulera Date: Tue, 23 Jun 2026 00:58:20 +0530 Subject: [PATCH 21/21] fix the datasync settings to the ones that we have used in the staging process --- hq_s3_migration/datasync.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/hq_s3_migration/datasync.py b/hq_s3_migration/datasync.py index a8a774b6b7..b83d15b55f 100644 --- a/hq_s3_migration/datasync.py +++ b/hq_s3_migration/datasync.py @@ -69,8 +69,10 @@ def create_datasync_task(ctx: S3MigrationContext, TaskMode='ENHANCED', Options={ 'VerifyMode': 'ONLY_FILES_TRANSFERRED', - 'OverwriteMode': 'ALWAYS', - 'PreserveDeletedFiles': 'REMOVE', + # NEVER + PRESERVE so the backfill never clobbers or deletes + # objects that live replication is managing in the destination. + 'OverwriteMode': 'NEVER', + 'PreserveDeletedFiles': 'PRESERVE', 'PreserveDevices': 'NONE', 'PosixPermissions': 'NONE', 'Uid': 'NONE',