Python Automation That Saved Our Client $1000/Month On AWS Cloud Bills

A complete guide to shrinking over-provisioned RDS Postgres storage with Python, pgsync, ECS and Lambda

Akhilesh Mishra

Published Oct 3, 2026

Last modified Oct 3, 2026

I was working on a project to reduce cloud costs across multiple AWS accounts. We discovered many RDS Postgres databases were over-provisioned, wasting terabytes of storage.

When people built these databases initially, they gave each database 100 GB, 200 GB, sometimes 1 TB or 2 TB of disk. I saw a lot of databases with 2 TB allocated where the usage after five years was only 200 GB. Every app is not Uber. Every app is not Netflix. Some of them are only used by a few hundred enterprise users, and they will never need that much disk.

When we ran the numbers across ~70 databases, the extra provisioned storage alone was costing us around $1,000 per month.

AWS doesn't let you simply reduce allocated RDS storage. You can't even restore from a snapshot with less space. To fix this, you need to create a new RDS instance with less storage and move the data there. It's not straightforward, but it was necessary for a lot of databases.

To solve this, I wrote a Python script to automate the process. I packaged it as a Docker container, deployed it to ECS, and used a Lambda function to trigger it. This approach allowed us to efficiently right-size our databases and significantly cut down on unnecessary cloud spending.

Why the storage costs so much

RDS cost comes with two things. One is the compute capacity of the RDS instance plus the management charges, and the other is the storage charges.

Open the RDS console, start creating a database, and play with the storage value. Bump a single instance from 50 GB to 500 GB and watch the monthly estimate jump by roughly $60. Now think about it: three databases per environment, five environments, and each one carrying a few hundred GB of extra disk. That adds up very quickly.

One thing to keep in mind before you start cutting: database performance is not only about the disk type, it is also about the disk size. On gp3, for example, a small volume gives you a baseline of 3,000 IOPS, and larger volumes let you go much higher. So people think about disk type, disk size and instance type together for a given workload. Shrinking storage is a balance between performance and bill, not just "cut everything."

Deciding which databases to shrink

Before automating anything, we built a report. For every database we pulled:

Allocated storage → from the RDS instance config (describe_db_instances)
Free storage → from CloudWatch (FreeStorageSpace metric)

Then we put the data in an Excel sheet with some formulas. We did not touch databases where the saving was only 20 GB or 30 GB (a dollar or two). We focused on databases with a significant saving, both in GB and in percentage.

For example, 100 GB allocated and 30 GB used means we can cut ~70%. But we don't make it exactly 30 GB. We leave a buffer for the next 6 months or so:

Environment	Used	Buffer	New size
Dev / non-prod	30 GB	+20%	36 GB
Prod	500 GB	+30%	650 GB

Non-prod databases are just for testing, and you can always do the cleanup there. Production stores the real data, so the buffer is bigger. We also turned on storage autoscaling so if a database grows over a weekend, it scales on its own and nobody has to wake up for it.

When we calculated the whole sheet, the saving came to almost $1,000 per month. Good enough to put some human hours into automating it.

I have a script that generates this report as an Excel file in my python-for-devops repo: python3 rds_storage_report.py. In the automation below, I calculate the new size automatically with the same idea.

Why pgsync and not pg_dump/pg_restore

In my first version I used pg_dump to take the backup of the database and pg_restore to restore it to the new instance. It worked fine for 10–30 GB databases. But for a 100 GB database the backup and restore took almost 3 hours, and we had databases with 200 GB and 500 GB.

So I switched to pgsync . It connects to both databases at the same time and copies the data directly from source to destination. No dump file, no S3 bucket in the middle. It was much faster for our use case.

pgsync still uses pg_dump/pg_restore for one small part: copying the schema (the table definitions). That detail will matter later when we build the Docker image.

Architecture

1
2
3
4
5
6
7
8
9
10
11
12
13
                         ┌──────────────────────────── VPC ────────────────────────────┐
                         │                                                             │
 {"db_link": "..."}      │   ┌──────────────────────┐        ┌─────────────────────┐   │
 ─────────────►  Lambda ─┼─► │  ECS Fargate task    │ ─5432─►│  RDS  mydb          │   │
  (manual trigger)       │   │  python3 main.py     │        │  (original, 500 GB) │   │
                         │   │  + pgsync            │        └─────────────────────┘   │
                         │   │  + postgresql-client │                                  │
                         │   │                      │ ─5432─►┌─────────────────────┐   │
                         │   └──────────┬───────────┘        │  RDS  mydbnew       │   │
                         │              │ logs               │  (copy, 65 GB)      │   │
                         │              ▼                    └─────────────────────┘   │
                         │        CloudWatch Logs                                      │
                         └─────────────────────────────────────────────────────────────┘

Why this setup:

