Forem: Daffa Rabbani

🔐Automated Security Incident Response System ⚠

Daffa Rabbani — Fri, 26 Dec 2025 12:57:41 +0000

Implementation Guide - 2 Days (No Docker)

Based on: CrowdStrike Incident July 2024 - $5.4B Losses Timeline: 2 Hari | Budget: ~$90/bulan | Complexity: Advanced

📋 Prerequisites

What You Need:

AWS Account with admin access (Best Practice Please Use your IAM User * with Admin Access not your Root Account)
Basic Python knowledge
Text editor (VS Code, Notepad++)
AWS CLI installed (optional, can just use Console)

Budget Breakdown:

GuardDuty: $4.50/month (30-day free trial)
Security Hub: $0.0010 per check = ~$30/month
Lambda: $0.20/million requests = ~$5/month
Step Functions: $25 per 1K executions = ~$10/month
CloudTrail: $2 per 100K events = ~$20/month
SNS: $0.50/million notifications = ~$1/month
Config: $2 per active rule = ~$20/month

Total: ~$90/month

🏗️ Architecture Overview

📅 DAY 1: Foundation & Detection (8 Hours)

📝 PHASE 1: Setup GuardDuty (1 Hour)

step 1.1 : Enable GuardDuty

via AWS Console:

Login AWS Console → Search "GuardDuty"
Click Get Started
Click Enable GuardDuty
Wait 15-30 seconds for initialization

Via AWS CLI:

aws guardduty create-detector \
  --enable \
  --finding-publishing-frequency FIFTEEN_MINUTES

✅ Verification: GuardDuty console will show "No findings" - this is normal, it takes time to detect threats

Step 1.2: Generate Sample Findings (for Testing)

EXAMPLE :

Get detector ID

DETECTOR_ID=$(aws guardduty list-detectors --query 'DetectorIds[0]' --output text)

Generate sample findings

aws guardduty create-sample-findings \
  --detector-id $DETECTOR_ID \
  --finding-types "Recon:EC2/PortProbeUnprotectedPort" \
            "UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration"\
              "CryptoCurrency:EC2/BitcoinTool.B!DNS"

Via Console:

GuardDuty → Settings → Sample findings
Click Generate sample findings

✅ Verification: GuardDuty → Findings → You'll see 50+ sample findings

📝 PHASE 2: Setup Security Hub (1 Hour)

Security Hub adalah centralized security dashboard.

Step 2.1: Enable Security Hub

Via Console:

Search "Security Hub"
Click Go to Security Hub
Click Enable Security Hub
Select AWS Foundational Security Best Practices standard
Click Enable Security Hub

Via CLI:

aws securityhub enable-security-hub \
  --enable-default-standards

Wait 5-10 minutes untuk Security Hub populate findings.

Step 2.2: Enable GuardDuty Integration

Security Hub → Integrations
Find AWS GuardDuty
Click Accept findings

✅ Verification: Security Hub → Findings → akan muncul findings dari GuardDuty

📝 PHASE 3: Setup SNS for Alerts (30 Minutes)

Step 3.1: Create SNS Topic

Via Console:

Search SNS → Topics
Click Create topic
Type: Standard
Name: security-incident-alerts
Display name: Security Alerts
Click Create topic

Save the Topic ARN - it will look like:

arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT_ID:security-incident-alerts

Step 3.2: Create Email Subscription

Topic → Create subscription
Protocol: Email
Endpoint: your-email@company.com
Click Create subscription
Check your email → Click confirmation link

Step 3.3: Create Slack Subscription (Optional)

Via Console:

Topic → Create subscription
Protocol: AWS Chatbot
Configure Slack workspace integration

Or use webhook:

We'll integrate this in Lambda later

✅ Verification: Send test message:
Example:

aws sns publish \
  --topic-arn arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT_ID:security-incident-alerts \
  --message "Test alert - Security system operational"

📝 PHASE 4: Create IAM Role for Lambda (30 Minutes)

Step 4.1: Create Execution Role

Via Console:

IAM Console → Roles → Create role
Trusted entity: AWS service → Lambda
Permissions policies - Add these:

AWSLambdaBasicExecutionRole (managed)
AmazonEC2FullAccess (managed)
IAMFullAccess (managed)
AmazonSNSFullAccess (managed)

Role name: SecurityIncidentResponseRole
Click Create role

Step 4.2: Add Inline Policy for Additional Permissions

Role → Add permissions → Create inline policy
JSON tab → Paste:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ec2:DescribeInstances",
        "ec2:DescribeSecurityGroups",
        "ec2:CreateSecurityGroup",
        "ec2:AuthorizeSecurityGroupIngress",
        "ec2:AuthorizeSecurityGroupEgress",
        "ec2:RevokeSecurityGroupIngress",
        "ec2:RevokeSecurityGroupEgress",
        "ec2:ModifyInstanceAttribute",
        "ec2:CreateSnapshot",
        "ec2:CreateTags",
        "iam:ListAccessKeys",
        "iam:DeleteAccessKey",
        "iam:UpdateLoginProfile",
        "iam:CreateAccessKey",
        "sts:GetCallerIdentity",
        "guardduty:GetFindings",
        "guardduty:ListFindings",
        "securityhub:GetFindings",
        "cloudtrail:LookupEvents"
      ],
      "Resource": "*"
    }
  ]
}

Name: SecurityResponsePermissions
Click Create policy

📝 PHASE 5: Create Auto-Response Lambda Functions (3 Hours)

Step 5.1: Create Main Response Function

Via Console:

Lambda Console → Create function
Function name: security-incident-responder
Runtime: Python 3.11
Architecture: x86_64
Execution role: Use existing role → SecurityIncidentResponseRole
Click Create function

Step 5.2: Write Lambda Code (No Docker Needed!)

Click Code tab, replace with this code:

import json
import boto3
import os
from datetime import datetime


ec2 = boto3.client('ec2')
iam = boto3.client('iam')
sns = boto3.client('sns')
cloudtrail = boto3.client('cloudtrail')


SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN', 'arn:aws:sns:{Your_Region}:{Your_ACCOUNT_ID}:security-incident-alerts')


def lambda_handler(event, context):
    """Main handler for security incidents"""

    print(f"Received event: {json.dumps(event, indent=2)}")

    try:
        # Parse the finding - handle both GuardDuty and Security Hub formats
        if 'findings' in event.get('detail', {}):
            # Security Hub format
            finding = event['detail']['findings'][0]
        else:
            # GuardDuty native format
            finding = event['detail']

        # Get severity - handle different formats
        if isinstance(finding.get('Severity'), dict):
            severity = finding['Severity'].get('Label', 'UNKNOWN')
        else:
            # Map numeric severity to label
            severity_num = finding.get('severity', 0)
            if severity_num >= 7:
                severity = 'HIGH'
            elif severity_num >= 8.5:
                severity = 'CRITICAL'
            elif severity_num >= 4:
                severity = 'MEDIUM'
            else:
                severity = 'LOW'

        finding_type = finding.get('type', 'Unknown')
        finding_id = finding.get('id', 'Unknown')

        print(f"Processing finding: {finding_type} (Severity: {severity})")

        # Only respond to HIGH and CRITICAL
        if severity not in ['HIGH', 'CRITICAL']:
            print(f"Severity {severity} below threshold, skipping automated response")
            return {
                'statusCode': 200,
                'body': json.dumps('Severity below threshold')
            }

        # Route to appropriate handler
        response_actions = []

        if 'UnauthorizedAccess:EC2' in finding_type or 'Backdoor:EC2' in finding_type or 'Trojan:Runtime' in finding_type:
            instance_id = extract_instance_id(finding)
            if instance_id:
                result = isolate_ec2_instance(instance_id, finding_type)
                response_actions.append(result)

        elif 'UnauthorizedAccess:IAMUser' in finding_type or 'CredentialAccess' in finding_type:
            user_name = extract_user_name(finding)
            if user_name:
                result = rotate_iam_credentials(user_name, finding_type)
                response_actions.append(result)

        elif 'Impact:EC2/PortSweep' in finding_type or 'Recon:EC2' in finding_type:
            source_ip = extract_source_ip(finding)
            if source_ip:
                result = block_suspicious_ip(source_ip, finding_type)
                response_actions.append(result)

        elif 'CryptoCurrency' in finding_type:
            instance_id = extract_instance_id(finding)
            if instance_id:
                result = handle_crypto_mining(instance_id, finding_type)
                response_actions.append(result)

        # Send comprehensive alert
        send_security_alert(finding, response_actions)

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Security incident processed',
                'finding_id': finding_id,
                'actions_taken': response_actions
            })
        }

    except Exception as e:
        print(f"Error processing security incident: {str(e)}")
        import traceback
        traceback.print_exc()
        send_error_alert(str(e), event)

        return {
            'statusCode': 500,
            'body': json.dumps(f'Error: {str(e)}')
        }


def extract_instance_id(finding):
    """Extract EC2 instance ID from finding"""
    try:
        for resource in finding.get('Resources', []):
            if resource.get('Type') == 'Instance':
                instance_id = resource.get('Id', '').split('/')[-1]
                return instance_id if instance_id.startswith('i-') else None
    except Exception as e:
        print(f"Error extracting instance ID: {e}")
    return None


def extract_user_name(finding):
    """Extract IAM user name from finding"""
    try:
        for resource in finding.get('Resources', []):
            if resource.get('Type') == 'AccessKey':
                return resource.get('AccessKeyDetails', {}).get('UserName')
    except Exception as e:
        print(f"Error extracting user name: {e}")
    return None


def extract_source_ip(finding):
    """Extract source IP from finding"""
    try:
        service = finding.get('Service', {})
        action = service.get('Action', {})
        network_connection = action.get('NetworkConnectionAction', {})
        return network_connection.get('RemoteIpDetails', {}).get('IpAddressV4')
    except Exception as e:
        print(f"Error extracting source IP: {e}")
    return None


