DevOps NetworkDevOpsNetwork
Join Free
Practical infra coding

Plan a safe cleanup of stale snapshots

hardCloud SDK automation

Problem statement

Old disk snapshots pile up and cost money, but deleting the wrong one can destroy the only backup of a database or break a machine image. Write a cleanup tool that plans deletions, shows its reasoning for every snapshot, and only deletes with --apply.

The tool talks to a FakeEC2 client whose responses are simplified from the AWS EC2 API and keep its well-known field names: describe_snapshots(NextToken=...) returns Snapshots (with SnapshotId, StartTime, State, VolumeId, VolumeSize, Tags) and a NextToken while more pages remain; describe_images() returns Images whose BlockDeviceMappings[].Ebs.SnapshotId point at the snapshots they are built from; delete_snapshot(SnapshotId=...) deletes one, or raises APIError.

A snapshot may be deleted only if all of these hold:

  1. State is completed (never touch one still being created);
  2. it has no keep tag whose value is true (case- and space-insensitive);
  3. no image references it;
  4. it is older than 90 days relative to a fixed now.

Print the plan (deletions first, then kept snapshots with the reason, each group sorted by ID). In dry-run mode print how much space would be freed. With --apply, delete each planned snapshot, report failures without stopping, and exit 1 if any failed.

◈ DIAGRAM
snapshot pages (DescribeSnapshots, simplified)
page 1: snap-01 2026-01-10T03:00:00Z completed vol-a 100 GiB tags: {}
snap-02 2026-02-01T03:00:00Z completed vol-b 50 GiB tags: {keep: "true"}
page 2: snap-03 2026-03-15T03:00:00Z completed vol-c 200 GiB tags: {}
snap-04 2026-09-01T03:00:00Z completed vol-a 100 GiB tags: {}
page 3: snap-05 2026-04-20T03:00:00Z pending vol-d 20 GiB tags: {}
snap-06 2026-05-05T03:00:00Z completed vol-e 30 GiB tags: {keep: "TRUE "}
page 4: snap-07 2025-12-24T03:00:00Z completed vol-f 500 GiB tags: {}
snap-08 2026-06-26T00:00:00Z completed vol-g 10 GiB tags: {}
images (DescribeImages, simplified)
ami-111 BlockDeviceMappings -> snap-03
ami-222 BlockDeviceMappings -> snap-99 (a snapshot in another account), and one
mapping with no Ebs key (an instance-store device)
delete_snapshot("snap-07") fails with InvalidSnapshot.InUse
now = 2026-09-24T00:00:00Z, max age = 90 days

Examples

Example 1

Input: `python solution.py` (dry run)

Output: == DRY RUN: 2 to delete, 6 kept == delete snap-01 (100 GiB): 256 days old, unused delete snap-07 (500 GiB): 273 days old, unused keep snap-02: tagged keep=true keep snap-03: used by ami-111 keep snap-04: only 22 days old keep snap-05: state is pending keep snap-06: tagged keep=true keep snap-08: only 90 days old would free 600 GiB; re-run with --apply

Explanation: snap-08 is exactly 90 days old (26 June to 24 September at midnight), and the rule is *older than* 90 days, so it is kept. snap-06 is protected because "TRUE " normalizes to true.

Example 2

Input: `python solution.py --apply; echo "exit: $?"` (last lines)

Output: FAILED snap-07: InvalidSnapshot.InUse: snap-07 is in use by a pending copy deleted 1, failed 1 exit: 1

Explanation: The plan lines are printed first, as in the dry run. snap-01 is deleted; the failure on snap-07 is reported and reflected in the exit code.

Hints

Approach

Optimal

  1. Read everything, paginated. all_snapshots follows NextToken until it is absent. Planning from the first page only would make the tool look fine in testing and miss most of a real account.
  2. Index the images. snapshots_used_by_images builds snapshot -> image from every block device mapping. Mappings without an Ebs key (instance-store devices) are skipped instead of crashing, and a reference to a snapshot that is not in this account is harmless.
  3. One decision function. decide checks the rules in order of how cheap and how important they are: state, then the keep tag (normalized with strip().lower(), so "TRUE " still protects the snapshot), then image usage, then age. It always returns a reason, and the plan prints the reason for every snapshot, including kept ones. That is what lets a reviewer trust the plan.
  4. Time. StartTime is parsed into an aware datetime; now is a fixed aware value, so the output is reproducible and tests do not drift as days pass. Comparing aware and naive datetimes would raise TypeError, which is a common bug here.
  5. Dry run by default. Nothing is deleted without --apply. The dry run prints the space that would be freed, which is usually what the person approving the change wants to know.
  6. Apply, tolerating failures. Each delete is wrapped in its own try. One snapshot that is in use must not stop the rest of the cleanup, but it must show up in the output and in the exit code.

s is the number of snapshots and m the number of image block device mappings.