ECS, not Lambda, runs the migration. Lambda is good for a job that runs for a few seconds or minutes, but it has a 15-minute limit. A database migration can take hours. So the real work runs as an ECS Fargate task.
Lambda is the trigger. You pass the database details in the Lambda event, and Lambda starts the ECS task with those details as the container command.
Everything runs inside the VPC. In the real world your database is never public. The ECS task runs in the same VPC (if you are going across VPCs, you need VPC peering) and talks to both databases over private connections.

Migration steps

As part of the database migration, I will follow the below steps.

Use the DB link passed as input to figure out the instance ID, DB name, user, password and port.
Use boto3 to pull the RDS instance details and its storage utilization from CloudWatch, and calculate the new storage size.
Open port 5432 on the database security group for the ECS task's security group.
Duplicate the RDS instance with the new storage size, and name it with a new suffix. Everything else (subnet group, parameter group, security groups, tags, KMS key…) stays the same.
Wait for the new database to be ready to take connections, with a kill switch if it never comes up.
Copy the schema and sync the data from old to new with pgsync.
Swap the original database and the duplicated one by renaming them.
Stop the old DB instance and manually delete it after a few days.
Revoke the security group rule we opened in step 3.

Project structure

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
rds-migration/
├── generate-dummydata-rds.py     # seed a test DB with fake data
├── requirements.txt
└── implementation/
    ├── migrator/                 # the automation (runs on ECS)
    │   ├── main.py
    │   ├── requirements.txt
    │   └── Dockerfile
    ├── lambda/
    │   └── main.py               # trigger
    └── infra/                    # Terraform: ECR, ECS, IAM, Lambda, logs
        ├── ecr.tf  ecs.tf  iam.tf  lambda.tf
        ├── image.tf  awsdocker.sh
        ├── cloudwatch.tf  data.tf
        └── variables.tf  versions.tf

Building the automation, one function at a time

The input: a DB link

The first design decision is what to pass into the automation. You have two choices:

Pass only the database name, and assume all your passwords live in Secrets Manager under a standard naming convention (like /dev/rds/<db-name>).
Pass a full DB link and let the automation parse everything out of it.

If your company follows a naming convention everywhere, option 1 is great. For this post I went with option 2 because it does not depend on how somebody named their secret. The DB link looks like this:

1
postgres://{db_user}:{db_password}@{db_host_endpoint}:{db_port}/{database_name}

All the code below lives in implementation/migrator/main.py.

0. Imports, clients and logging

I create the boto3 clients outside the functions because multiple functions need the same client and we don't want to create it again and again.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import subprocess
import boto3
import logging
import psycopg2
from psycopg2 import OperationalError
import time
import argparse
import sys
from os import getenv
from botocore.exceptions import ClientError
from typing import Tuple
from datetime import datetime, timedelta
import math

# (host, db, user, password, port)
DBDetails = Tuple[str, str, str, str, str]

ec2_client = boto3.client("ec2")
rds = boto3.client("rds")
cloudwatch = boto3.client("cloudwatch")

Most people in production don't use print for logs, they use the logging module. This setup gives a clean format and silences the noisy boto3 logs, so you only see what matters:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def log_setup() -> None:
    logger = logging.getLogger()
    for handler in logger.handlers:
        logger.removeHandler(handler)
    handler = logging.StreamHandler(sys.stdout)

    dformat = "[%(filename)s:%(lineno)d] :%(levelname)8s: %(message)s"
    handler.setFormatter(logging.Formatter(dformat))
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG if getenv("DEBUG", False) else logging.INFO)

    # Suppress the more verbose modules
    for noisy in ("botocore", "s3transfer", "boto3", "urllib3"):
        logging.getLogger(noisy).setLevel(logging.WARN)

1. Parse the DB link

Plain split() calls. The first part of the RDS hostname is always the DB instance identifier, so we get the instance ID for free:

1
2
3
4
5
6
7
8
def get_db_details(db_link: str) -> DBDetails:
    # postgres://{user}:{password}@{host}:{port}/{db_name}
    user = db_link.split(":")[1].split("/")[-1]
    password = db_link.split(":")[2].split("@")[0]
    db = db_link.split("/")[-1]
    host = db_link.split(":")[2].split("@")[-1].split(".")[0]   # instance ID
    port = db_link.split(":")[3].split("/")[0]
    return host, db, user, password, port

For postgres://postgres:Secret123@rdsmigration.abc123.ap-south-1.rds.amazonaws.com:5432/rdsmigration this returns ("rdsmigration", "rdsmigration", "postgres", "Secret123", "5432").

2. Get RDS details of the original DB instance

1
2
def source_rds_instance(dbinstance: str) -> dict:
    return rds.describe_db_instances(DBInstanceIdentifier=dbinstance)["DBInstances"][0]

3. Get the storage utilization for the DB from CloudWatch monitoring