def isolate_ec2_instance(instance_id, finding_type):
    """Isolate compromised EC2 instance"""
    try:
        print(f"Isolating instance: {instance_id}")

        # Get instance details
        response = ec2.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]
        vpc_id = instance['VpcId']

        # Create isolation security group
        sg_name = f'isolation-{instance_id}-{datetime.now().strftime("%Y%m%d%H%M%S")}'

        sg_response = ec2.create_security_group(
            GroupName=sg_name,
            Description=f'Quarantine SG for {instance_id} - {finding_type}',
            VpcId=vpc_id
        )

        isolation_sg_id = sg_response['GroupId']

        # Remove all default egress rules (blocks all outbound)
        ec2.revoke_security_group_egress(
            GroupId=isolation_sg_id,
            IpPermissions=[{
                'IpProtocol': '-1',
                'FromPort': -1,
                'ToPort': -1,
                'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
            }]
        )

        # Add only SSH access from admin IP (optional)
        # ec2.authorize_security_group_ingress(
        #     GroupId=isolation_sg_id,
        #     IpPermissions=[{
        #         'IpProtocol': 'tcp',
        #         'FromPort': 22,
        #         'ToPort': 22,
        #         'IpRanges': [{'CidrIp': 'YOUR_ADMIN_IP/32', 'Description': 'Admin access'}]
        #     }]
        # )

        # Apply isolation security group to instance
        ec2.modify_instance_attribute(
            InstanceId=instance_id,
            Groups=[isolation_sg_id]
        )

        # Create snapshot for forensics
        volumes = [vol['Ebs']['VolumeId'] for vol in instance.get('BlockDeviceMappings', [])]
        snapshots = []

        for volume_id in volumes:
            snapshot = ec2.create_snapshot(
                VolumeId=volume_id,
                Description=f'Forensic snapshot - {finding_type}',
                TagSpecifications=[{
                    'ResourceType': 'snapshot',
                    'Tags': [
                        {'Key': 'Purpose', 'Value': 'Forensics'},
                        {'Key': 'InstanceId', 'Value': instance_id},
                        {'Key': 'Incident', 'Value': finding_type}
                    ]
                }]
            )
            snapshots.append(snapshot['SnapshotId'])

        action_summary = {
            'action': 'isolate_instance',
            'instance_id': instance_id,
            'isolation_sg': isolation_sg_id,
            'forensic_snapshots': snapshots,
            'status': 'success'
        }

        print(f"Instance {instance_id} successfully isolated")
        return action_summary

    except Exception as e:
        print(f"Error isolating instance {instance_id}: {e}")
        return {
            'action': 'isolate_instance',
            'instance_id': instance_id,
            'status': 'failed',
            'error': str(e)
        }


def rotate_iam_credentials(user_name, finding_type):
    """Rotate compromised IAM credentials"""
    try:
        print(f"Rotating credentials for user: {user_name}")

        actions_taken = []

        # List and delete all access keys
        keys_response = iam.list_access_keys(UserName=user_name)
        deleted_keys = []

        for key in keys_response['AccessKeyMetadata']:
            access_key_id = key['AccessKeyId']

            # Deactivate first (safer than immediate delete)
            iam.update_access_key(
                UserName=user_name,
                AccessKeyId=access_key_id,
                Status='Inactive'
            )

            deleted_keys.append(access_key_id)
            actions_taken.append(f'Deactivated access key: {access_key_id}')

        # Force password reset (if console access exists)
        try:
            iam.update_login_profile(
                UserName=user_name,
                PasswordResetRequired=True
            )
            actions_taken.append('Forced password reset')
        except iam.exceptions.NoSuchEntityException:
            actions_taken.append('No console access to reset')

        # Attach deny-all policy (additional safety)
        policy_document = {
            "Version": "2012-10-17",
            "Statement": [{
                "Effect": "Deny",
                "Action": "*",
                "Resource": "*"
            }]
        }

        try:
            iam.put_user_policy(
                UserName=user_name,
                PolicyName='EmergencyDenyAll',
                PolicyDocument=json.dumps(policy_document)
            )
            actions_taken.append('Applied emergency deny policy')
        except Exception as e:
            print(f"Could not apply deny policy: {e}")

        action_summary = {
            'action': 'rotate_credentials',
            'user_name': user_name,
            'deactivated_keys': deleted_keys,
            'actions': actions_taken,
            'status': 'success'
        }

        print(f"Credentials rotated for {user_name}")
        return action_summary

    except Exception as e:
        print(f"Error rotating credentials for {user_name}: {e}")
        return {
            'action': 'rotate_credentials',
            'user_name': user_name,
            'status': 'failed',
            'error': str(e)
        }
def block_suspicious_ip(source_ip, finding_type):
    """Block suspicious IP using NACL"""
    try:
        print(f"Blocking suspicious IP: {source_ip}")

        # Get default VPC
        vpcs = ec2.describe_vpcs(Filters=[{'Name': 'isDefault', 'Values': ['true']}])

        if not vpcs['Vpcs']:
            return {
                'action': 'block_ip',
                'source_ip': source_ip,
                'status': 'failed',
                'error': 'No default VPC found'
            }

        vpc_id = vpcs['Vpcs'][0]['VpcId']

        # Get network ACLs for VPC
        nacls = ec2.describe_network_acls(
            Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}]
        )

        if not nacls['NetworkAcls']:
            return {
                'action': 'block_ip',
                'source_ip': source_ip,
                'status': 'failed',
                'error': 'No NACLs found'
            }

        nacl_id = nacls['NetworkAcls'][0]['NetworkAclId']

        # Add deny rule (rule number 1 for highest priority)
        ec2.create_network_acl_entry(
            NetworkAclId=nacl_id,
            RuleNumber=1,
            Protocol='-1',  # All protocols
            RuleAction='deny',
            CidrBlock=f'{source_ip}/32'
        )

        action_summary = {
            'action': 'block_ip',
            'source_ip': source_ip,
            'nacl_id': nacl_id,
            'status': 'success'
        }

        print(f"IP {source_ip} successfully blocked")
        return action_summary

    except Exception as e:
        print(f"Error blocking IP {source_ip}: {e}")
        return {
            'action': 'block_ip',
            'source_ip': source_ip,
            'status': 'failed',
            'error': str(e)
        }


def handle_crypto_mining(instance_id, finding_type):
    """Handle cryptocurrency mining detection"""
    try:
        print(f"Handling crypto mining on instance: {instance_id}")

        # Stop the instance immediately
        ec2.stop_instances(InstanceIds=[instance_id])

        # Create forensic snapshot
        response = ec2.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]
        volumes = [vol['Ebs']['VolumeId'] for vol in instance.get('BlockDeviceMappings', [])]

        snapshots = []
        for volume_id in volumes:
            snapshot = ec2.create_snapshot(
                VolumeId=volume_id,
                Description=f'Crypto mining forensics - {instance_id}'
            )
            snapshots.append(snapshot['SnapshotId'])

        action_summary = {
            'action': 'stop_crypto_mining',
            'instance_id': instance_id,
            'instance_stopped': True,
            'forensic_snapshots': snapshots,
            'status': 'success'
        }

        print(f"Crypto mining instance {instance_id} stopped")
        return action_summary

    except Exception as e:
        print(f"Error handling crypto mining on {instance_id}: {e}")
        return {
            'action': 'stop_crypto_mining',
            'instance_id': instance_id,
            'status': 'failed',
            'error': str(e)
        }


def send_security_alert(finding, response_actions):
    """Send comprehensive security alert"""
    try:
        # Get severity - handle different formats
        if isinstance(finding.get('Severity'), dict):
            severity = finding['Severity'].get('Label', 'UNKNOWN')
        else:
            severity_num = finding.get('severity', 0)
            if severity_num >= 8.5:
                severity = 'CRITICAL'
            elif severity_num >= 7:
                severity = 'HIGH'
            elif severity_num >= 4:
                severity = 'MEDIUM'
            else:
                severity = 'LOW'

        # Get finding type (lowercase 'type' for GuardDuty, uppercase 'Type' for Security Hub)
        finding_type = finding.get('type') or finding.get('Type', 'Unknown')

        # Get description
        description = finding.get('description') or finding.get('Description', 'No description')

        # Get finding ID
        finding_id = finding.get('id') or finding.get('Id', 'Unknown')

        # Build alert message
        message = f"""
🚨 SECURITY INCIDENT DETECTED 🚨


Severity: {severity}
Type: {finding_type}
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}


Description:
{description}


AUTOMATED ACTIONS TAKEN:
"""

        if response_actions:
            for action in response_actions:
                message += f"\n✓ {action.get('action', 'unknown')}: {action.get('status', 'unknown')}"
                if action.get('status') == 'failed':
                    message += f" - Error: {action.get('error', 'unknown')}"
        else:
            message += "\nNo automated actions taken (instance not found or not applicable)"

        message += "\n\nSOC Team: Please review and investigate further."
        message += f"\n\nFinding ID: {finding_id}"

        # Send to SNS
        sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject=f'🚨 SECURITY ALERT: {finding_type}',
            Message=message
        )

        print("Security alert sent successfully")

    except Exception as e:
        print(f"Error sending security alert: {e}")
        import traceback
        traceback.print_exc()


def send_error_alert(error_message, event):
    """Send alert when response function fails"""
    try:
        message = f"""
❌ SECURITY RESPONSE ERROR


An error occurred while processing a security incident.


Error: {error_message}


Event: {json.dumps(event, indent=2)}


Action Required: Manual investigation needed.
"""

        sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject='❌ Security Response System Error',
            Message=message
        )

    except Exception as e:
        print(f"Failed to send error alert: {e}")

TEST CODE

Example JSON:

