Auditing Unused Cloud Resources with Python
Every long-lived account collects things nobody deleted: volumes detached during an instance rebuild, snapshots from a migration that finished two years ago, a load balancer left standing when its service moved to a different account. This guide, part of IaC cost and governance under Python IaC fundamentals and strategy, builds a typed boto3 auditor that enumerates those resources properly, proves disuse from more than one independent signal, cross-checks every finding against what your stacks actually manage, and produces a report a human approves — rather than a script that deletes and hopes.
Context
The reason orphan cleanup keeps getting deferred is that the cost of a mistake is asymmetric. Leaving an unattached 500 GiB gp3 volume alive costs roughly forty dollars a month. Deleting the one volume that held a database's last pre-migration copy costs an incident review. Any tool you build has to be biased toward that asymmetry: cheap to run, conservative in what it claims, and completely separate from anything that can call a Delete* API.
The second reason is enumeration is harder than it looks. DescribeVolumes pages; DescribeAddresses does not. DescribeSnapshots without an owner filter returns every publicly shared snapshot in the region, which is tens of thousands of records you did not want. Half-finished audit scripts fail at exactly this layer — they read page one, report forty findings, and quietly miss the other nine hundred.
Third, "unused" is a claim about the future dressed up as a measurement of the past. A volume with no I/O for ninety days is not proof that nobody plans to reattach it. The best a program can do is gather independent evidence, state how confident it is, and hand the judgement to whoever owns the workload.
Prerequisites
- Python 3.9+ with
boto3>=1.34and, for editor and CI type checking,boto3-stubs[ec2,cloudwatch,logs,elbv2] - A read-only IAM role for the auditor:
ec2:Describe*,elasticloadbalancing:Describe*,logs:DescribeLogGroups,cloudwatch:GetMetricData,tag:GetResources. No delete permissions of any kind. - Read access to your IaC state — a Pulumi backend you can
pulumi stack exportfrom, or the Terraform state that CDKTF writes - An owner tag convention already in place, or a plan to introduce one
# CLI: confirm the auditor role really cannot delete anything
aws sts get-caller-identity --profile audit-readonly
aws ec2 delete-volume --volume-id vol-0000000000000dead --profile audit-readonly
# expected: An error occurred (UnauthorizedOperation) when calling the DeleteVolume operationImplementation
The auditor splits into three stages that are worth keeping in separate modules: collection, scoring, and correlation. Collection talks only to the cloud APIs. Scoring adds evidence. Correlation subtracts everything your infrastructure code owns.
1. Page every API into one candidate record type
Normalise each resource kind into a single frozen dataclass so later stages never care which API a record came from. Pagination is the part to get right: use get_paginator wherever boto3 offers one, and note explicitly in the code where it does not.
# audit/collect.py — normalise five different APIs into one typed record
# CLI: python -m audit.collect --profile audit-readonly --region eu-west-1
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Iterator, Literal
import boto3
from botocore.config import Config
ResourceKind = Literal["ebs-volume", "ebs-snapshot", "elastic-ip", "load-balancer", "log-group"]
# Provider note: adaptive retries absorb the RequestLimitExceeded you will hit
# paging DescribeSnapshots across an account with years of backup history.
RETRY = Config(retries={"max_attempts": 10, "mode": "adaptive"})
@dataclass(frozen=True)
class Candidate:
kind: ResourceKind
identifier: str # vol-…, snap-…, eipalloc-…, ARN, or log group name
region: str
created: datetime | None
monthly_usd: float
tags: dict[str, str] = field(default_factory=dict)
facts: dict[str, str] = field(default_factory=dict)
def unattached_volumes(session: boto3.Session, region: str) -> Iterator[Candidate]:
ec2 = session.client("ec2", region_name=region, config=RETRY)
pages = ec2.get_paginator("describe_volumes").paginate(
Filters=[{"Name": "status", "Values": ["available"]}],
PaginationConfig={"PageSize": 500},
)
for page in pages:
for vol in page["Volumes"]:
yield Candidate(
kind="ebs-volume",
identifier=vol["VolumeId"],
region=region,
created=vol["CreateTime"],
monthly_usd=round(vol["Size"] * 0.08, 2),
tags={t["Key"]: t["Value"] for t in vol.get("Tags", [])},
facts={"size_gib": str(vol["Size"]), "volume_type": vol["VolumeType"]},
)
def unassociated_addresses(session: boto3.Session, region: str) -> Iterator[Candidate]:
ec2 = session.client("ec2", region_name=region, config=RETRY)
# Provider note: DescribeAddresses has NO paginator — one response holds every
# address in the region, and there is no CreateTime field to age against.
for addr in ec2.describe_addresses()["Addresses"]:
if addr.get("AssociationId"):
continue
yield Candidate(
kind="elastic-ip",
identifier=addr["AllocationId"],
region=region,
created=None,
monthly_usd=3.60,
tags={t["Key"]: t["Value"] for t in addr.get("Tags", [])},
facts={"public_ip": addr["PublicIp"], "domain": addr["Domain"]},
)
def self_owned_snapshots(session: boto3.Session, region: str) -> Iterator[Candidate]:
ec2 = session.client("ec2", region_name=region, config=RETRY)
# Provider note: OwnerIds=["self"] is mandatory. Without it DescribeSnapshots
# returns every public snapshot in the region and the paginator never ends.
pages = ec2.get_paginator("describe_snapshots").paginate(
OwnerIds=["self"], PaginationConfig={"PageSize": 1000},
)
for page in pages:
for snap in page["Snapshots"]:
yield Candidate(
kind="ebs-snapshot",
identifier=snap["SnapshotId"],
region=region,
created=snap["StartTime"],
monthly_usd=round(snap["VolumeSize"] * 0.05, 2),
tags={t["Key"]: t["Value"] for t in snap.get("Tags", [])},
facts={"source_volume": snap["VolumeId"], "description": snap.get("Description", "")},
)
def empty_log_groups(session: boto3.Session, region: str) -> Iterator[Candidate]:
logs = session.client("logs", region_name=region, config=RETRY)
for page in logs.get_paginator("describe_log_groups").paginate():
for group in page["logGroups"]:
if group.get("storedBytes", 0) > 0:
continue
yield Candidate(
kind="log-group",
identifier=group["logGroupName"],
region=region,
created=datetime.fromtimestamp(group["creationTime"] / 1000, tz=timezone.utc),
monthly_usd=0.0,
facts={"retention_days": str(group.get("retentionInDays", "never-expires"))},
)Yielding rather than returning lists matters at scale: an account with 40,000 snapshots should not need all of them resident before the first one is scored.
2. Require agreement between independent signals
Three signals are available, and each on its own produces false positives. Attachment state is a point-in-time fact — a volume detached ninety seconds ago during a rolling replacement looks identical to one detached in 2023. Creation age is only a proxy; a two-year-old volume can be a live cold archive. CloudWatch activity is the strongest of the three and also the most misread.
The misreading is this: RequestCount on AWS/ApplicationELB is a sparse metric. A load balancer that serves nothing publishes no datapoints at all, so zero datapoints and never instrumented are indistinguishable from the API response. GetMetricData will happily return an empty Values list for a metric name you typed wrong. Treat an empty result as "no evidence", not as "evidence of zero".
# audit/signals.py — score a candidate; never let one signal decide alone
# CLI: python -m audit.signals --window-days 45 --min-age-days 30
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Literal
import boto3
from audit.collect import Candidate
@dataclass(frozen=True)
class Verdict:
candidate: Candidate
age_days: int | None
datapoints: int
confidence: Literal["high", "medium", "unknown"]
reason: str
def alb_request_datapoints(
session: boto3.Session, region: str, lb_dimension: str, window_days: int = 45,
) -> int:
"""lb_dimension is the ARN suffix, e.g. 'app/checkout-alb/50dc6c495c0c9188'."""
cw = session.client("cloudwatch", region_name=region)
end = datetime.now(tz=timezone.utc)
response = cw.get_metric_data(
MetricDataQueries=[{
"Id": "requests",
"MetricStat": {
"Metric": {
"Namespace": "AWS/ApplicationELB",
"MetricName": "RequestCount",
"Dimensions": [{"Name": "LoadBalancer", "Value": lb_dimension}],
},
"Period": 86_400,
"Stat": "Sum",
},
"ReturnData": True,
}],
StartTime=end - timedelta(days=window_days),
EndTime=end,
ScanBy="TimestampDescending",
)
# Provider note: an empty Values list means "no published datapoints", which is
# NOT the same as a measured zero. RequestCount is sparse on idle balancers.
return len(response["MetricDataResults"][0]["Values"])
def score(candidate: Candidate, datapoints: int, min_age_days: int = 30) -> Verdict:
now = datetime.now(tz=timezone.utc)
age = (now - candidate.created).days if candidate.created else None
if age is None:
return Verdict(candidate, None, datapoints, "unknown",
"no creation timestamp available from the API")
if age < min_age_days:
return Verdict(candidate, age, datapoints, "unknown",
f"only {age}d old — likely mid-deployment churn")
if datapoints > 0:
return Verdict(candidate, age, datapoints, "unknown",
f"{datapoints} activity datapoints in the window")
if candidate.tags.get("owner"):
return Verdict(candidate, age, datapoints, "medium",
"idle and aged, but an owner tag exists — ask before acting")
return Verdict(candidate, age, datapoints, "high",
f"idle {age}d, no activity datapoints, no owner tag")Note what the function never does: return True. It returns a confidence level and a sentence explaining it, because that sentence is what ends up in the report and in the conversation with the team that owns the workload.
3. Subtract everything your infrastructure code manages
This is the stage that turns a dangerous script into a safe one. A resource that appears in stack state is not an orphan — it is a resource whose deletion belongs in a pull request against the program, not in a cleanup job. Deleting it out of band creates drift that the next deployment will either recreate or fail on, a failure mode covered in depth in detecting and remediating state drift.
# audit/correlate.py — drop any candidate a stack already owns
# CLI: python -m audit.correlate --stacks prod,staging,shared --report findings.json
from __future__ import annotations
import json
import subprocess
from typing import Iterable
from audit.signals import Verdict
def pulumi_managed_ids(stack: str) -> set[str]:
raw = subprocess.run(
["pulumi", "stack", "export", "--stack", stack],
capture_output=True, text=True, check=True,
).stdout
document = json.loads(raw)
# State implication: `pulumi stack export` reads the checkpoint, not the cloud.
# A resource deleted from the program is already absent here, which is exactly
# why it will show up as an orphan if the real deletion silently failed.
managed: set[str] = set()
for resource in document["deployment"].get("resources", []):
physical_id = resource.get("id")
if physical_id:
managed.add(physical_id)
return managed
def terraform_managed_ids(state_path: str) -> set[str]:
with open(state_path, encoding="utf-8") as handle:
state = json.load(handle)
return {
instance["attributes"]["id"]
for resource in state.get("resources", [])
for instance in resource.get("instances", [])
if "id" in instance.get("attributes", {})
}
def unmanaged(verdicts: Iterable[Verdict], managed: set[str]) -> list[Verdict]:
return [v for v in verdicts if v.candidate.identifier not in managed]The union has to cover every stack that can create resources in the account. If one stack's backend is unreachable, the correct behaviour is to abort the run, not to report a longer list. An incomplete managed set makes managed resources look like orphans, which is precisely the mistake you built this stage to prevent.
Verification
Run the auditor read-only and check three things: the enumeration is complete, the correlation actually removed managed resources, and the costs are in the right order of magnitude.
# CLI: enumeration completeness — the auditor's count must match a raw API count
aws ec2 describe-volumes --filters Name=status,Values=available \
--query 'length(Volumes)' --profile audit-readonly
# CLI: run the audit and inspect the report
python -m audit.correlate --stacks prod,staging,shared --report findings.json
jq '[.findings[] | select(.confidence=="high")] | length, (map(.monthly_usd) | add)' findings.json
# CLI: prove correlation works — pick a volume you know a stack manages
jq -r '.findings[].identifier' findings.json | grep vol-0abc1234def567890 || echo "correctly excluded"If the high-confidence total is a large fraction of your monthly bill, the scoring is too loose. Realistic first runs on a mature account surface single-digit percentages.
Gotchas & Edge Cases
Snapshots backing an AMI are not orphans. Deleting one returns InvalidSnapshot.InUse: The snapshot snap-0123456789abcdef0 is currently in use by ami-0fedcba9876543210. Build the exclusion set first with describe_images(Owners=["self"]) and walk BlockDeviceMappings[].Ebs.SnapshotId.
Opt-in regions are invisible by default. Iterate ec2.describe_regions(AllRegions=True) and check OptInStatus; a region that is not-opted-in cannot be audited, and one that is opted in but never used still needs a pass.
storedBytes on a log group updates on a delay. A group that received its first events minutes ago can still report zero. Combine it with creationTime and never treat a young group as empty.
Deleting an active empty log group is harmless but lossy. The next Lambda invocation recreates it — without your retention setting, which silently reverts to never-expire and starts costing money again.
An Elastic IP attached to a network interface still shows an AssociationId. Only addresses with no association are billed as idle, but an address associated with a detached interface is billed too and will not appear in your findings.
Operational Notes
Reporting is not the end of the workflow; it is the start of one. The loop below is deliberately slow, and the slowness is the safety mechanism.
Tag first, delete much later. Writing lifecycle/status = candidate-for-deletion and lifecycle/reviewed = 2026-08-02 onto a resource is reversible, visible in the console to whoever stumbles on it, and gives an owner something to object to. Give the objection window at least fourteen days and announce it in the channel where the owning team actually reads messages.
For anything holding data, snapshot before deleting, and tag the snapshot with the identifier of what it replaced. A 500 GiB volume costs forty dollars a month; its snapshot costs twenty-five and can be discarded after a quarter. That is a cheap insurance premium against the one deletion you get wrong.
Finally, feed the outcome back into policy. If the audit keeps finding untagged volumes from the same pipeline, the fix is a tagging rule enforced at deployment time — see enforcing tagging policies with Pulumi CrossGuard — not a monthly cleanup ritual.
FAQ
How long should a volume be unattached before I delete it?
Thirty days is a defensible floor because it clears monthly batch jobs and most rolling-deployment churn. Extend to ninety for anything tagged as belonging to a data or analytics team, where quarterly reprocessing is normal.
Can the auditor run inside the same process as my Pulumi program?
It can, using the same session-handling patterns described in using boto3 inside Pulumi and CDKTF, but keep it in a separate scheduled job. Mixing a read-only audit into a deployment path means an audit failure blocks a release.
Why not just delete everything without an owner tag?
Because tag coverage is a measure of your tagging discipline, not of resource usefulness. Untagged resources are usually the oldest ones, which correlates with importance as often as with abandonment.
How do I audit dozens of accounts?
Assume the same read-only role name in each account via sts:AssumeRole, run the collectors per account and region, and write one combined report keyed by account ID. Keep concurrency modest — CloudWatch GetMetricData throttles well before the Describe APIs do.
Does this replace Cost Explorer?
No. Cost Explorer tells you which service is expensive; this auditor tells you which specific resource inside that service is unused. Use the first to decide where to point the second.
Related
- IaC Cost and Governance — the parent topic covering spend control and policy for Python infrastructure code.
- Detecting and Remediating State Drift in Python IaC — what happens when out-of-band deletion meets a stack that still expects the resource.
- Enforcing Tagging Policies with Pulumi CrossGuard — stop untagged resources at deploy time so audits get shorter.