If you open the Monitoring tab of your RDS instance and search for "free storage", that's the FreeStorageSpace metric. I take the minimum over the last 7 days, so we size for the worst point, not the average.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def get_db_freestorage(DBInstanceIdentifier):
    response = cloudwatch.get_metric_data(
        MetricDataQueries=[
            {
                "Id": "fetching_FreeStorageSpace",
                "MetricStat": {
                    "Metric": {
                        "Namespace": "AWS/RDS",
                        "MetricName": "FreeStorageSpace",
                        "Dimensions": [
                            {"Name": "DBInstanceIdentifier", "Value": DBInstanceIdentifier}
                        ],
                    },
                    "Period": 86400 * 7,
                    "Stat": "Minimum",
                },
            }
        ],
        StartTime=(datetime.now() - timedelta(seconds=86400 * 7)).timestamp(),
        EndTime=datetime.now().timestamp(),
        ScanBy="TimestampDescending",
    )
    return round(response["MetricDataResults"][0]["Values"][0] / 1024**3, 2)  # bytes -> GB

4. Compare the allocated and free disk space, and calculate the new size

Used storage = allocated − free. New size = used + 40% buffer. gp3 needs a minimum of 20 GB, so we never go below that.

1
2
3
4
5
6
7
8
9
def evaluate_db_storage(dbinstance):
    free_storage = get_db_freestorage(dbinstance)
    allocated_storage = source_rds_instance(dbinstance)["AllocatedStorage"]
    used_storage = allocated_storage - free_storage
    new_allocated_storage = 1.4 * used_storage
    # gp3 min limit is 20 GB
    if new_allocated_storage < 20:
        return 20
    return math.ceil(new_allocated_storage)

Tune the 1.4 multiplier to your own policy. We used +20% for non-prod and +30% for prod, plus storage autoscaling.

5. Duplicate the RDS instance with the new storage value

This function will duplicate the RDS instance with the new value for allocated disk space. The rest of the values will be the same as the original RDS instance.

This is the most important part. A real database in your company has a lot of config, and you have to copy every single one: same security groups, same subnet group, same parameter group, same option group, same tags, same KMS key, same backup window. You cannot copy only five things. The application should not see any difference.

Note: If you are using gp2 disks, this function migrates them to gp3 for you (gp3 is cheaper and gives better baseline performance). You can change that in the StorageType key.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def duplicate_rds(db_link: str, new_allocated_storage: int) -> dict:
    dbinstance, db, user, password, port = get_db_details(db_link)
    source_rds_data = source_rds_instance(dbinstance)

    instance_params = {
        "DBName": db,
        "DBInstanceIdentifier": f"{source_rds_data['DBInstanceIdentifier']}new",
        "AllocatedStorage": new_allocated_storage,
        "DBInstanceClass": source_rds_data["DBInstanceClass"],
        "Engine": source_rds_data["Engine"],
        "MasterUsername": user,
        "MasterUserPassword": password,
        "Port": int(port),
        "DBSecurityGroups": [
            items["DBSecurityGroupName"] for items in source_rds_data["DBSecurityGroups"]
        ],
        "VpcSecurityGroupIds": [
            items["VpcSecurityGroupId"] for items in source_rds_data["VpcSecurityGroups"]
        ],
        "AvailabilityZone": source_rds_data["AvailabilityZone"],
        "DBSubnetGroupName": source_rds_data["DBSubnetGroup"]["DBSubnetGroupName"],
        "PreferredMaintenanceWindow": source_rds_data["PreferredMaintenanceWindow"],
        "DBParameterGroupName": source_rds_data["DBParameterGroups"][0]["DBParameterGroupName"],
        "BackupRetentionPeriod": source_rds_data["BackupRetentionPeriod"],
        "PreferredBackupWindow": source_rds_data["PreferredBackupWindow"],
        "MultiAZ": source_rds_data["MultiAZ"],
        "EngineVersion": source_rds_data["EngineVersion"],
        "AutoMinorVersionUpgrade": source_rds_data["AutoMinorVersionUpgrade"],
        "LicenseModel": source_rds_data["LicenseModel"],
        "OptionGroupName": source_rds_data["OptionGroupMemberships"][0]["OptionGroupName"],
        "PubliclyAccessible": source_rds_data["PubliclyAccessible"],
        "Tags": source_rds_data["TagList"],
        "StorageType": (
            "gp3" if source_rds_data["StorageType"] == "gp2" else source_rds_data["StorageType"]
        ),
        "StorageEncrypted": source_rds_data["StorageEncrypted"],
        "KmsKeyId": source_rds_data["KmsKeyId"],
        "CopyTagsToSnapshot": source_rds_data["CopyTagsToSnapshot"],
        "EnableIAMDatabaseAuthentication": source_rds_data["IAMDatabaseAuthenticationEnabled"],
        "EnablePerformanceInsights": source_rds_data["PerformanceInsightsEnabled"],
        "DeletionProtection": source_rds_data["DeletionProtection"],
        "EnableCustomerOwnedIp": source_rds_data["CustomerOwnedIpEnabled"],
        "BackupTarget": source_rds_data["BackupTarget"],
        "NetworkType": source_rds_data["NetworkType"],
        "CACertificateIdentifier": source_rds_data["CACertificateIdentifier"],
    }

    # keep storage autoscaling if the source has it
    if source_rds_data.get("MaxAllocatedStorage"):
        instance_params["MaxAllocatedStorage"] = source_rds_data["MaxAllocatedStorage"]

    return rds.create_db_instance(**instance_params)