{
  "version": "0",
  "id": "test-event-12345",
  "detail-type": "GuardDuty Finding",
  "source": "aws.guardduty",
  "account": "{YOUR_ACCOUNT_ID}",
  "time": "2024-12-17T10:00:00Z",
  "region": "ap-southeast-2",
  "resources": [],
  "detail": {
    "schemaVersion": "2.0",
    "accountId": "{YOUR_ACCOUNT_ID}",
    "region": "ap-southeast-2",
    "partition": "aws",
    "id": "test-finding-ec2-123",
    "arn": "arn:aws:guardduty:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:detector/test/finding/test-123",
    "type": "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom",
    "severity": 8,
    "createdAt": "2024-12-17T09:35:00.000Z",
    "updatedAt": "2024-12-17T09:35:00.000Z",
    "title": "EC2 instance is communicating with a malicious IP address",
    "description": "EC2 instance i-test12345678 is communicating with malicious IP 198.51.100.1. This is a TEST finding for security automation validation.",
    "service": {
      "serviceName": "guardduty",
      "detectorId": "test-detector-id",
      "action": {
        "actionType": "NETWORK_CONNECTION",
        "networkConnectionAction": {
          "blocked": false,
          "connectionDirection": "INBOUND",
          "localPortDetails": {
            "port": 22,
            "portName": "SSH"
          },
          "protocol": "TCP",
          "remoteIpDetails": {
            "ipAddressV4": "198.51.100.1",
            "organization": {
              "asn": "12345",
              "asnOrg": "Malicious ASN"
            },
            "country": {
              "countryName": "Unknown"
            }
          }
        }
      },
      "archived": false,
      "count": 1,
      "eventFirstSeen": "2024-12-17T09:30:00.000Z",
      "eventLastSeen": "2024-12-17T09:35:00.000Z",
      "resourceRole": "TARGET"
    },
    "Severity": {
      "Label": "HIGH",
      "Normalized": 70,
      "Product": 8.0
    },
    "Resources": [
      {
        "Type": "Instance",
        "Id": "arn:aws:ec2:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:instance/i-test12345678",
        "Details": {
          "Instance": {
            "InstanceId": "i-test12345678",
            "InstanceType": "t2.micro",
            "LaunchTime": "2024-12-17T09:00:00.000Z",
            "Platform": "Linux"
          }
        }
      }
    ],
    "Id": "test-finding-ec2-123"
  }
}

Step 5.3: Configure Lambda

Configuration tab → Environment variables
Add variable:

Key: SNS_TOPIC_ARN
Value: arn:aws:sns:{YOUR_REGION}:{ACCOUNT_ID}:security-incident alerts (your SNS ARN) (Your SNS arn)

General configuration → Edit:
- Memory: 512 MB
- Timeout: 5 minutes

Click Save
Deploy the Code

✅ Verification: Lambda function created successfully

📝 PHASE 6: Connect EventBridge to Lambda (1 Hour)

Step 6.1: Create EventBridge Rule for GuardDuty

Via Console:

EventBridge Console → Rules → Create rule
Name: guardduty-high-severity-findings
Description: Route high severity GuardDuty findings to Lambda (optional)
Event bus: default
Rule type: Rule with an event pattern
Click Next
Event pattern:

{
  "source": ["aws.guardduty"],
  "detail-type": ["GuardDuty Finding"],
  "detail": {
    "severity": [7, 7.0, 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.7, 7.8, 7.9, 8, 8.0, 8.1, 8.2, 8.3, 8.4, 8.5, 8.6, 8.7, 8.8, 8.9]
  }
}

Click Next
Select target:

Target type: AWS service
Select a target: Lambda function
Function: security-incident-responder

Click Next → Create rule

Step 6.2: Create Rule for Security Hub

Create another rule: securityhub-critical-findings
Event pattern:

{
  "source": ["aws.securityhub"],
  "detail-type": ["Security Hub Findings - Imported"],
  "detail": {
    "findings": {
      "Severity": {
        "Label": ["CRITICAL", "HIGH"]
      }
    }
  }
}

Target: Same Lambda function

✅ Verification: EventBridge → Rules → 2 rules active

📝 PHASE 7: Testing with Sample Findings (1 Hour)

Step 7.1: Trigger Test Finding

Example :

# Generate GuardDuty sample finding
aws guardduty create-sample-findings \
  --detector-id $DETECTOR_ID \
  --finding-types "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom"

Via Console:

GuardDuty → Settings → Sample findings
Generate sample findings

Step 7.2: Monitor Lambda Execution

Lambda Console → security-incident-responder
Monitor tab → View logs in CloudWatch
Check latest log stream

You should see logs like:

Processing finding: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom (Severity: HIGH)
Instance i-xxxxx successfully isolated
Security alert sent successfully

Step 7.3: Check Email

Check your email - you should receive alert:

🚨 SECURITY INCIDENT DETECTED 🚨
Severity: HIGH
Type: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom

✅ Day 1 Complete! Basic automated response working.

📅 DAY 2: Advanced Workflows & Compliance (8 Hours)

📝 PHASE 8: Setup CloudTrail (1 Hour)

CloudTrail logs all API calls for audit trail.

Step 8.1: Enable CloudTrail

Via Console:

CloudTrail Console → Create trail
Trail name: security-audit-trail
Storage location: Create new S3 bucket
Bucket name: security-audit-logs-ACCOUNT-ID
Log file SSE-KMS encryption: Enabled(use default AWS managed key)
CloudWatch Logs: Enabled
- Log group name: /aws/cloudtrail/security-audit
Click Next
Event type:
- Management events: ✓ Read and ✓ Write
- Data events: Skip for now (save cost)
Click Next → Create trail

✅ Verification: CloudTrail → Trails → Status "Logging"

📝 PHASE 9: Setup AWS Config (1.5 Hours)

AWS Config monitors resource configurations for compliance.

Step 9.1: Enable Config

Via Console:

AWS Config Console →Get started
Resource types to record:
- Select:Record all resources
- Include global resources: Yes
Amazon S3 bucket:
- Create a bucket: Yes
- Bucket name: config-audit-ACCOUNT-ID
SNS topic:
- Stream configuration changes: Yes
- Select existing SNS topic: security-incident-alerts
AWS Config role:
- Create AWS Config service-linked role: Yes
Click Next

Step 9.2: Add Compliance Rules

Add these managed rules:

iam-password-policy
- Ensures strong password policy
root-account-mfa-enabled
- Ensures root account has MFA
ec2-instance-managed-by-systems-manager
- Ensures EC2 instances are managed
encrypted-volumes
- Ensures EBS volumes are encrypted
s3-bucket-public-read-prohibited
- Ensures S3 buckets not publicly readable
cloudtrail-enabled
- Ensures CloudTrail is enabled

To add each rule:
1.Config → Rules → Add rule
2.Search for rule name
3.Click rule → Next
4.Leave defaults → Save

✅ Verification: Config → Rules → 6 rules active, evaluating compliance

📝 PHASE 10: Build Step Functions Workflow (2 Hours)

Step Functions untuk complex multi-step response workflows.

Step 10.1: Create Step Functions State Machine

Via Console:

Step Functions Console → Create state machine
Choose template: Blank
Type: Standard
Name: security-incident-workflow
Definition (paste this JSON):

{
  "Comment": "Advanced Security Incident Response Workflow",
  "StartAt": "ParseIncident",
  "States": {
    "ParseIncident": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "security-incident-responder",
        "Payload.$": "$"
      },
      "ResultPath": "$.responseResult",
      "Next": "CheckSeverity",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "NotifyFailure"
      }]
    },
    "CheckSeverity": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.detail.findings[0].Severity.Label",
          "StringEquals": "CRITICAL",
          "Next": "CriticalIncidentPath"
        },
        {
          "Variable": "$.detail.findings[0].Severity.Label",
          "StringEquals": "HIGH",
          "Next": "HighIncidentPath"
        }
      ],
      "Default": "LogAndExit"
    },
    "CriticalIncidentPath": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "CreateTicket",
          "States": {
            "CreateTicket": {
              "Type": "Task",
              "Resource": "arn:aws:states:::sns:publish",
              "Parameters": {
                "TopicArn": "arn:aws:sns:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:security-incident-alerts",
                "Subject": "CRITICAL Security Incident - Immediate Action Required",
                "Message.$": "$.detail.findings[0].Description"
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "WaitForSOC",
          "States": {
            "WaitForSOC": {
              "Type": "Wait",
              "Seconds": 300,
              "Next": "CheckIfResolved"
            },
            "CheckIfResolved": {
              "Type": "Task",
              "Resource": "arn:aws:states:::lambda:invoke",
              "Parameters": {
                "FunctionName": "check-incident-status",
                "Payload": {
                  "findingId.$": "$.detail.findings[0].Id"
                }
              },
              "End": true
            }
          }
        }
      ],
      "Next": "LogSuccess"
    },
    "HighIncidentPath": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:security-incident-alerts",
        "Subject": "HIGH Security Incident Detected",
        "Message.$": "$.detail.findings[0].Description"
      },
      "Next": "LogSuccess"
    },
    "LogSuccess": {
      "Type": "Succeed"
    },
    "LogAndExit": {
      "Type": "Succeed"
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT:security-incident-alerts",
        "Subject": "Security Response Workflow Failed",
        "Message": "The security incident response workflow encountered an error."
      },
      "Next": "FailState"
    },
    "FailState": {
      "Type": "Fail",
      "Error": "WorkflowFailed",
      "Cause": "Security response workflow failed to complete"
    }
  }
}

Replace REGION and ACCOUNT with your Value
Click Next
Permissions:
- Create new role: Yes
- Role name: StepFunctions-SecurityWorkflow-Role
  1. Click Create state machine

Step 10.2: Update EventBridge to Trigger Step Functions

EventBridge → Rules → Create new rule
Name: critical-findings-to-stepfunctions
Event pattern:

{
  "source": ["aws.guardduty", "aws.securityhub"],
  "detail": {
    "findings": {
      "Severity": {
        "Label": ["CRITICAL"]
      }
    }
  }
}

Target: Step Functions state machine
State machine: security-incident-workflow
Click Create

📝 PHASE 11: Advanced Response Functions (2 Hours)

Create additional Lambda functions untuk specific scenarios.

Step 11.1: Create Forensics Collection Function

Via Console:

Lambda → Create function
Name: forensics-collector
Runtime: Python 3.11
Role: SecurityIncidentResponseRole

Code

import boto3
import json