ComplexityTime O(s + m)Space O(s + m)
Python
import sys
from datetime import datetime, timedelta, timezone
NOW = datetime(2026, 9, 24, tzinfo=timezone.utc)
MAX_AGE = timedelta(days=90)
class APIError(Exception):
def __init__(self, code, message):
super().__init__(f"{code}: {message}")
self.code = code
def snap(sid, start, state, vol, size, tags=None):
return {"SnapshotId": sid, "StartTime": start, "State": state, "VolumeId": vol,
"VolumeSize": size, "Tags": [{"Key": k, "Value": v} for k, v in (tags or {}).items()]}
class FakeEC2:
# Simplified stand-in for an EC2-style client. Paginates with NextToken.
SNAPSHOT_PAGES = [
[snap("snap-01", "2026-01-10T03:00:00Z", "completed", "vol-a", 100),
snap("snap-02", "2026-02-01T03:00:00Z", "completed", "vol-b", 50, {"keep": "true"})],
[snap("snap-03", "2026-03-15T03:00:00Z", "completed", "vol-c", 200),
snap("snap-04", "2026-09-01T03:00:00Z", "completed", "vol-a", 100)],
[snap("snap-05", "2026-04-20T03:00:00Z", "pending", "vol-d", 20),
snap("snap-06", "2026-05-05T03:00:00Z", "completed", "vol-e", 30, {"keep": "TRUE "})],
[snap("snap-07", "2025-12-24T03:00:00Z", "completed", "vol-f", 500),
snap("snap-08", "2026-06-26T00:00:00Z", "completed", "vol-g", 10)],
]
IMAGES = [
{"ImageId": "ami-111", "BlockDeviceMappings": [{"DeviceName": "/dev/xvda",
"Ebs": {"SnapshotId": "snap-03"}}]},
{"ImageId": "ami-222", "BlockDeviceMappings": [{"DeviceName": "/dev/xvda",
"Ebs": {"SnapshotId": "snap-99"}},
{"DeviceName": "/dev/sdb",
"VirtualName": "ephemeral0"}]},
]
def __init__(self):
self.deleted = []
def describe_snapshots(self, NextToken=None):
i = int(NextToken or 0)
nxt = str(i + 1) if i + 1 < len(self.SNAPSHOT_PAGES) else None
page = {"Snapshots": self.SNAPSHOT_PAGES[i]}
if nxt:
page["NextToken"] = nxt
return page
def describe_images(self):
return {"Images": self.IMAGES}
def delete_snapshot(self, SnapshotId):
if SnapshotId == "snap-07":
raise APIError("InvalidSnapshot.InUse", f"{SnapshotId} is in use by a pending copy")
self.deleted.append(SnapshotId)
def all_snapshots(ec2):
token = None
while True:
page = ec2.describe_snapshots(NextToken=token)
yield from page.get("Snapshots", [])
token = page.get("NextToken")
if not token:
return
def snapshots_used_by_images(ec2):
used = {}
for image in ec2.describe_images().get("Images", []):
for mapping in image.get("BlockDeviceMappings", []):
sid = mapping.get("Ebs", {}).get("SnapshotId") # instance-store has no Ebs
if sid:
used[sid] = image["ImageId"]
return used
def parse_time(text):
return datetime.fromisoformat(text.replace("Z", "+00:00"))
def decide(s, used):
# Returns (delete?, reason).
tags = {t["Key"]: t.get("Value", "") for t in s.get("Tags") or []}
age = NOW - parse_time(s["StartTime"])
if s.get("State") != "completed":
return False, f"state is {s.get('State')}"
if tags.get("keep", "").strip().lower() == "true":
return False, "tagged keep=true"
if s["SnapshotId"] in used:
return False, f"used by {used[s['SnapshotId']]}"
if age <= MAX_AGE:
return False, f"only {age.days} days old"
return True, f"{age.days} days old, unused"
def main(apply):
ec2 = FakeEC2()
used = snapshots_used_by_images(ec2)
delete, keep = [], []
for s in all_snapshots(ec2):
ok, reason = decide(s, used)
(delete if ok else keep).append((s["SnapshotId"], s["VolumeSize"], reason))
print(f"== {'APPLY' if apply else 'DRY RUN'}: {len(delete)} to delete, {len(keep)} kept ==")
for sid, size, reason in sorted(delete):
print(f"delete {sid} ({size} GiB): {reason}")
for sid, size, reason in sorted(keep):
print(f"keep {sid}: {reason}")
if not apply:
print(f"would free {sum(size for _, size, _ in delete)} GiB; re-run with --apply")
return 0
failures = 0
for sid, size, _ in sorted(delete):
try:
ec2.delete_snapshot(SnapshotId=sid)
except APIError as e:
failures += 1
print(f"FAILED {sid}: {e}")
print(f"deleted {len(ec2.deleted)}, failed {failures}")
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main(apply="--apply" in sys.argv[1:]))

Follow-up questions

  • Keep the newest snapshot of every volume even if it is old, so no volume loses its last backup.
  • Add --max-delete N and refuse to apply a plan larger than that.
  • Run the plan across all regions in parallel, with a per-region summary.

Frequently asked questions

Snapshot and backup sprawl is a standard cloud cost problem, and cleanup scripts are a classic source of incidents when they delete something still needed. Interviewers want to see pagination, several independent safety checks, a dry-run default and graceful failure handling, not just the age filter.

snap-08 is exactly 90 days old. Whether the rule is > or >= decides whether it is deleted today or tomorrow, and a reviewer should be able to predict that from the policy text. Write the rule in words ("older than 90 days"), match it in code (age <= MAX_AGE means keep), and put a case exactly on the boundary in the test data.

They protect different things. The image check is automatic: a snapshot behind a machine image cannot be deleted without breaking launches, and the API may refuse anyway. The keep tag is for humans: a pre-migration backup is not used by any image but must survive. Relying on either one alone leaves a gap.