The new instance gets the name <original>new. Since the endpoint format is always <instance-id>.<random-id>.<region>.rds.amazonaws.com and the middle part never changes within one account and region, we can build the new endpoint ourselves:

1
2
3
4
old_db_endpoint = source_rds_data["Endpoint"]["Address"]
# rdsmigration.abc123.ap-south-1.rds.amazonaws.com
new_db_endpoint = new_id + "." + ".".join(old_db_endpoint.split(".")[1:])
# rdsmigrationnew.abc123.ap-south-1.rds.amazonaws.com

split() breaks a string into a list using a delimiter. join() is the opposite: it takes a list and joins the items into a string with a delimiter.

6. Check if RDS is ready (with a kill switch)

The RDS status saying available does not mean your database is ready to take connections. The only reliable way to know is to actually try to connect. So we try, wait, try again.

But we also need a kill switch. If something is misconfigured and the database never comes up, you don't want to trigger the automation today and come back tomorrow to find it still looping. A normal duplication finishes in 5–15 minutes, so if it is not ready in 30 minutes, something is off, and we stop.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def check_rds_availability(host, port, dbname, user, password,
                           timeout: int = 1800, interval: int = 30) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            conn = psycopg2.connect(
                host=host, port=port, dbname=dbname,
                user=user, password=password, connect_timeout=10,
            )
            conn.close()
            logging.info(f"Connection established successfully with {host}.")
            return True
        except OperationalError:
            logging.info(f"{host} is not ready to take connections yet. Retrying in {interval}s...")
            time.sleep(interval)

    logging.error(f"Timed out after {timeout}s. {host} never became ready.")
    return False

Improvement over the repo version: the repo uses signal.alarm() for the timeout, but the finally: signal.alarm(0) inside the loop resets the alarm after the first attempt, so the kill switch never fires, and migrate_rds never checked the return value. The deadline-based loop above is simpler and actually stops.

7. Open the database security group for the ECS task

In production your database only allows connections from certain security groups. The ECS task that runs this automation has its own security group, so before syncing, we add a rule that allows port 5432 from the task's SG to the database's SG (plus the matching egress rule on the task side). After the job is done, we remove it.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def _sg_rule(port: int, group_id: str) -> list:
    return [{
        "IpProtocol": "tcp",
        "FromPort": port,
        "ToPort": port,
        "UserIdGroupPairs": [{"Description": "rds-migration access", "GroupId": group_id}],
    }]

def allow_sgs(from_sg: str, to_sg: str, port: int) -> None:
    for call, group, peer in (
        (ec2_client.authorize_security_group_egress, from_sg, to_sg),
        (ec2_client.authorize_security_group_ingress, to_sg, from_sg),
    ):
        try:
            call(GroupId=group, IpPermissions=_sg_rule(port, peer))
        except ClientError as error:
            if error.response["Error"]["Code"] != "InvalidPermission.Duplicate":
                logging.error("SG change failed: %s", error)
                raise
    logging.info("Allowed %s -> %s on port %s", from_sg, to_sg, port)

def revoke_sgs(from_sg: str, to_sg: str, port: int) -> None:
    for call, group, peer in (
        (ec2_client.revoke_security_group_egress, from_sg, to_sg),
        (ec2_client.revoke_security_group_ingress, to_sg, from_sg),
    ):
        try:
            call(GroupId=group, IpPermissions=_sg_rule(port, peer))
        except ClientError as error:
            if error.response["Error"]["Code"] != "InvalidPermission.NotFound":
                logging.error("SG change failed: %s", error)
                raise
    logging.info("Revoked %s -> %s on port %s", from_sg, to_sg, port)

The try/except with ClientError (it comes with boto3) makes it safe to re-run: if the rule already exists or is already gone, we just move on.

The repo has these as two longer functions with the four boto3 calls written out. Same logic, I just folded them into a loop here.

8. Sync the data with pgsync

pgsync needs a config file called .pgsync.yml with minimum three things:

from: the source DB link (the original database)
to: the destination DB link (the new database)
to_safe: true: by default pgsync refuses to write to any destination that is not localhost, as a safety net against overwriting the wrong database. This flag says "yes, I know, this remote destination is safe to write to."