config = boto3.client('config')
iam = boto3.client('iam')
ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    """Check compliance status and generate report"""

    compliance_report = {
        'timestamp': event['time'],
        'checks': {}
    }

    # 1. Check Config rules compliance
    config_rules = config.describe_compliance_by_config_rule()

    non_compliant = []
    for rule in config_rules['ComplianceByConfigRules']:
        if rule['Compliance']['ComplianceType'] == 'NON_COMPLIANT':
            non_compliant.append(rule['ConfigRuleName'])

    compliance_report['checks']['config_rules'] = {
        'total': len(config_rules['ComplianceByConfigRules']),
        'non_compliant': non_compliant
    }

    # 2. Check IAM password policy
    try:
        password_policy = iam.get_account_password_policy()
        compliance_report['checks']['iam_password_policy'] = {
            'exists': True,
            'minimum_length': password_policy['PasswordPolicy'].get('MinimumPasswordLength', 0),
            'require_symbols': password_policy['PasswordPolicy'].get('RequireSymbols', False),
            'require_numbers': password_policy['PasswordPolicy'].get('RequireNumbers', False)
        }
    except iam.exceptions.NoSuchEntityException:
        compliance_report['checks']['iam_password_policy'] = {
            'exists': False,
            'compliant': False
        }

    # 3. Check MFA on root account
    try:
        summary = iam.get_account_summary()
        compliance_report['checks']['root_mfa'] = {
            'enabled': summary['SummaryMap'].get('AccountMFAEnabled', 0) == 1
        }
    except Exception as e:
        compliance_report['checks']['root_mfa'] = {'error': str(e)}

    # 4. Check for unencrypted EBS volumes
    volumes = ec2.describe_volumes()
    unencrypted = [v['VolumeId'] for v in volumes['Volumes'] if not v.get('Encrypted', False)]

    compliance_report['checks']['ebs_encryption'] = {
        'total_volumes': len(volumes['Volumes']),
        'unencrypted': unencrypted,
        'compliant': len(unencrypted) == 0
    }

    # 5. Check for public S3 buckets
    s3 = boto3.client('s3')
    buckets = s3.list_buckets()['Buckets']

    public_buckets = []
    for bucket in buckets:
        try:
            acl = s3.get_bucket_acl(Bucket=bucket['Name'])
            for grant in acl['Grants']:
                if grant['Grantee'].get('Type') == 'Group' and 'AllUsers' in grant['Grantee'].get('URI', ''):
                    public_buckets.append(bucket['Name'])
                    break
        except:
            pass

    compliance_report['checks']['s3_public_access'] = {
        'public_buckets': public_buckets,
        'compliant': len(public_buckets) == 0
    }

    # Calculate overall compliance score
    compliant_checks = sum([
        1 for check in compliance_report['checks'].values()
        if isinstance(check, dict) and check.get('compliant', False)
    ])
    total_checks = len(compliance_report['checks'])

    compliance_report['overall_score'] = (compliant_checks / total_checks * 100) if total_checks > 0 else 0

    return {
        'statusCode': 200,
        'body': json.dumps(compliance_report, default=str)
    }

📝 PHASE 12: Create CloudWatch Dashboard (1.5 Hours)

Step 12.1: Create Custom Dashboard

Via Console:

CloudWatch Console → Dashboards → Create dashboard
Dashboard name: SecurityIncidentResponse
Click Create dashboard

Step 12.2: Add Widgets

Widget 1: GuardDuty Findings Count

Add widget → Number
Metrics → GuardDuty → Select:
- AWS/GuardDuty → FindingCount
Statistic: Sum
Period: 1 hour
Widget title: "GuardDuty Findings (Last Hour)"

Widget 2: Lambda Invocation Count

Add widget → Line
Metrics → Lambda → By Function Name
Select: security-incident-responder → Invocations
Period: 5 minutes
Title: "Auto-Response Invocations"

Widget 3: Lambda Errors

Add widget → Line
Select: security-incident-responder → Errors
Period: 5 minutes
Title: "Response Function Errors"

Widget 4: Security Hub Findings by Severity

Add widget → Stacked area
Custom namespace: Add metric → Security Hub
Title: "Security Hub Findings by Severity"

Widget 5: Recent CloudTrail Events

Add widget → Logs table
Log group: /aws/cloudtrail/security-audit
Query:

  fields @timestamp, userIdentity.principalId, eventName, sourceIPAddress
| filter errorCode like /Unauthorized/ or errorCode like /AccessDenied/
| sort @timestamp desc
| limit 20

Title: "Recent Unauthorized Access Attempts"

Widget 6: Step Functions Executions (optional)

Add widget → Number
Metrics → Step Functions → ExecutionsStarted
State machine: security-incident-workflow
Title: "Workflow Executions"
Click Save dashboard

📝 Phase 13: Create Runbook via AWS Console

Option 1: Save in S3 (Recommended)

Step 1: Create S3 Bucket

S3 Console → Create bucket
Bucket name: security-runbooks-YOUR_ACCOUNT_ID
Region: Asia Pacific (Sydney) ap-southeast-2
Block Public Access: ✅** Keep all checked (default)**
Bucket Versioning: Enable (optional)
Encryption: Enable (SSE-S3)
Click Create bucket

Step 2: Create Runbook File in your computer

Open Text editor or Notepad
Copy-paste this text:

Security Incident Response Runbook
===================================

System Overview
---------------
- Primary Function: Automated security incident detection and response
- Detection: AWS GuardDuty
- Response: Lambda automated actions
- Alerting: SNS email notifications
- Region: ap-southeast-2 (Sydney)

Quick Response Guide
--------------------

HIGH SEVERITY INCIDENTS:
1. Check email alert from SNS topic: security-incident-alerts
2. Review CloudWatch Dashboard: SecurityIncidentResponse
3. Check Lambda logs: /aws/lambda/security-incident-responder
4. Verify automated actions taken (isolate EC2, rotate IAM, etc)

MANUAL INTERVENTION REQUIRED:
If automated response failed:
1. Check error message in email alert
2. Review CloudWatch logs for detailed error
3. Manually execute remediation actions
4. Update this runbook with learnings

Automated Response Actions
--------------------------

UnauthorizedAccess:EC2 Incidents:
- Action: Isolate EC2 instance with new security group
- Action: Create forensic EBS snapshots
- Action: Send alert to SOC team

IAM Credential Compromise:
- Action: Deactivate all IAM access keys
- Action: Force password reset
- Action: Apply emergency deny-all policy

Crypto Mining Detection:
- Action: Stop EC2 instance immediately
- Action: Create forensic snapshots
- Action: Send critical alert

Malicious IP Detection:
- Action: Block source IP via Network ACL
- Action: Log incident details

System Components
-----------------

GuardDuty:
- Detector ID: 52cd95edfb049e79eada14b660d424b7
- Status: Check via Console
- Sample findings: Can generate for testing

Lambda Functions:
- security-incident-responder: Main response function
- Memory: 512 MB
- Timeout: 5 minutes
- Region: ap-southeast-2

EventBridge Rules:
- guardduty-high-severity-findings: Routes HIGH/CRITICAL findings
- Trigger: Lambda function

SNS Topics:
- security-incident-alerts: Email notifications
- Subscribers: [Your email]

CloudWatch Dashboard:
- Name: SecurityIncidentResponse
- Widgets: Lambda metrics, SNS, Logs

Step Functions:
- security-incident-workflow: Complex incident workflows
- State Machine ARN: Check Console

Contact Information
-------------------
Security Team Email: security@company.com (ganti dengan email real)
On-Call Phone: +62-xxx-xxx-xxxx (ganti dengan nomor real)
Escalation: [Manager name and contact]

AWS Account ID: YOUR_ACCOUNT_ID
Primary Region: ap-southeast-2 (Sydney)

Monthly Testing Procedures
---------------------------

Test Schedule: First Monday of each month, 10:00 AM

Test 1 - Generate Sample Finding:
1. Open GuardDuty Console
2. Settings → Sample findings
3. Click "Generate sample findings"
4. Wait 2 minutes

Test 2 - Verify Automated Response:
1. Check email for alert (should arrive within 2 min)
2. Check CloudWatch Dashboard for Lambda invocation
3. Review Lambda logs for processing details
4. Verify no errors in execution

Test 3 - Dashboard Validation:
1. Open CloudWatch Dashboard: SecurityIncidentResponse
2. Verify all widgets showing data
3. Check Lambda Invocations widget increased
4. Review Recent Logs widget for new entries

Recovery Procedures
-------------------

Rollback Isolated EC2 Instance:
1. EC2 Console → Instances
2. Find instance with isolation security group
3. Actions → Security → Change security groups
4. Replace with original security group
5. Document incident resolution

Re-enable IAM User After False Positive:
1. IAM Console → Users → Select user
2. Security credentials → Activate access keys
3. Remove "EmergencyDenyAll" policy
4. Reset password (user will change on next login)
5. Notify user via approved channel

Unblock IP from Network ACL:
1. VPC Console → Network ACLs
2. Find NACL with deny rule for IP
3. Inbound/Outbound rules → Delete rule number 1
4. Document reason for unblock

Service Costs (Monthly Estimates)
----------------------------------
GuardDuty: $4.50/month
Security Hub: $30/month (can disable to save)
Lambda: $5/month (only when invoked)
SNS: $1/month
CloudWatch: $10-20/month
CloudTrail: $2-5/month
Config: $20/month (can stop to save)
Step Functions: $10/month

TOTAL RUNNING: ~$82-95/month
TOTAL STOPPED: ~$0/month (when services suspended)

To Save Costs When Not Using:
- Suspend GuardDuty
- Disable Security Hub
- Stop Config recorder
- Stop CloudTrail logging
- Disable EventBridge rules

Services Resume Time: 2-5 minutes

Known Issues & Solutions
------------------------

Issue: Lambda timeout when isolating instance
Solution: Increase timeout to 5 minutes (already configured)

Issue: No email alerts received
Solution: 
1. Check SNS subscription is "Confirmed"
2. Check spam folder
3. Verify SNS_TOPIC_ARN environment variable in Lambda

Issue: GuardDuty metrics not in dashboard
Solution: Normal if no real findings yet. Use custom metrics from Lambda instead.

Issue: Step Functions execution failed
Solution: Check IAM role has correct permissions for Lambda and SNS

Change Log
----------
2024-12-17: Initial system setup
2024-12-17: Day 1 completed - Core automation working
2024-12-17: Day 2 completed - Dashboard and documentation

Next Review Date: [30 days from today]

Notes
-----
- This is a learning/demo implementation
- For production, add AWS WAF for additional protection
- Consider integrating with SIEM (Splunk, Elastic)
- Enable AWS Shield for DDoS protection
- Add MFA for all IAM users
- Implement least privilege IAM policies
- Regular security audits recommended

Save as: runbook.txt

Step 3: Upload ke S3

S3 Console → Buckets → security-runbooks-YOUR_ACCOUNT_ID
Click "Upload"
Add files → Select runbook.txt
Scroll down → Click "Upload"
Done! ✅

📝 PHASE 14: Final Testing & Validation (1 Hour)

Test 1: EC2 Compromise Simulation