Then we run pgsync --schema-first --all-schemas. --schema-first copies the table definitions before the data, and --all-schemas includes every schema, not only public.

To run a command from Python you can use os.system, but it does not give you the output back. subprocess does, so we can put pgsync's output in our logs.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def sync_dbs(old_db: str, new_db: str) -> None:
    logging.info("Creating .pgsync.yml with source and destination db links")
    with open(".pgsync.yml", "w") as f:
        f.write(f"from: {old_db}\n")
        f.write(f"to: {new_db}\n")
        f.write("to_safe: true\n")

    logging.info("Syncing DBs")
    result = subprocess.run(
        ["pgsync", "--schema-first", "--all-schemas"],
        capture_output=True,
        text=True,
    )
    logging.info(result.stdout)
    if result.returncode != 0:
        logging.error(f"pgsync failed (code {result.returncode}): {result.stderr}")
        sys.exit(1)   # never swap databases after a failed sync
    logging.info("Sync completed")

Improvement over the repo version: the repo logs the failure but returns normally, so the automation goes on to swap the databases even if the sync failed. That would point your application at an empty database. Here we exit, and the original database stays untouched.

9. Rename the RDS instance

1
2
3
4
5
6
7
8
9
def rename_rds(old: str, new: str) -> None:
    try:
        rds.modify_db_instance(
            DBInstanceIdentifier=old, NewDBInstanceIdentifier=new, ApplyImmediately=True
        )
        logging.info(f"DB renamed - {new}")
    except Exception as e:
        logging.error(f"Issue with renaming {old} -> {new} : {e}")
        sys.exit(1)

10. Swap the RDS instance

The application points at an endpoint like rdsmigration.abc123.... We don't want to touch the application config. So:

rdsmigration → rdsmigration-old
rdsmigrationnew → rdsmigration

Now the original endpoint belongs to the new, smaller database. The application connects to the new database without seeing any difference.

I used the sleep from the time module to wait between the renames instead of a boto3 waiter, because the old name has to be fully released before the new instance can take it.

1
2
3
4
5
6
def swap_db(old: str, new: str) -> None:
    logging.info(f"Renaming db: {old} -> {old}-old")
    rename_rds(old, f"{old}-old")
    time.sleep(300)  # wait for the rename to finish and the name to be freed
    logging.info(f"Renaming db: {new} -> {old}")
    rename_rds(new, old)

11. Stop the RDS instance

In production you don't go and delete the old database straight away. You stop it, watch the application for a few days, and if everything is fine, delete it manually.

1
2
3
4
5
6
7
def stop_rds(dbinstance: str) -> None:
    try:
        logging.info(f"Stopping the RDS instance - {dbinstance}")
        rds.stop_db_instance(DBInstanceIdentifier=dbinstance)
    except Exception as e:
        logging.error(f"Issue with stopping - {dbinstance} -> {e}")
        sys.exit(1)

AWS automatically starts a stopped RDS instance again after 7 days. Delete the -old instance within that window, or it will come back up and start costing money again.

We call all the above functions to migrate the Postgres RDS instance

Now we put all the pieces together in the right order. The try/finally makes sure the security group rule is always removed, even if something fails in the middle.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def migrate_rds(db_link: str) -> None:
    dbinstance, db, user, password, port = get_db_details(db_link)
    source_rds_data = source_rds_instance(dbinstance)
    current_size = source_rds_data["AllocatedStorage"]
    size = evaluate_db_storage(dbinstance)

    if size >= current_size:
        logging.info(f"{dbinstance}: {current_size} GB allocated, {size} GB needed. Nothing to save.")
        return

    from_sg = getenv("SG_ID")   # ECS task SG, injected by the task definition
    to_sg = source_rds_data["VpcSecurityGroups"][0]["VpcSecurityGroupId"]

    allow_sgs(from_sg, to_sg, int(port))
    try:
        logging.info(f"Duplicating {dbinstance}: {current_size} GB -> {size} GB")
        new_id = duplicate_rds(db_link, size)["DBInstance"]["DBInstanceIdentifier"]

        old_db_endpoint = source_rds_data["Endpoint"]["Address"]
        new_db_endpoint = new_id + "." + ".".join(old_db_endpoint.split(".")[1:])

        if not check_rds_availability(new_db_endpoint, port, db, user, password):
            logging.error(f"{new_id} never became ready. Aborting; {dbinstance} untouched.")
            sys.exit(1)

        source_db_link = f"postgresql://{user}:{password}@{old_db_endpoint}:{port}/{db}"
        destination_db_link = f"postgresql://{user}:{password}@{new_db_endpoint}:{port}/{db}"
        sync_dbs(source_db_link, destination_db_link)

        logging.info("Swapping dbs")
        swap_db(dbinstance, new_id)

        logging.info(f"Stopping {dbinstance}-old")
        stop_rds(f"{dbinstance}-old")
    finally:
        revoke_sgs(from_sg, to_sg, int(port))