Step 1: Generate Sample Finding

GuardDuty Console → Settings
Scroll to Sample findings
Click "Generate sample findings"
Wait 30-60 seconds

Step 2: Check Lambda Triggered

Lambda Console → Functions → security-incident-responder
Monitor tab
Metrics section:
- Check "Invocations" graph - should show spike
- Check "Duration" - should be < 5 seconds
- Check "Errors" - should be 0

Step 3: Check CloudWatch Logs

Lambda → Monitor tab
Click "View CloudWatch logs"
Click latest log stream (top)
Look for:

Processing finding: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom (Severity: HIGH)
   Found instance ID: i-99999999
   Error isolating instance... (EXPECTED - test instance)
   Security alert sent successfully

Step 4: Verify Email Alert

Check your email (in 2 minutes)
Subject: 🚨 SECURITY ALERT: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom
Body should show:

Severity: HIGH
   Type: UnauthorizedAccess:EC2/...
   AUTOMATED ACTIONS TAKEN:
   ✓ isolate_instance: failed - Error: Instance not found

Step 5: Check Dashboard

CloudWatch Console → Dashboards → SecurityIncidentResponse
Verify widgets updated:
- Lambda Invocations: +1
- Lambda Duration: shows recent execution
- Recent Logs: shows new entry

✅ Test 1 PASSED if all 5 checks OK!

Test 2: IAM Credential Compromise

Step 1: Generate IAM Finding

GuardDuty Console → Settings → Sample findings
Click "Generate sample findings"
Or specific:
- GuardDuty → Findings
- Filter: Type contains "IAM"
- Look for: UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration

Step 2: Verify Lambda Processing

Lambda Console → security-incident-responder → Monitor
CloudWatch logs (latest stream)
Look for:

Processing finding: UnauthorizedAccess:IAMUser/...
   Rotating credentials for user: test-user
   Error: User not found (EXPECTED - test user)
   Security alert sent successfully

Step 3: Check Email

Email should show:

Type: UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration
AUTOMATED ACTIONS TAKEN:
✓ rotate_credentials: failed - Error: User not found

✅ Test 2 PASSED if Lambda executed + email received!

Step 14.2: Performance Validation via Console

Check Dashboard Metrics:

CloudWatch Console → Dashboards → SecurityIncidentResponse
Verify each widget:

Widget: Lambda Invocations
✅ Shows 3+ invocations (from 3 tests)
✅ Recent activity visible

Widget: Lambda Errors
✅ Should show 0 errors
✅ (Failures like "Instance not found" are logged but not Lambda errors)

Widget: Lambda Duration
✅ Average < 5 seconds
✅ No timeouts

Widget: SNS Messages Published
✅ Shows 3+ messages
✅ Matches number of tests

Widget: CloudWatch Logs
✅ Shows recent log entries
✅ "Security alert sent successfully" appears

Check Lambda Performance Details:

Lambda Console → security-incident-responder → Monitor
Metrics tab:
- Duration: Average < 5000ms ✅
- Concurrent executions: < 10 ✅
- Throttles: 0 ✅
- Error rate: 0% ✅

Configuration tab → General configuration:
- Memory: 512 MB
- Timeout: 300 seconds (5 min)
- ✅ Sufficient for our use case

Step 14.3: Compliance Validation

Option 1: Check AWS Config Compliance (Simple)

AWS Config Console → Rules
Check compliance status:

Rule Name	status	Expected
iam-password-policy	✅Compliant	Strong Password Policy
root-account-mfa-enabled	✅Compliant	Root Has MFA
encrypted-volumes	⚠️ May vary	EBS encryption
Cloudtrail-enabled	✅Compliant	CloudTrail active

Overall compliance score: Should be > 70%

🏆 Resume Bullet Points
After completing this project:

"Built automated security incident response system processing 1000+ findings/month with < 2 minute response time"
"Implemented cloud-native SOAR (Security Orchestration, Automation & Response) using AWS Lambda, Step Functions, reducing manual intervention by 85%"
"Designed multi-layered threat detection using GuardDuty, Security Hub, and Config, achieving 95% threat detection accuracy"
"Architected forensics collection pipeline with automated evidence preservation compliant with SOC2 and ISO 27001"
"Reduced security incident response time from 2 hours (manual) to 2 minutes (automated), saving $150K annually in labor costs"

Congratulations! You've built a production-grade automated security incident response system! 🎉

AWS Proton: Microservices Infrastructure Automation

Daffa Rabbani — Tue, 25 Mar 2025 23:29:06 +0000

Table of contents

What is AWS Proton?
AWS Proton & AWS Cloudformation is same?
How AWS Proton Works?
conclusion

Hey folks 👋,

maybe you just found out that AWS has released one of the new services again in 2020 which even I forgot about the new update :) AWS gave the name of the service called AWS Proton. maybe this proton is similar to the Cloudformation model infrastructure as code.
in this post I will introduce you to AWS Proton and the difference with cloudformation. 🌟

What is AWS Proton? 🤔

✅ AWS Proton is a powerful automation solution that simplifies infrastructure management and application deployment for both administrators and developers.

✅ For administrators, AWS Proton provides a comprehensive framework to create and maintain service templates containing complete application stacks.

✅ For developers, AWS Proton offers a self-service interface where they can easily choose pre-defined service templates that fit their project needs.

What is the difference between AWS Proton & AWS Cloudformation? ⚙

AWS CloudFormation :

✅ A general-purpose infrastructure as code (IaC) service

✅ Allows you to define and provision infrastructure resources using templates

✅ Supports a wide range of AWS resources and can be used for almost any infrastructure setup

✅ Primarily focused on infrastructure provisioning

✅ Works at a lower-level of abstraction, giving developers more granular control

AWS Proton :

✅ A higher-level, more opinionated service specifically for container and serverless applications

✅ Provides standardized, pre-approved templates for entire application stacks

✅ Includes built-in CI/CD pipeline configuration

✅ Focuses on creating a self-service platform for developers

✅ Automates not just infrastructure provisioning, but also deployment pipelines

How AWS Proton Works ? 👨‍💻

When you choose an environment template with AWS Proton, you provide values for the relevant input parameters.
AWS Proton prepares your environment based on the environment template and parameter settings.
When you use AWS Proton to pick a service template, you supply values for the relevant input parameters. You also pick an environment in which to deploy your application or service.
To supply your service, AWS Proton utilizes the service template as well as the values of your service and the environment parameters you have defined.

You choose the input parameters to modify your template for reuse and numerous use cases, applications, or services.

To make this work, you generate environment or service template bundles and submit them to the appropriate registered environment or service templates.

Use Cases

E-Commerce Platform Microservices

Detailed Scenario :
An e-commerce platform with multiple microservices requires a robust, scalable, and secure infrastructure that can handle varying loads and ensure consistent deployment across different services.

Specific Challenges

Infrastructure Inconsistency: Each microservice (Product, Cart, Payment, User) was previously deployed with slight variations in configuration
Security Complexities: Ensuring uniform security protocols across different services
Deployment Overhead: Manual configuration and deployment processes were time-consuming

AWS Proton Implementation

Standardized Templates:
- Create a base template for all e-commerce microservices
- Define consistent networking, security groups, and compute resources
- Include standard monitoring and logging configurations

Deployment Workflow:

Platform team creates a master template for e-commerce services
Developers select and slightly customize the template
- Configure service-specific parameters
- Set scaling requirements
Proton automatically provisions infrastructure
- Consistent network configuration
- Uniform security settings
- Standardized monitoring

Benefits Realized:

40% reduction in deployment time
Improved security consistency
Easier onboarding for new services
Simplified infrastructure management

2. Digital Banking Application

Comprehensive Microservices Ecosystem

A digital banking platform requires extreme reliability, security, and compliance across multiple critical services.

Complex Requirements

Regulatory Compliance: Must meet strict financial industry regulations
High-Security Demands: Protection of sensitive financial data
Performance Critical: Low-latency transactions
Scalability Needs: Handle sudden traffic spikes

AWS Proton Solution

Compliance-Driven Templates:
- Embed security best practices directly in infrastructure templates
- Automatic encryption configurations
- Network isolation between services
- Compliance checkpoints in deployment pipeline

Advanced Implementation:

Create separate environment templates for:
- Development
- Staging
- Production
Implement service-specific security groups
Automate compliance checks during deployment
Enable automatic scaling based on transaction volumes

Key Advantages:

Rapid compliance verification
Consistent security posture
Automated risk mitigation
Faster time-to-market for new financial services

Conclusion 📃

AWS Proton emerges as an innovative cloud infrastructure automation solution that transforms how organizations manage and deploy serverless and container-based applications, providing administrators with a powerful framework for creating standardized templates and developers with a self-service interface to streamline application deployments.

Unlike the more generic CloudFormation, Proton takes a more opinionated approach with features like version management, automatic CI/CD pipeline configuration, and one-click infrastructure updates, allowing businesses to create more efficient, consistent, and scalable cloud deployment strategies.

By bridging the gap between infrastructure management and application development, AWS Proton enables technology teams to optimize development processes, reduce manual configuration overhead, and ensure alignment with organizational best practices, ultimately changing the way modern cloud-native applications are designed, provisioned, and maintained.

Implementing Performance optimization with AWS Aurora PostgreSQL/MySQL for high-throughput workloads 👨‍💻⚙

Daffa Rabbani — Sun, 16 Mar 2025 03:15:54 +0000

Amazon Aurora is AWS’s cloud-native relational database, built for high performance and high availability while staying compatible with MySQL and PostgreSQL. It’s faster and more efficient than standard deployments, but if you’re handling high-throughput workloads, a few extra optimizations can make a big difference.

In this guide, we’ll walk through everything you need to know to fine-tune Aurora for demanding use cases. From identifying performance bottlenecks to optimizing queries, leveraging Aurora’s advanced features, and setting up proper monitoring—this article has got you covered!

🔍 Understanding Aurora’s Architecture

Before jumping into optimizations, let’s take a quick look at what makes Aurora different:

Distributed Storage – Aurora automatically replicates data across multiple Availability Zones (AZs) in an AWS Region. This separation of compute and storage allows for independent scaling and improved resilience.
Fault-Tolerant Storage System – Your data is stored in six copies across three AZs, ensuring durability and availability.
Efficient Write Operations – Aurora handles replication asynchronously, allowing high-throughput performance even for write-heavy applications.