And the CLI entry point, so the automation takes the DB link as a command-line argument:

1
2
3
4
5
6
7
8
9
10
11
12
def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Shrink RDS Postgres storage.")
    parser.add_argument("db_link", help="postgres://user:password@host:port/db")
    return parser.parse_args()

def run() -> None:
    log_setup()
    args = parse_arguments()
    migrate_rds(args.db_link)

if __name__ == "__main__":
    run()

requirements.txt:

1
2
boto3
psycopg2-binary

Packaging it with Docker

The container needs three things: Python with our dependencies, pgsync (it's a Ruby gem), and the Postgres client.

Here is the gotcha that cost me some time. pgsync uses pg_dump/pg_restore to copy the schema, and the Postgres client version has to be equal to or higher than your database version. Ubuntu 22.04's default repo ships postgresql-client-14. My database was on Postgres 16, so the schema copy failed.

The fix is to add the official PostgreSQL (PGDG) apt repo and install postgresql-client-16:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
FROM ubuntu:22.04

# Add the official PostgreSQL (PGDG) apt repo so we can install
# postgresql-client-16 instead of Ubuntu 22.04's default v14
RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        wget ca-certificates gnupg lsb-release && \
    wget --quiet -O - https://www.postgresql.org/media/keys/ACCC4CF8.asc | \
        gpg --dearmor -o /usr/share/keyrings/postgresql.gpg && \
    echo "deb [signed-by=/usr/share/keyrings/postgresql.gpg] http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" \
        > /etc/apt/sources.list.d/pgdg.list

RUN apt-get update && \
    apt-get install -y \
        python3 python3-pip ruby-dev libpq-dev build-essential \
        postgresql-client-16

# Install pgsync gem
RUN gem install pgsync

RUN apt-get clean && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
COPY main.py main.py

CMD ["python3", "main.py"]

Build it and verify the client version before you go anywhere near ECS:

1
2
3
4
cd implementation/migrator
docker build -t rds-migration .
docker run --rm rds-migration pg_dump --version
# pg_dump (PostgreSQL) 16.x

Match postgresql-client-XX to the highest Postgres version you are migrating.

The Lambda trigger

The Lambda function takes the DB link from the event and starts the ECS task, overriding the container command with python3 main.py <db_link>. All the cluster, task definition, subnet and SG details come from environment variables set by Terraform.

implementation/lambda/main.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import boto3
from os import getenv
import json
import logging
import sys

def log_setup():
    logger = logging.getLogger()
    for handler in logger.handlers:
        logger.removeHandler(handler)
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(logging.Formatter("[%(filename)s:%(lineno)d] :%(levelname)8s: %(message)s"))
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG if getenv("DEBUG", False) else logging.INFO)

def run_ecs_task(db_link):
    ecs = boto3.client("ecs")
    response = ecs.run_task(
        cluster=getenv("ECS_CLUSTER"),
        taskDefinition=getenv("ECS_TASK_DEF").split("/")[-1],
        launchType="FARGATE",
        networkConfiguration={
            "awsvpcConfiguration": {
                "subnets": json.loads(getenv("SUBNET_GROUP")),
                "securityGroups": [getenv("SG")],
                "assignPublicIp": "ENABLED",
            }
        },
        overrides={
            "containerOverrides": [
                {
                    "name": getenv("ECS_CONTAINER"),
                    "command": ["python3", "main.py", db_link],
                }
            ]
        },
    )
    return response["tasks"][0]["taskArn"]

def handler(event, context):
    log_setup()
    # event = {"db_link": "postgres://{user}:{password}@{host}:{port}/{db_name}"}
    db_link = event["db_link"]
    logging.info(f'Triggering ECS task for RDS {db_link.split(":")[2].split("@")[-1]}')
    ecs_task = run_ecs_task(db_link)
    logging.info(f"ECS task triggered -> {ecs_task}")
    return {"taskArn": ecs_task}

assignPublicIp: ENABLED is there because I use the default VPC's public subnets, and the task needs to reach ECR and the AWS APIs. In a private subnet, set it to DISABLED and give the subnet a NAT gateway or VPC endpoints.

Infrastructure with Terraform

Everything is in implementation/infra/. Here is what each file creates and why.

ECR + image build (ecr.tf, image.tf, awsdocker.sh): an ECR repo, and a null_resource that builds the migrator image for linux/amd64 and pushes it during terraform apply. So you need Docker running on the machine where you run Terraform.

Security group (ecs.tf): the task's SG only has egress on 443 (AWS APIs and ECR). Port 5432 to the database is added at runtime by allow_sgs() and removed by revoke_sgs(). The database is never left open.