With this foundation in mind, let’s dive into the best ways to optimize your Aurora database for peak performance! 🚀

Key Factors Affecting Performance

Instance Types and Sizing

Aurora performance is significantly influenced by the instance class you choose. For high-throughput workloads, consider:

Memory-optimized instance types (r5, r6g, r6i) for workloads with large working sets
The latest generation instances which typically offer better performance
Instance size appropriate for your workload's CPU and memory requirements

Network Bandwidth

Network bandwidth can become a bottleneck for high-throughput workloads, especially with larger instance types. Ensure you're using instance types with sufficient network bandwidth for your workload requirements.

Hands-On Lab: Setting Up an Optimized Aurora Environment

Let's start with creating an Aurora environment optimized for high throughput.

Step 1: Create an Aurora Cluster with Appropriate Configuration

aws rds create-db-cluster \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --engine-version 15.3 \
    --master-username admin \
    --master-user-password YourStrongPassword123! \
    --db-subnet-group-name your-subnet-group \
    --vpc-security-group-ids sg-0123456789abcdef \
    --backup-retention-period 7 \
    --preferred-backup-window 07:00-09:00 \
    --preferred-maintenance-window sat:10:00-sat:12:00 \
    --db-cluster-parameter-group-name high-throughput-params \
    --storage-encrypted \
    --region us-east-1

Step 2: Create Instance(s) with Appropriate Sizing

aws rds create-db-instance \
    --db-instance-identifier high-throughput-writer \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.4xlarge \
    --region us-east-1

For read scaling:

aws rds create-db-instance \
    --db-instance-identifier high-throughput-reader-1 \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.2xlarge \
    --region us-east-1

Step 3: Configure Parameter Groups

Create a custom parameter group:

aws rds create-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --db-parameter-group-family aurora-postgresql15 \
    --description "Parameters optimized for high throughput" \
    --region us-east-1

Modify parameters for PostgreSQL:

aws rds modify-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --parameters "ParameterName=max_connections,ParameterValue=5000,ApplyMethod=pending-reboot" \
    --region us-east-1

aws rds modify-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --parameters "ParameterName=shared_buffers,ParameterValue={DBInstanceClassMemory/32768},ApplyMethod=pending-reboot" \
    --region us-east-1

Schema Optimization Techniques

Efficient Table Design

Proper table design is fundamental to database performance. For high-throughput workloads:

Choose appropriate data types (e.g., use INT instead of VARCHAR for numeric values)
Normalize tables appropriately, but consider strategic denormalization for performance-critical queries
Implement table partitioning for large tables

Let's create an example of a partitioned table:

-- For PostgreSQL
CREATE TABLE orders (
    order_id BIGINT NOT NULL,
    customer_id INT NOT NULL,
    order_date DATE NOT NULL,
    total_amount DECIMAL(10,2),
    status VARCHAR(20)
) PARTITION BY RANGE (order_date);

-- Create partitions by month
CREATE TABLE orders_202501 PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE orders_202502 PARTITION OF orders
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

CREATE TABLE orders_202503 PARTITION OF orders
    FOR VALUES FROM ('2025-03-01') TO ('2025-04-01');

Indexing Strategy

Proper indexing is critical for high-throughput workloads:

Create indexes on columns frequently used in WHERE, JOIN, and ORDER BY clauses
Consider composite indexes for queries filtering on multiple columns
Be cautious about over-indexing, as indexes consume space and slow down writes

Example of creating efficient indexes:

-- Primary key
ALTER TABLE orders ADD PRIMARY KEY (order_id, order_date);

-- Index for frequent queries filtering by customer and date range
CREATE INDEX idx_orders_customer_date ON orders (customer_id, order_date);

-- Index for status filtering
CREATE INDEX idx_orders_status ON orders (status);

For high write throughput, consider using INCLUDE in PostgreSQL indexes to reduce the need for index-only scans:

CREATE INDEX idx_orders_date_include_status ON orders (order_date) INCLUDE (status);

Query Optimization

Analyzing and Improving Slow Queries

Let's look at how to identify and optimize slow queries:

Enable Performance Insights to track database load and identify top queries
Use the PostgreSQL/MySQL EXPLAIN command to analyze query execution plans

-- Enable query logging for slow queries (PostgreSQL)
ALTER SYSTEM SET log_min_duration_statement = '1000';  -- Log queries taking more than 1 second
ALTER SYSTEM SET log_statement = 'none';
SELECT pg_reload_conf();

-- Analyze a query using EXPLAIN ANALYZE
EXPLAIN ANALYZE
SELECT o.order_id, o.order_date, o.total_amount, c.customer_name
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE o.order_date BETWEEN '2025-01-01' AND '2025-01-31'
AND o.status = 'COMPLETED'
ORDER BY o.total_amount DESC
LIMIT 100;

Query Rewriting Techniques

Some common query rewriting techniques for high-throughput environments:

Use EXISTS instead of IN for checking existence:

-- Instead of:
SELECT * FROM orders WHERE customer_id IN (SELECT customer_id FROM premium_customers);

-- Use:
SELECT * FROM orders o WHERE EXISTS (
    SELECT 1 FROM premium_customers p WHERE p.customer_id = o.customer_id
);

Avoid functions on indexed columns:

-- Instead of:
SELECT * FROM orders WHERE EXTRACT(MONTH FROM order_date) = 3;

-- Use:
SELECT * FROM orders WHERE order_date BETWEEN '2025-03-01' AND '2025-03-31';

Use UNION ALL instead of UNION when duplicates are acceptable:

SELECT * FROM orders_202501 WHERE status = 'COMPLETED'
UNION ALL
SELECT * FROM orders_202502 WHERE status = 'COMPLETED';

Connection Management

Aurora connection management is crucial for high-throughput applications:

Implement connection pooling with tools like PgBouncer, ProxySQL, or AWS RDS Proxy
Monitor and tune max_connections parameter based on workload

Let's set up RDS Proxy:

aws rds create-db-proxy \
    --db-proxy-name aurora-high-throughput-proxy \
    --engine-family POSTGRESQL \
    --auth '{"AuthScheme":"SECRETS","SecretArn":"arn:aws:secretsmanager:us-east-1:123456789012:secret:aurora-credentials","IAMAuth":"DISABLED"}' \
    --role-arn arn:aws:iam::123456789012:role/rds-proxy-role \
    --vpc-subnet-ids subnet-0123456789abcdef0 subnet-0123456789abcdef1 \
    --vpc-security-group-ids sg-0123456789abcdef \
    --require-tls \
    --idle-client-timeout 1800 \
    --debug-logging \
    --region us-east-1

aws rds register-db-proxy-targets \
    --db-proxy-name aurora-high-throughput-proxy \
    --db-cluster-identifiers high-throughput-demo \
    --region us-east-1

Leveraging Aurora-Specific Features

Read Scaling with Aurora Replicas

For high-throughput read workloads, distribute reads across multiple Aurora Replicas:

import psycopg2
import random

# List of reader endpoints
reader_endpoints = [
    "high-throughput-reader-1.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com",
    "high-throughput-reader-2.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com"
]

def get_read_connection():
    # Randomly select a reader endpoint
    endpoint = random.choice(reader_endpoints)

    conn = psycopg2.connect(
        host=endpoint,
        port=5432,
        database="mydatabase",
        user="admin",
        password="YourStrongPassword123!"
    )
    return conn

def get_write_connection():
    # Write connection always goes to the primary instance
    conn = psycopg2.connect(
        host="high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com",
        port=5432,
        database="mydatabase",
        user="admin",
        password="YourStrongPassword123!"
    )
    return conn

Aurora Parallel Query

Enable Parallel Query for analytical workloads:

aws rds modify-db-cluster \
    --db-cluster-identifier high-throughput-demo \
    --enable-http-endpoint \
    --region us-east-1

In PostgreSQL, configure a session to use parallel query:

-- For PostgreSQL
SET aurora.parallel_query = ON;

-- Check if a query uses parallel query
EXPLAIN
SELECT customer_id, SUM(total_amount)
FROM orders
WHERE order_date BETWEEN '2025-01-01' AND '2025-03-31'
GROUP BY customer_id;

Monitoring and Continuous Optimization

Setting Up Comprehensive Monitoring

Enable Enhanced Monitoring and Performance Insights:

aws rds modify-db-instance \
    --db-instance-identifier high-throughput-writer \
    --monitoring-interval 5 \
    --monitoring-role-arn arn:aws:iam::123456789012:role/rds-monitoring-role \
    --enable-performance-insights \
    --performance-insights-retention-period 7 \
    --region us-east-1

Create CloudWatch alarms for critical metrics:

aws cloudwatch put-metric-alarm \
    --alarm-name HighCPUUtilization-Aurora \
    --alarm-description "Alarm when CPU exceeds 80%" \
    --metric-name CPUUtilization \
    --namespace AWS/RDS \
    --statistic Average \
    --period 300 \
    --threshold 80 \
    --comparison-operator GreaterThanThreshold \
    --dimensions Name=DBInstanceIdentifier,Value=high-throughput-writer \
    --evaluation-periods 3 \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:rds-alerts \
    --region us-east-1

Create custom metrics dashboard:

aws cloudwatch put-dashboard \
    --dashboard-name AuroraHighThroughputDashboard \
    --dashboard-body '{
        "widgets": [
            {
                "type": "metric",
                "x": 0,
                "y": 0,
                "width": 12,
                "height": 6,
                "properties": {
                    "metrics": [
                        [ "AWS/RDS", "CPUUtilization", "DBInstanceIdentifier", "high-throughput-writer" ]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": "Writer CPU"
                }
            },
            {
                "type": "metric",
                "x": 12,
                "y": 0,
                "width": 12,
                "height": 6,
                "properties": {
                    "metrics": [
                        [ "AWS/RDS", "DMLLatency", "DBInstanceIdentifier", "high-throughput-writer" ]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": "Write Latency"
                }
            }
        ]
    }' \
    --region us-east-1

Benchmarking Your Aurora Deployment

Use pgbench (PostgreSQL) or sysbench (MySQL) to establish performance baselines:

# For PostgreSQL
pgbench -h high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com \
    -U admin \
    -p 5432 \
    -d mydatabase \
    -c 20 \
    -j 4 \
    -T 60 \
    -P 10

Advanced Optimization Techniques

Memory Optimization

Fine-tune memory-related parameters:

-- For PostgreSQL
ALTER SYSTEM SET work_mem = '64MB';
ALTER SYSTEM SET maintenance_work_mem = '1GB';
ALTER SYSTEM SET effective_cache_size = '24GB';  -- Adjust based on instance memory
SELECT pg_reload_conf();

Vacuum and Analyze Operations

Regular vacuum and analyze operations are critical for PostgreSQL performance:

-- Set up autovacuum parameters
ALTER SYSTEM SET autovacuum_vacuum_scale_factor = 0.05;
ALTER SYSTEM SET autovacuum_analyze_scale_factor = 0.025;
ALTER SYSTEM SET autovacuum_vacuum_cost_limit = 2000;
SELECT pg_reload_conf();

-- Manual VACUUM ANALYZE for immediate effect
VACUUM ANALYZE orders;

Advanced Optimization Techniques

Using Aurora Global Database for Multi-Region Deployments

For global applications requiring low-latency reads across regions:

# Create a global database
aws rds create-global-cluster \
    --global-cluster-identifier global-high-throughput \
    --source-db-cluster-identifier arn:aws:rds:us-east-1:123456789012:cluster:high-throughput-demo \
    --region us-east-1

Add a secondary region:

aws rds create-db-cluster \
    --db-cluster-identifier high-throughput-demo-eu \
    --engine aurora-postgresql \
    --engine-version 15.3 \
    --db-subnet-group-name your-eu-subnet-group \
    --vpc-security-group-ids sg-0123456789abcdef \
    --global-cluster-identifier global-high-throughput \
    --region eu-west-1

Add instances to the secondary region:

aws rds create-db-instance \
    --db-instance-identifier high-throughput-reader-eu \
    --db-cluster-identifier high-throughput-demo-eu \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.2xlarge \
    --region eu-west-1

Optimizing for Specific Workload Types

Write-Intensive Workloads

For applications with heavy write loads:

Batch operations where possible:

-- Instead of individual inserts
INSERT INTO orders (order_id, customer_id, order_date, total_amount, status)
VALUES
    (1001, 101, '2025-03-15', 299.99, 'PENDING'),
    (1002, 102, '2025-03-15', 199.50, 'PENDING'),
    (1003, 103, '2025-03-15', 599.75, 'PENDING'),
    (1004, 104, '2025-03-15', 149.99, 'PENDING'),
    (1005, 105, '2025-03-15', 849.99, 'PENDING');

Reduce index overhead on write-heavy tables:

-- Temporarily disable triggers during bulk loads
ALTER TABLE orders DISABLE TRIGGER ALL;
-- Perform bulk insert
-- Then re-enable triggers
ALTER TABLE orders ENABLE TRIGGER ALL;

Consider UNLOGGED tables (PostgreSQL) for temporary data:

CREATE UNLOGGED TABLE temp_orders (
    order_id BIGINT NOT NULL,
    customer_id INT NOT NULL,
    order_date DATE NOT NULL,
    total_amount DECIMAL(10,2),
    status VARCHAR(20)
);

Read-Intensive Workloads

For read-heavy applications:

Implement materialized views for complex queries:

-- For PostgreSQL
CREATE MATERIALIZED VIEW monthly_sales AS
SELECT
    DATE_TRUNC('month', order_date) AS month,
    customer_id,
    SUM(total_amount) AS monthly_total,
    COUNT(*) AS order_count
FROM orders
GROUP BY DATE_TRUNC('month', order_date), customer_id;

-- Create index on the materialized view
CREATE INDEX idx_monthly_sales_customer ON monthly_sales (customer_id, month);

-- Refresh as needed
REFRESH MATERIALIZED VIEW monthly_sales;

Use Query Caching at the application level:

import redis
import json
import hashlib
import psycopg2

# Initialize Redis client
redis_client = redis.Redis(host='your-redis-host.cache.amazonaws.com', port=6379)

def get_cached_query_results(query, params=None, ttl=300):
    # Create a cache key based on the query and parameters
    cache_key = f"db:query:{hashlib.md5((query + str(params)).encode()).hexdigest()}"

    # Check if results exist in cache
    cached_result = redis_client.get(cache_key)
    if cached_result:
        return json.loads(cached_result)

    # If not in cache, execute query
    conn = get_read_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    columns = [desc[0] for desc in cursor.description]
    results = [dict(zip(columns, row)) for row in cursor.fetchall()]
    cursor.close()
    conn.close()

    # Store in cache with TTL
    redis_client.setex(cache_key, ttl, json.dumps(results))
    return results

Lab: Real-Time Performance Testing and Tuning

Let's create a comprehensive testing setup to optimize our Aurora deployment under load.

Step 1: Create Test Data

We'll use pgbench to initialize a test database and then extend it with our custom schema:

# Connect to your Aurora instance
psql "host=high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com port=5432 dbname=postgres user=admin password=YourStrongPassword123!"

# Create test database
CREATE DATABASE perf_test;
\c perf_test

# Initialize with pgbench schema (100 scaling factor ~ 10 million rows)
\! pgbench -i -s 100 -h high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com -p 5432 -U admin perf_test

Step 2: Create Custom Test Schema

-- Create orders table with partitioning
CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,
    customer_id INT NOT NULL,
    order_date TIMESTAMP NOT NULL,
    total_amount DECIMAL(10,2) NOT NULL,
    status VARCHAR(20) NOT NULL,
    items JSONB,
    shipping_address TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) PARTITION BY RANGE (order_date);

-- Create partitions
CREATE TABLE orders_q1_2025 PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

CREATE TABLE orders_q2_2025 PARTITION OF orders
    FOR VALUES FROM ('2025-04-01') TO ('2025-07-01');

-- Create appropriate indexes
CREATE INDEX idx_orders_customer_date ON orders (customer_id, order_date);
CREATE INDEX idx_orders_status ON orders (status);
CREATE INDEX idx_orders_items ON orders USING GIN (items);

-- Create customers table
CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    tier VARCHAR(20) NOT NULL,
    signup_date DATE NOT NULL,
    last_order_date TIMESTAMP,
    total_orders INT DEFAULT 0,
    total_spent DECIMAL(12,2) DEFAULT 0
);

CREATE INDEX idx_customers_tier ON customers (tier);
CREATE INDEX idx_customers_signup ON customers (signup_date);

Step 3: Generate Test Data

Create a Python script to generate realistic test data:

import psycopg2
import random
import json
from datetime import datetime, timedelta
import time

# Connect to database
conn = psycopg2.connect(
    host="high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com",
    port=5432,
    database="perf_test",
    user="admin",
    password="YourStrongPassword123!"
)
conn.autocommit = False
cursor = conn.cursor()

# Generate customers
customer_tiers = ["BRONZE", "SILVER", "GOLD", "PLATINUM"]
product_names = ["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard", "Mouse", "Tablet", "Camera", "Printer", "Speaker"]
statuses = ["PENDING", "PROCESSING", "SHIPPED", "DELIVERED", "RETURNED", "CANCELLED"]

# Insert customers
print("Generating customers...")
for i in range(1, 100001):
    name = f"Customer {i}"
    email = f"customer{i}@example.com"
    tier = random.choice(customer_tiers)
    signup_date = datetime(2024, 1, 1) + timedelta(days=random.randint(0, 365))

    cursor.execute(
        "INSERT INTO customers (name, email, tier, signup_date) VALUES (%s, %s, %s, %s)",
        (name, email, tier, signup_date)
    )

    if i % 1000 == 0:
        conn.commit()
        print(f"Inserted {i} customers")

conn.commit()
print("Finished generating customers")

# Insert orders
print("Generating orders...")
start_date = datetime(2025, 1, 1)
end_date = datetime(2025, 6, 30)
days_range = (end_date - start_date).days

for i in range(1, 500001):
    customer_id = random.randint(1, 100000)
    order_date = start_date + timedelta(days=random.randint(0, days_range))

    # Generate between 1 and 5 items for each order
    num_items = random.randint(1, 5)
    items = []
    total_amount = 0

    for _ in range(num_items):
        product = random.choice(product_names)
        price = round(random.uniform(9.99, 999.99), 2)
        quantity = random.randint(1, 3)
        item_total = price * quantity
        total_amount += item_total

        items.append({
            "product": product,
            "price": price,
            "quantity": quantity,
            "item_total": item_total
        })

    status = random.choice(statuses)
    shipping_address = f"{random.randint(100, 999)} Main St, City {random.randint(1, 100)}, State {random.randint(1, 50)}"

    # Insert order
    cursor.execute(
        "INSERT INTO orders (customer_id, order_date, total_amount, status, items, shipping_address) VALUES (%s, %s, %s, %s, %s, %s)",
        (customer_id, order_date, total_amount, status, json.dumps(items), shipping_address)
    )

    # Update customer stats (in real application, consider using triggers for this)
    cursor.execute(
        "UPDATE customers SET total_orders = total_orders + 1, total_spent = total_spent + %s, last_order_date = %s WHERE customer_id = %s",
        (total_amount, order_date, customer_id)
    )

    if i % 1000 == 0:
        conn.commit()
        print(f"Inserted {i} orders")

conn.commit()
print("Finished generating orders")

# Update statistics
print("Analyzing tables...")
cursor.execute("VACUUM ANALYZE customers")
cursor.execute("VACUUM ANALYZE orders")
print("Done!")

cursor.close()
conn.close()

Step 4: Create Test Workload Scripts

Create scripts to simulate real-world read and write patterns:

A. Read_workload.py

# read_workload.py - Simulates high-throughput read queries
import psycopg2
import random
import time
import concurrent.futures
from datetime import datetime, timedelta

# Read connection pool
read_endpoints = [
    "high-throughput-reader-1.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com",
    "high-throughput-reader-2.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com"
]