1
2
3
4
5
6
7
8
9
10
11
12
resource "aws_security_group" "rds_migration_sg" {
  name   = "${var.environment}-rds-shrink"
  vpc_id = data.aws_vpc.default.id

  egress {
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
    description = "Access to AWS API"
  }
}

Task definition (ecs.tf): Fargate, 256 CPU / 2 GB memory, 30 GB ephemeral storage, logs to CloudWatch, and the task SG ID passed in as SG_ID (that's what migrate_rds reads with getenv("SG_ID")).

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
resource "aws_ecs_task_definition" "rds_migration" {
  family = "${var.environment}-rds-migration"
  container_definitions = jsonencode([{
    name      = "rds-migration"
    image     = "${local.repository}:${local.tag}"
    essential = true
    logConfiguration = {
      logDriver = "awslogs"
      options = {
        "awslogs-group"         = aws_cloudwatch_log_group.rds_migration.name
        "awslogs-region"        = data.aws_region.current.region
        "awslogs-stream-prefix" = "ecs"
      }
    }
    environment = [
      { name = "SG_ID", value = aws_security_group.rds_migration_sg.id },
      { name = "ENVIRONMENT", value = var.environment },
    ]
  }])

  cpu                      = 256
  memory                   = 2048
  network_mode             = "awsvpc"
  requires_compatibilities = ["FARGATE"]
  execution_role_arn       = aws_iam_role.rds_migration_ecs_execution_role.arn  # pull image, write logs
  task_role_arn            = aws_iam_role.rds_migration_ecs_task_role.arn       # what our Python calls

  ephemeral_storage {
    size_in_gib = 30
  }
}

IAM (iam.tf): two roles, and it's worth knowing the difference because people ask this in interviews.

Execution role: used by ECS itself to pull the image from ECR and push logs. Uses the AWS-managed AmazonECSTaskExecutionRolePolicy.
Task role: used by your code inside the container. It needs exactly what the script calls:

1
2
3
4
5
6
rds:DescribeDBInstances      rds:CreateDBInstance       rds:AddTagsToResource
rds:ModifyDBInstance         rds:StopDBInstance
ec2:AuthorizeSecurityGroupIngress   ec2:AuthorizeSecurityGroupEgress
ec2:RevokeSecurityGroupIngress      ec2:RevokeSecurityGroupEgress
cloudwatch:GetMetricData
kms:*   (to create an encrypted instance with the source's KMS key)

Lambda (lambda.tf): zips ../lambda, runs on python3.13, and gets ecs:RunTask plus iam:PassRole (it has to pass the two ECS roles to the task). Its environment variables wire it to the cluster, task definition, subnets and SG.

Logs (cloudwatch.tf): a log group with 7-day retention, plus a saved Logs Insights query that hides the noisy "waiting" lines.

Running it end to end

Step 1: Create a test database and fill it with data

Create a Postgres RDS instance with 50 GB of storage in your default VPC (I used Postgres 16, db.t4g.micro). Then seed it with fake data so the sync has something to copy:

1
2
3
4
5
6
7
8
export PGHOST="rdsmigration.xxxxxxxx.ap-south-1.rds.amazonaws.com"
export PGPORT="5432"
export PGDATABASE="rdsmigration"
export PGUSER="postgres"
export PGPASSWORD="<your-password>"

pip install psycopg2-binary faker
python3 generate-dummydata-rds.py --rows 1000000 --batch-size 5000

Step 2: Authenticate to AWS

Don't keep long-lived access keys on your laptop. With the latest AWS CLI v2 you can run:

1
aws login

It opens a browser, you sign in, and it keeps temporary credentials locally that expire on their own. If someone asks you in an interview how you log in to the AWS CLI, this is a good answer.

Step 3: Deploy the infrastructure

1
2
3
cd implementation/infra
terraform init
terraform apply

This builds and pushes the Docker image, and creates the ECS cluster, task definition, IAM roles, security group, Lambda and log group.

Step 4: Trigger the migration

From the Lambda console, create a test event:

1
{"db_link": "postgres://postgres:<password>@rdsmigration.xxxxxxxx.ap-south-1.rds.amazonaws.com:5432/rdsmigration"}

Or from the CLI:

1
2
3
4
5
aws lambda invoke \
  --function-name boot-rds-migration-lambda \
  --cli-binary-format raw-in-base64-out \
  --payload '{"db_link":"postgres://postgres:<password>@rdsmigration.xxxxxxxx.ap-south-1.rds.amazonaws.com:5432/rdsmigration"}' \
  response.json

Step 5: Watch the logs

1
aws logs tail /aws/ecs/boot-rds-migration-may25 --follow

You will see the automation move through each stage:

1
2
3
4
5
6
7
8
9
10
11
[main.py:..] :    INFO: Allowed sg-0ecs... -> sg-0rds... on port 5432
[main.py:..] :    INFO: Duplicating rdsmigration: 50 GB -> 20 GB
[main.py:..] :    INFO: rdsmigrationnew.xxxx... is not ready to take connections yet. Retrying in 30s...
...
[main.py:..] :    INFO: Connection established successfully with rdsmigrationnew.xxxx...
[main.py:..] :    INFO: Syncing DBs
[main.py:..] :    INFO: Sync completed
[main.py:..] :    INFO: Renaming db: rdsmigration -> rdsmigration-old
[main.py:..] :    INFO: Renaming db: rdsmigrationnew -> rdsmigration
[main.py:..] :    INFO: Stopping rdsmigration-old
[main.py:..] :    INFO: Revoked sg-0ecs... -> sg-0rds... on port 5432

In the RDS console you should now see rdsmigration with 20 GB gp3 storage and rdsmigration-old stopped. Connect to rdsmigration and check the row count matches.

Testing locally first

You can run the same container from your laptop before deploying to ECS. Your database has to be reachable from your machine for this (I temporarily made my test database publicly accessible; never do that for a real one):

1
2
3
4
5
docker run --rm \
  -v ~/.aws:/root/.aws:ro \
  -e AWS_DEFAULT_REGION=ap-south-1 \
  -e SG_ID=<a-security-group-id> \
  rds-migration python3 main.py "postgres://postgres:<password>@rdsmigration.xxxx.ap-south-1.rds.amazonaws.com:5432/rdsmigration"

Cleanup

1
2
3
aws rds delete-db-instance --db-instance-identifier rdsmigration-old --skip-final-snapshot
aws rds delete-db-instance --db-instance-identifier rdsmigration --skip-final-snapshot
cd implementation/infra && terraform destroy

Pitfalls

pg_dump/pg_restore is slow for big databases. That's why I moved to pgsync. 100 GB took almost 3 hours with dump and restore.

Plan for a write freeze. pgsync is a fast one-time copy, not continuous replication. Anything written to the old database after the sync starts will not be on the new one. Run it in a maintenance window, or stop the application writes during the migration. If you truly need zero downtime, look at logical replication or AWS DMS.

The swap is not instant. Between renaming the old instance and the new one taking its name, the endpoint does not resolve. With the 300-second wait, expect around 5–10 minutes where the application cannot connect. Schedule it.

Postgres client version. It must be equal to or higher than your database version, or the schema copy fails. Check with pg_dump --version inside the image.

Not every config maps 1:1. A few things to watch in duplicate_rds():

Multi-AZ instances: create_db_instance rejects AvailabilityZone together with MultiAZ=True. Drop the AZ for those.
Unencrypted instances have no KmsKeyId key in the response. Use source_rds_data.get("KmsKeyId") and leave it out if it's None.
Read replicas, IAM-only auth and custom option groups need extra handling.

Special characters in passwords. The split() parsing breaks if the password contains :, @ or /. A safer parser:

1
2
3
4
5
6
from urllib.parse import urlparse, unquote

def get_db_details(db_link: str):
    u = urlparse(db_link)
    return (u.hostname.split(".")[0], u.path.lstrip("/"),
            u.username, unquote(u.password), str(u.port or 5432))

Passwords in the Lambda event. The DB link travels in the Lambda event and the ECS command override, so it shows up in CloudTrail and the ECS console. For production, pass only the instance ID and let the container read the password from Secrets Manager (that's option 1 from the input decision above).

Stopped instances come back. AWS restarts a stopped RDS instance after 7 days. Delete -old once you are confident.

# finops # amazon-ecs # python

Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.

Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article

Builder Center

Python Automation That Saved Our Client $1000/Month On AWS Cloud Bills

Akhilesh Mishra

Why the storage costs so much

Deciding which databases to shrink

Why pgsync and not pg_dump/pg_restore

Architecture

Migration steps

Project structure

Building the automation, one function at a time

The input: a DB link

0. Imports, clients and logging

1. Parse the DB link

2. Get RDS details of the original DB instance

3. Get the storage utilization for the DB from CloudWatch monitoring

4. Compare the allocated and free disk space, and calculate the new size

5. Duplicate the RDS instance with the new storage value

6. Check if RDS is ready (with a kill switch)

7. Open the database security group for the ECS task

8. Sync the data with pgsync

9. Rename the RDS instance

10. Swap the RDS instance

11. Stop the RDS instance

We call all the above functions to migrate the Postgres RDS instance

Packaging it with Docker

The Lambda trigger

Infrastructure with Terraform

Running it end to end

Step 1: Create a test database and fill it with data

Step 2: Authenticate to AWS

Step 3: Deploy the infrastructure

Step 4: Trigger the migration

Step 5: Watch the logs

Testing locally first

Cleanup

Pitfalls