# Common queries representing typical workload
queries = [
    # Query 1: Get recent orders for a customer
    """
    SELECT o.order_id, o.order_date, o.total_amount, o.status
    FROM orders o
    WHERE o.customer_id = %s
    ORDER BY o.order_date DESC
    LIMIT 10
    """,

    # Query 2: Get orders in a date range with filtering
    """
    SELECT o.order_id, o.customer_id, o.order_date, o.total_amount, o.status
    FROM orders o
    WHERE o.order_date BETWEEN %s AND %s
    AND o.status = %s
    ORDER BY o.total_amount DESC
    LIMIT 100
    """,

    # Query 3: Get summary statistics
    """
    SELECT
        COUNT(*) as order_count,
        SUM(total_amount) as total_revenue,
        AVG(total_amount) as avg_order_value,
        MIN(total_amount) as min_order,
        MAX(total_amount) as max_order
    FROM orders
    WHERE order_date BETWEEN %s AND %s
    """,

    # Query 4: Complex aggregation with join
    """
    SELECT
        c.tier,
        DATE_TRUNC('month', o.order_date) as month,
        COUNT(DISTINCT c.customer_id) as unique_customers,
        COUNT(o.order_id) as order_count,
        SUM(o.total_amount) as total_revenue
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    WHERE o.order_date BETWEEN %s AND %s
    GROUP BY c.tier, DATE_TRUNC('month', o.order_date)
    ORDER BY month, tier
    """,

    # Query 5: JSONB filtering
    """
    SELECT o.order_id, o.total_amount
    FROM orders o
    WHERE o.items @> '[{"product": %s}]'
    LIMIT 50
    """
]

def get_connection():
    endpoint = random.choice(read_endpoints)
    return psycopg2.connect(
        host=endpoint,
        port=5432,
        database="perf_test",
        user="admin",
        password="YourStrongPassword123!"
    )

def execute_query(query_idx):
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Generate parameters based on query type
        if query_idx == 0:
            # Random customer ID
            params = (random.randint(1, 100000),)
        elif query_idx == 1:
            # Date range and status
            start_date = datetime(2025, 1, 1) + timedelta(days=random.randint(0, 150))
            end_date = start_date + timedelta(days=30)
            status = random.choice(["PENDING", "PROCESSING", "SHIPPED", "DELIVERED"])
            params = (start_date, end_date, status)
        elif query_idx == 2 or query_idx == 3:
            # Date range
            start_date = datetime(2025, 1, 1) + timedelta(days=random.randint(0, 150))
            end_date = start_date + timedelta(days=30)
            params = (start_date, end_date)
        elif query_idx == 4:
            # Product for JSONB query
            params = (random.choice(["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard"]),)

        cursor.execute(queries[query_idx], params)
        cursor.fetchall()  # Actually fetch the results

        end_time = time.time()
        return end_time - start_time
    except Exception as e:
        print(f"Query error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def run_workload(num_queries=1000, max_workers=20):
    print(f"Starting read workload simulation with {num_queries} queries...")
    query_times = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = []

    for _ in range(num_queries):
        # Select a random query type
        query_idx = random.randint(0, len(queries) - 1)
        futures.append(executor.submit(execute_query, query_idx))

    for future in concurrent.futures.as_completed(futures):
        exec_time = future.result()
        if exec_time is not None:
            query_times.append(exec_time)

# Calculate statistics
if query_times:
    avg_time = sum(query_times) / len(query_times)
    min_time = min(query_times)
    max_time = max(query_times)
    p95_time = sorted(query_times)[int(len(query_times) * 0.95)]

    print(f"Completed {len(query_times)} queries")
    print(f"Average query time: {avg_time:.4f} seconds")
    print(f"Min query time: {min_time:.4f} seconds")
    print(f"Max query time: {max_time:.4f} seconds")
    print(f"95th percentile query time: {p95_time:.4f} seconds")
else:
    print("No successful queries")
if name == "main":
run_workload(num_queries=5000, max_workers=50)

B. Write_workload.py

Create a write workload script:

# write_workload.py - Simulates high-throughput write operations
import psycopg2
import random
import time
import concurrent.futures
from datetime import datetime, timedelta
import json

# Write endpoint (primary instance)
write_endpoint = "high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com"

# Product names for generating random orders
product_names = ["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard", "Mouse", "Tablet", "Camera", "Printer", "Speaker"]
statuses = ["PENDING", "PROCESSING"]

def get_connection():
    return psycopg2.connect(
        host=write_endpoint,
        port=5432,
        database="perf_test",
        user="admin",
        password="YourStrongPassword123!"
    )

def execute_insert():
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Generate a random order
        customer_id = random.randint(1, 100000)
        order_date = datetime.now()

        # Generate between 1 and 5 items
        num_items = random.randint(1, 5)
        items = []
        total_amount = 0

        for _ in range(num_items):
            product = random.choice(product_names)
            price = round(random.uniform(9.99, 999.99), 2)
            quantity = random.randint(1, 3)
            item_total = price * quantity
            total_amount += item_total

            items.append({
                "product": product,
                "price": price,
                "quantity": quantity,
                "item_total": item_total
            })

        status = random.choice(statuses)
        shipping_address = f"{random.randint(100, 999)} Main St, City {random.randint(1, 100)}, State {random.randint(1, 50)}"

        # Insert order
        cursor.execute(
            "INSERT INTO orders (customer_id, order_date, total_amount, status, items, shipping_address) VALUES (%s, %s, %s, %s, %s, %s) RETURNING order_id",
            (customer_id, order_date, total_amount, status, json.dumps(items), shipping_address)
        )
        order_id = cursor.fetchone()[0]

        # Update customer stats
        cursor.execute(
            "UPDATE customers SET total_orders = total_orders + 1, total_spent = total_spent + %s, last_order_date = %s WHERE customer_id = %s",
            (total_amount, order_date, customer_id)
        )

        conn.commit()
        end_time = time.time()

        return {"order_id": order_id, "exec_time": end_time - start_time}
    except Exception as e:
        conn.rollback()
        print(f"Insert error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def execute_update():
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Get a random order ID
        cursor.execute("SELECT order_id FROM orders ORDER BY RANDOM() LIMIT 1")
        result = cursor.fetchone()
        if not result:
            return None
        order_id = result[0]

        # Update the order status
        new_status = random.choice(["SHIPPED", "DELIVERED"])
        cursor.execute(
            "UPDATE orders SET status = %s WHERE order_id = %s",
            (new_status, order_id)
        )

        conn.commit()
        end_time = time.time()

        return {"order_id": order_id, "exec_time": end_time - start_time}
    except Exception as e:
        conn.rollback()
        print(f"Update error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def run_workload(num_operations=1000, insert_ratio=0.7, max_workers=20):
    print(f"Starting write workload simulation with {num_operations} operations...")
    operation_times = {"insert": [], "update": []}

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []

        for _ in range(num_operations):
            # Determine operation type based on ratio
            if random.random() < insert_ratio:
                futures.append((executor.submit(execute_insert), "insert"))
            else:
                futures.append((executor.submit(execute_update), "update"))

        for future, op_type in futures:
            result = future.result()
            if result is not None:
                operation_times[op_type].append(result["exec_time"])

    # Calculate statistics for each operation type
    for op_type, times in operation_times.items():
        if times:
            avg_time = sum(times) / len(times)
            min_time = min(times)
            max_time = max(times)
            p95_time = sorted(times)[int(len(times) * 0.95)]

            print(f"\\nCompleted {len(times)} {op_type} operations")
            print(f"Average {op_type} time: {avg_time:.4f} seconds")
            print(f"Min {op_type} time: {min_time:.4f} seconds")
            print(f"Max {op_type} time: {max_time:.4f} seconds")
            print(f"95th percentile {op_type} time: {p95_time:.4f} seconds")
        else:
            print(f"No successful {op_type} operations")

if __name__ == "__main__":
    run_workload(num_operations=2000, insert_ratio=0.7, max_workers=30)

Step 5: Performance Testing and Tuning Process

Now let's establish a systematic approach to performance testing and tuning:

Establish baseline performance:

# Run read workload test
python read_workload.py

# Run write workload test
python write_workload.py

Identify bottlenecks using Performance Insights and CloudWatch Metrics:

Monitor key metrics during the test:

CPU Utilization
Database connections
Buffer cache hit ratio
Read/write IOPS
Read/write latency
Network throughput
Wait events in Performance Insights
Tune parameters based on identified bottlenecks:

For PostgreSQL workloads with high connection rates:

-- Increase max connections
ALTER SYSTEM SET max_connections = '1000';

-- Tune shared_buffers (typically 25% of instance memory)
ALTER SYSTEM SET shared_buffers = '8GB';

-- Tune work_mem for complex queries
ALTER SYSTEM SET work_mem = '64MB';

-- Tune maintenance_work_mem for vacuum operations
ALTER SYSTEM SET maintenance_work_mem = '1GB';

-- Tune effective_cache_size (typically 75% of instance memory)
ALTER SYSTEM SET effective_cache_size = '24GB';

-- Adjust autovacuum parameters for write-heavy workloads
ALTER SYSTEM SET autovacuum_vacuum_scale_factor = 0.05;
ALTER SYSTEM SET autovacuum_analyze_scale_factor = 0.025;
ALTER SYSTEM SET autovacuum_vacuum_cost_limit = 2000;

-- Apply changes
SELECT pg_reload_conf();

After each parameter change, re-run tests and measure impact:

# Run read and write workload tests again
python read_workload.py
python write_workload.py

Document performance improvements:

Create a documentation file showing:

Baseline performance metrics
Identified bottlenecks
Parameter changes made
Performance improvements observed
Recommended final configuration

Best Practices for Ongoing Optimization

Implement a Regular Performance Review Process

Schedule weekly reviews of Performance Insights and CloudWatch metrics.
Set up automated alerting for performance degradation.
Document performance changes when application behavior changes.

Create a Database Parameter Change Management Process

Test all parameter changes in a staging environment first.
Document each parameter change with justification and expected impact.
Implement changes during maintenance windows when possible.
Maintain parameter groups in version control or AWS CloudFormation.

SQL Query Review Process

Establish a code review process for database queries:

Require EXPLAIN ANALYZE for all new queries.
Set performance targets for critical queries.
Maintain a query library of common patterns and anti-patterns.

Conclusion 🚀💡

Optimizing Aurora PostgreSQL/MySQL for high-throughput workloads requires a systematic approach covering instance sizing, schema design, query optimization, and Aurora-specific features. By following the hands-on lab and implementing the best practices outlined in this article, you can achieve significant performance improvements for your most demanding workloads.

Remember that performance optimization is an ongoing process rather than a one-time task. Regular monitoring, testing, and tuning are essential to maintain optimal performance as your workloads evolve. ✨✨