scripts/wa_review.py
#!/usr/bin/env python3
"""DocumentDB Well-Architected Review — standalone CLI.
Runs 41 automated checks across 6 pillars against a DocumentDB cluster.
Infrastructure checks use boto3 (AWS APIs). Database-level checks use
a pre-collected analysis JSON file (from pymongo collStats/indexStats).
Usage:
python3 wa_review.py --cluster-id <id> --region <region> [--analysis-data <path>]
Output:
wa_review_results.json — structured check results
wa_review_report.md — human-readable summary
Requires: boto3, AWS credentials with docdb/cloudwatch/ec2/secretsmanager read access.
"""
import argparse
import json
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
try:
import boto3
except ImportError:
print("ERROR: boto3 required. Install with: pip install boto3")
sys.exit(1)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
CONN_LIMITS = {
"db.t3.medium": 1000,
"db.t4g.medium": 1000,
"db.r5.large": 3400,
"db.r6g.large": 3400,
"db.r6gd.large": 3400,
"db.r8g.large": 3400,
"db.r5.xlarge": 7000,
"db.r6g.xlarge": 7000,
"db.r6gd.xlarge": 7000,
"db.r8g.xlarge": 7000,
"db.r5.2xlarge": 14200,
"db.r6g.2xlarge": 14200,
"db.r6gd.2xlarge": 14200,
"db.r8g.2xlarge": 14200,
"db.r5.4xlarge": 28400,
"db.r6g.4xlarge": 28400,
"db.r6gd.4xlarge": 28400,
"db.r8g.4xlarge": 28400,
"db.r5.8xlarge": 60000,
"db.r6g.8xlarge": 60000,
"db.r6gd.8xlarge": 60000,
"db.r8g.8xlarge": 60000,
"db.r5.12xlarge": 60000,
"db.r6g.12xlarge": 60000,
"db.r6gd.12xlarge": 60000,
"db.r8g.12xlarge": 60000,
"db.r5.16xlarge": 60000,
"db.r6g.16xlarge": 60000,
"db.r6gd.16xlarge": 60000,
"db.r8g.16xlarge": 60000,
"db.r5.24xlarge": 60000,
}
INSTANCE_RAM_GIB = {
"db.t3.medium": 4,
"db.t4g.medium": 4,
"db.r5.large": 16,
"db.r6g.large": 16,
"db.r6gd.large": 16,
"db.r8g.large": 16,
"db.r5.xlarge": 32,
"db.r6g.xlarge": 32,
"db.r6gd.xlarge": 32,
"db.r8g.xlarge": 32,
"db.r5.2xlarge": 64,
"db.r6g.2xlarge": 64,
"db.r6gd.2xlarge": 64,
"db.r8g.2xlarge": 64,
"db.r5.4xlarge": 128,
"db.r6g.4xlarge": 128,
"db.r6gd.4xlarge": 128,
"db.r8g.4xlarge": 128,
"db.r5.8xlarge": 256,
"db.r6g.8xlarge": 256,
"db.r6gd.8xlarge": 256,
"db.r8g.8xlarge": 256,
"db.r5.12xlarge": 384,
"db.r6g.12xlarge": 384,
"db.r6gd.12xlarge": 384,
"db.r8g.12xlarge": 384,
"db.r5.16xlarge": 512,
"db.r6g.16xlarge": 512,
"db.r6gd.16xlarge": 512,
"db.r8g.16xlarge": 512,
"db.r5.24xlarge": 768,
}
GRAVITON_FAMILIES = ("r6g", "r7g", "r8g", "t4g", "r6gd")
def _add(results, pillar, check_id, label, status, detail=""):
results.append(
{"pillar": pillar, "id": check_id, "label": label, "status": status, "detail": detail}
)
# ---------------------------------------------------------------------------
# Infrastructure checks (boto3)
# ---------------------------------------------------------------------------
def run_infra_checks(cluster_id, region):
results: list[dict] = []
docdb = boto3.client("docdb", region_name=region)
cw = boto3.client("cloudwatch", region_name=region)
ec2 = boto3.client("ec2", region_name=region)
try:
cl = docdb.describe_db_clusters(DBClusterIdentifier=cluster_id)["DBClusters"][0]
except Exception as e:
_add(results, "Other", "ERR", f"Cannot describe cluster: {e}", "fail")
return results
try:
insts = docdb.describe_db_instances(
Filters=[{"Name": "db-cluster-id", "Values": [cluster_id]}]
)["DBInstances"]
except Exception as e:
insts = []
_add(results, "Other", "ERR", f"Cannot describe instances: {e}", "fail")
# -- RELIABILITY --------------------------------------------------------
retention = cl.get("BackupRetentionPeriod", 1)
_add(
results,
"Reliability",
"REL1",
f"Backup retention period ({retention} days)",
"pass" if retention >= 7 else "warn" if retention >= 3 else "fail",
"Recommended: 7+ days for production" if retention < 7 else "",
)
del_prot = cl.get("DeletionProtection", False)
_add(
results,
"Reliability",
"REL2",
f"Deletion protection ({'enabled' if del_prot else 'disabled'})",
"pass" if del_prot else "fail",
"" if del_prot else "Enable deletion protection for production clusters",
)
n_inst = len(insts)
_add(
results,
"Reliability",
"REL5a",
f"Instance count ({n_inst})",
"pass" if n_inst >= 2 else "fail",
"Minimum 2 instances required for auto failover" if n_inst < 2 else "",
)
azs = {i.get("AvailabilityZone", "") for i in insts}
_add(
results,
"Reliability",
"REL5b",
f"Instances across {len(azs)} AZ(s)",
"pass" if len(azs) >= 2 else "fail",
"Single AZ -- no failover protection" if len(azs) < 2 else "",
)
engine_ver = cl.get("EngineVersion", "unknown")
major = engine_ver.split(".")[0] if engine_ver != "unknown" else ""
if major in ("3", "4"):
_add(
results,
"Reliability",
"REL6",
f"Engine version {engine_ver} (approaching or past end-of-life)",
"fail",
"Upgrade to DocumentDB 5.0 or 8.0",
)
elif major == "5":
_add(
results,
"Reliability",
"REL6",
f"Engine version {engine_ver}",
"pass",
"Consider upgrading to 8.0 for Zstandard compression and Query Planner v3",
)
else:
_add(results, "Reliability", "REL6", f"Engine version {engine_ver}", "pass")
# Recent failover events (14 days) — paginated
try:
evt_end = datetime.now(timezone.utc)
evt_start = evt_end - timedelta(days=13)
events_list = []
paginator = docdb.get_paginator("describe_events")
for page in paginator.paginate(
SourceIdentifier=cluster_id,
SourceType="db-cluster",
StartTime=evt_start,
EndTime=evt_end,
):
events_list.extend(page.get("Events", []))
failover_events = [
e
for e in events_list
if "failover" in e.get("Message", "").lower()
or "failover" in ",".join(e.get("EventCategories", [])).lower()
]
if failover_events:
_add(
results,
"Reliability",
"REL7",
f"{len(failover_events)} failover event(s) in last 13 days",
"warn",
f"Most recent: {failover_events[-1].get('Message', '')[:120]}",
)
else:
_add(results, "Reliability", "REL7", "No failover events in last 13 days", "pass")
except Exception as e:
_add(results, "Reliability", "REL7", f"Cannot check events: {e}", "warn")
# -- SECURITY -----------------------------------------------------------
encrypted = cl.get("StorageEncrypted", False)
_add(
results,
"Security",
"SEC1a",
f"Encryption at rest ({'enabled' if encrypted else 'disabled'})",
"pass" if encrypted else "fail",
"" if encrypted else "Enable encryption at rest (requires new cluster)",
)
tls_val = "unknown"
try:
pg_name = cl.get("DBClusterParameterGroup", "")
if pg_name:
params = docdb.describe_db_cluster_parameters(DBClusterParameterGroupName=pg_name).get(
"Parameters", []
)
for p in params:
if p.get("ParameterName") == "tls":
tls_val = p.get("ParameterValue", "enabled")
elif p.get("ParameterName") == "tls_version":
tv = p.get("ParameterValue", "")
if tv and "1.2" in tv and "1.0" not in tv and "1.1" not in tv:
_add(results, "Security", "SEC6", f"TLS minimum version: {tv}", "pass")
elif tv:
_add(
results,
"Security",
"SEC6",
f"TLS minimum version: {tv}",
"warn",
"Set tls_version to TLSv1.2 to disable older protocols",
)
except Exception as e:
_add(results, "Security", "SEC6", f"Cannot check TLS parameters: {e}", "warn")
# Ensure SEC6 is always present
if not any(r["id"] == "SEC6" for r in results):
_add(
results,
"Security",
"SEC6",
"TLS minimum version: unknown",
"warn",
"Could not determine tls_version parameter",
)
_add(
results,
"Security",
"SEC1b",
f"TLS ({tls_val})",
"pass" if tls_val == "enabled" else "warn" if tls_val == "unknown" else "fail",
(
"Could not determine TLS status"
if tls_val == "unknown"
else ("" if tls_val == "enabled" else "TLS should be enabled")
),
)
# Security groups
sg_open = False
sg_checked = 0
for vsg in cl.get("VpcSecurityGroups", []):
sg_id = vsg.get("VpcSecurityGroupId", "")
if not sg_id:
continue
try:
sg_detail = ec2.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0]
sg_checked += 1
for rule in sg_detail.get("IpPermissions", []):
for ip_range in rule.get("IpRanges", []):
if ip_range.get("CidrIp") == "0.0.0.0/0":
sg_open = True
_add(
results,
"Security",
"SEC2",
f"Security group {sg_id} open to 0.0.0.0/0",
"fail",
"Restrict to specific CIDR ranges",
)
for ip_range in rule.get("Ipv6Ranges", []):
if ip_range.get("CidrIpv6") == "::/0":
sg_open = True
_add(
results,
"Security",
"SEC2",
f"Security group {sg_id} open to ::/0",
"fail",
"Restrict to specific CIDR ranges",
)
except Exception as e:
_add(results, "Security", "SEC2", f"Cannot check SG {sg_id}: {e}", "warn")
if not sg_open and sg_checked > 0:
_add(
results,
"Security",
"SEC2",
f"Security groups properly restricted ({sg_checked} checked)",
"pass",
)
logs = cl.get("EnabledCloudwatchLogsExports", [])
audit_enabled = "audit" in logs
profiler_enabled = "profiler" in logs
_add(
results,
"Security",
"SEC5",
f"Audit logging ({'enabled' if audit_enabled else 'disabled'})",
"pass" if audit_enabled else "warn",
"" if audit_enabled else "Enable audit logging for compliance",
)
# Secrets Manager
try:
sm = boto3.client("secretsmanager", region_name=region)
found_secret = False
for page in sm.get_paginator("list_secrets").paginate():
for s in page.get("SecretList", []):
name = (s.get("Name", "") or "").lower()
desc = (s.get("Description", "") or "").lower()
if cluster_id.lower() in name or cluster_id.lower() in desc:
found_secret = True
break
if found_secret:
break
_add(
results,
"Security",
"SEC3",
f"Secrets Manager {'references' if found_secret else 'does not reference'} this cluster",
"pass" if found_secret else "warn",
"" if found_secret else "Store credentials in Secrets Manager",
)
except Exception as e:
_add(results, "Security", "SEC3", f"Cannot check Secrets Manager: {e}", "warn")
# -- OPERATIONAL EXCELLENCE ---------------------------------------------
sg_name = cl.get("DBSubnetGroup", "")
try:
if sg_name:
sg = docdb.describe_db_subnet_groups(DBSubnetGroupName=sg_name)["DBSubnetGroups"][0]
sg_azs = {s["SubnetAvailabilityZone"]["Name"] for s in sg.get("Subnets", [])}
_add(
results,
"Operational Excellence",
"OPS2",
f"Subnet group spans {len(sg_azs)} AZ(s)",
"pass" if len(sg_azs) >= 3 else "warn",
"Recommended: 3 AZs for failover flexibility" if len(sg_azs) < 3 else "",
)
except Exception as e:
_add(results, "Operational Excellence", "OPS2", f"Cannot check subnet group: {e}", "warn")
_add(
results,
"Operational Excellence",
"OPS5a",
f"Profiler logging ({'enabled' if profiler_enabled else 'disabled'})",
"pass" if profiler_enabled else "warn",
"" if profiler_enabled else "Enable profiler for slow query analysis",
)
pg_name = cl.get("DBClusterParameterGroup", "")
_add(
results,
"Operational Excellence",
"OPS5c",
f"Parameter group: {pg_name}",
"warn" if pg_name.startswith("default.") else "pass",
(
"Use a custom parameter group for workload-specific tuning"
if pg_name.startswith("default.")
else ""
),
)
_add(
results,
"Operational Excellence",
"OPS7",
f"Maintenance window: {cl.get('PreferredMaintenanceWindow', 'not set')}",
"info",
"Verify this window aligns with your lowest-traffic period",
)
try:
n_alarms = len(cw.describe_alarms(AlarmNamePrefix=cluster_id).get("MetricAlarms", []))
_add(
results,
"Operational Excellence",
"OPS5b",
f"CloudWatch alarms ({n_alarms} configured)",
"pass" if n_alarms >= 3 else "warn" if n_alarms > 0 else "fail",
(
"Recommended: alarms for CPU, FreeableMemory, DatabaseConnections"
if n_alarms < 3
else ""
),
)
except Exception as e:
_add(results, "Operational Excellence", "OPS5b", f"Cannot check alarms: {e}", "warn")
# -- COST OPTIMIZATION --------------------------------------------------
try:
n_tags = len(
docdb.list_tags_for_resource(ResourceName=cl["DBClusterArn"]).get("TagList", [])
)
_add(
results,
"Cost Optimization",
"COST6",
f"Cost allocation tags ({n_tags} tags)",
"pass" if n_tags >= 2 else "warn",
"Add cost allocation tags for expense tracking" if n_tags < 2 else "",
)
except Exception as e:
_add(results, "Cost Optimization", "COST6", f"Cannot check tags: {e}", "warn")
storage_type = cl.get("StorageType", "standard")
_add(
results,
"Cost Optimization",
"COST7",
f"Storage type: {storage_type}",
"info",
(
"Evaluate I/O-Optimized for write-heavy workloads"
if storage_type != "iopt1"
else "I/O-Optimized active -- no per-I/O charges"
),
)
# -- PER-INSTANCE CHECKS ------------------------------------------------
end = datetime.now(timezone.utc)
start = end - timedelta(days=7)
for inst in insts:
iid = inst["DBInstanceIdentifier"]
itype = inst["DBInstanceClass"]
dim = [{"Name": "DBInstanceIdentifier", "Value": iid}]
is_writer = inst.get("IsClusterWriter", False)
family = itype.replace("db.", "").split(".")[0] if itype.startswith("db.") else ""
# CPU — use hourly Maximum for P95 to capture peak usage within each hour
try:
raw_dps = cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="CPUUtilization",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average", "Maximum"],
).get("Datapoints", [])
if raw_dps:
avg_cpu = sum(d["Average"] for d in raw_dps) / len(raw_dps)
max_vals = sorted(d["Maximum"] for d in raw_dps)
p95_cpu = max_vals[int(len(max_vals) * 0.95)]
_add(
results,
"Cost Optimization",
"COST1",
f"CPU for {iid} (avg {avg_cpu:.1f}%, P95 {p95_cpu:.1f}%)",
"warn" if p95_cpu < 10 else "pass",
f"Instance {itype} may be oversized" if p95_cpu < 10 else "",
)
except Exception as e:
_add(results, "Cost Optimization", "COST1", f"Cannot check CPU for {iid}: {e}", "warn")
# Graviton
_add(
results,
"Sustainability",
"SUST1",
f"{iid} {'uses' if family in GRAVITON_FAMILIES else 'does not use'} Graviton ({itype})",
"pass" if family in GRAVITON_FAMILIES else "warn",
(
""
if family in GRAVITON_FAMILIES
else "Migrate to Graviton (r6g/r8g) for better price-performance"
),
)
# Buffer cache hit ratio
try:
dps = [
d["Average"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="BufferCacheHitRatio",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average"],
).get("Datapoints", [])
]
if dps:
avg_cache = sum(dps) / len(dps)
_add(
results,
"Performance Efficiency",
"PERF6",
f"Buffer cache hit ratio for {iid} ({avg_cache:.1f}%)",
"pass" if avg_cache >= 99 else "warn" if avg_cache >= 95 else "fail",
"Working set may not fit in memory" if avg_cache < 95 else "",
)
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF6",
f"Cannot check cache for {iid}: {e}",
"warn",
)
# Connections vs limits
try:
dps = [
d["Maximum"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="DatabaseConnections",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Maximum"],
).get("Datapoints", [])
]
limit = CONN_LIMITS.get(itype, 0)
if dps and limit:
max_conn = max(dps)
pct = max_conn / limit * 100
_add(
results,
"Performance Efficiency",
"PERF5",
f"Peak connections for {iid} ({int(max_conn)}/{limit} = {pct:.0f}%)",
"pass" if pct < 70 else "warn" if pct < 90 else "fail",
"Consider upsizing or connection pooling" if pct >= 70 else "",
)
elif dps:
_add(
results,
"Performance Efficiency",
"PERF5",
f"Connection limit unknown for {iid} ({itype})",
"warn",
"Instance type not in lookup table",
)
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF5",
f"Cannot check connections for {iid}: {e}",
"warn",
)
# Idle reader detection
if not is_writer:
try:
conn_dps = [
d["Average"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="DatabaseConnections",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average"],
).get("Datapoints", [])
]
io_dps = [
d["Average"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="ReadIOPS",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average"],
).get("Datapoints", [])
]
avg_conn = sum(conn_dps) / len(conn_dps) if conn_dps else 0
avg_iops = sum(io_dps) / len(io_dps) if io_dps else 0
if avg_conn < 2 and avg_iops < 5:
_add(
results,
"Cost Optimization",
"COST9",
f"Reader {iid} appears idle (avg {avg_conn:.0f} conn, {avg_iops:.0f} ReadIOPS)",
"warn",
"Consider removing this replica to reduce cost",
)
else:
_add(
results,
"Cost Optimization",
"COST9",
f"Reader {iid} is active (avg {avg_conn:.0f} conn, {avg_iops:.0f} ReadIOPS)",
"pass",
)
except Exception as e:
_add(
results, "Cost Optimization", "COST9", f"Cannot check reader {iid}: {e}", "warn"
)
# FreeableMemory
ram_gib = INSTANCE_RAM_GIB.get(itype, 0)
if ram_gib:
try:
dps = [
d["Minimum"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="FreeableMemory",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Minimum"],
).get("Datapoints", [])
]
if dps:
min_free = min(dps)
free_pct = min_free / (ram_gib * 1024**3) * 100
_add(
results,
"Performance Efficiency",
"PERF11",
f"FreeableMemory min for {iid}: {min_free / (1024**3):.1f} GiB ({free_pct:.0f}%)",
"fail" if free_pct < 5 else "warn" if free_pct < 10 else "pass",
"Instance under memory pressure" if free_pct < 10 else "",
)
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF11",
f"Cannot check memory for {iid}: {e}",
"warn",
)
else:
_add(
results,
"Performance Efficiency",
"PERF11",
f"Unknown instance type {itype} -- cannot check FreeableMemory",
"warn",
)
# SwapUsage
try:
dps = [
d["Maximum"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="SwapUsage",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Maximum"],
).get("Datapoints", [])
]
if dps and max(dps) > 0:
_add(
results,
"Performance Efficiency",
"PERF12",
f"SwapUsage max for {iid}: {max(dps) / (1024**2):.0f} MB",
"fail",
"Instance is swapping -- critically undersized",
)
elif dps:
_add(results, "Performance Efficiency", "PERF12", f"No swap on {iid}", "pass")
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF12",
f"Cannot check swap for {iid}: {e}",
"warn",
)
# DiskQueueDepth
try:
dps = [
d["Average"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="DiskQueueDepth",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average"],
).get("Datapoints", [])
]
if dps:
avg_dqd = sum(dps) / len(dps)
_add(
results,
"Performance Efficiency",
"PERF13",
f"DiskQueueDepth avg for {iid}: {avg_dqd:.1f}",
"warn" if avg_dqd > 5 else "pass",
"I/O backing up -- evaluate I/O-Optimized or upsizing" if avg_dqd > 5 else "",
)
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF13",
f"Cannot check DiskQueueDepth for {iid}: {e}",
"warn",
)
# IndexBufferCacheHitRatio
try:
dps = [
d["Average"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="IndexBufferCacheHitRatio",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Average"],
).get("Datapoints", [])
]
if dps:
avg_idx = sum(dps) / len(dps)
_add(
results,
"Performance Efficiency",
"PERF14",
f"IndexBufferCacheHitRatio for {iid}: {avg_idx:.1f}%",
"pass" if avg_idx >= 99 else "warn" if avg_idx >= 95 else "fail",
"Indexes do not fit in memory" if avg_idx < 95 else "",
)
except Exception as e:
_add(
results,
"Performance Efficiency",
"PERF14",
f"Cannot check index cache for {iid}: {e}",
"warn",
)
# DatabaseCursorsTimedOut
try:
dps = [
d["Sum"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="DatabaseCursorsTimedOut",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=86400,
Statistics=["Sum"],
).get("Datapoints", [])
]
total = sum(dps) if dps else 0
if total > 0:
_add(
results,
"Reliability",
"REL8",
f"{int(total)} cursor(s) timed out on {iid} in last 7 days",
"warn",
"Application may not be closing cursors properly",
)
else:
_add(results, "Reliability", "REL8", f"No cursor timeouts on {iid}", "pass")
except Exception as e:
_add(results, "Reliability", "REL8", f"Cannot check cursors for {iid}: {e}", "warn")
# AvailableMVCCIds (writer only)
if is_writer:
try:
dps = [
d["Minimum"]
for d in cw.get_metric_statistics(
Namespace="AWS/DocDB",
MetricName="AvailableMVCCIds",
Dimensions=dim,
StartTime=start,
EndTime=end,
Period=3600,
Statistics=["Minimum"],
).get("Datapoints", [])
]
if dps:
min_mvcc = min(dps)
pct = min_mvcc / 1_400_000_000 * 100
_add(
results,
"Reliability",
"REL9",
f"AvailableMVCCIds min: {min_mvcc:,.0f} ({pct:.0f}%)",
"fail" if pct < 25 else "warn" if pct < 50 else "pass",
(
"MVCC ID exhaustion risk -- investigate long-running transactions"
if pct < 50
else ""
),
)
except Exception as e:
_add(results, "Reliability", "REL9", f"Cannot check MVCCIds: {e}", "warn")
return results
# ---------------------------------------------------------------------------
# Database-level checks (from pre-collected analysis JSON)
# ---------------------------------------------------------------------------
def run_db_checks(analysis_data):
results: list[dict] = []
if not analysis_data:
return results
total_indexes = 0
unused_indexes = 0
redundant = 0
low_cardinality = 0
low_card_names = []
large_docs = []
ttl_colls = []
total_data_size = 0
total_index_size = 0
total_unused_bytes = 0
bloated_colls = []
over_indexed_colls = []
compression_disabled = []
collscan_candidates = []
write_amp_colls = []
for db_name, collections in analysis_data.items():
if not isinstance(collections, dict):
continue
for coll_name, stats in collections.items():
if not isinstance(stats, dict) or "error" in stats:
continue
indexes = stats.get("indexes", [])
total_indexes += len(indexes)
for idx in indexes:
if idx.get("usage", {}).get("potential_unused"):
unused_indexes += 1
if idx.get("cardinality", {}).get("is_low"):
low_cardinality += 1
low_card_names.append(f"{db_name}.{coll_name}.{idx['name']}")
if idx.get("expireAfterSeconds") is not None:
if f"{db_name}.{coll_name}" not in ttl_colls:
ttl_colls.append(f"{db_name}.{coll_name}")
avg_obj = stats.get("avgObjSize", 0)
if avg_obj > 8192:
large_docs.append(f"{db_name}.{coll_name} ({avg_obj:,} bytes)")
# Redundant indexes (prefix subset)
ordered = [tuple(idx.get("ordered_fields", [])) for idx in indexes]
for i, a in enumerate(ordered):
for j, b in enumerate(ordered):
if i != j and len(a) > 0 and len(a) < len(b) and b[: len(a)] == a:
redundant += 1
break
total_data_size += stats.get("size", 0)
for idx in indexes:
total_index_size += idx.get("size", 0)
unused_info = stats.get("unusedStorageSize", {})
unused_pct = unused_info.get("unusedPercent", 0.0)
total_unused_bytes += unused_info.get("unusedBytes", 0)
if unused_pct > 30:
bloated_colls.append(f"{db_name}.{coll_name} ({unused_pct:.0f}%)")
if len(indexes) > 10:
over_indexed_colls.append(f"{db_name}.{coll_name} ({len(indexes)} indexes)")
comp = stats.get("compression", {})
if not comp.get("enabled", False):
compression_disabled.append(f"{db_name}.{coll_name}")
doc_count = stats.get("count", 0)
non_id = [idx for idx in indexes if idx.get("name") != "_id_"]
if doc_count > 100000 and len(non_id) == 0:
collscan_candidates.append(f"{db_name}.{coll_name} ({doc_count:,} docs)")
coll_data = stats.get("size", 0)
coll_idx = sum(idx.get("size", 0) for idx in indexes)
if coll_data > 0 and coll_idx > 2 * coll_data:
write_amp_colls.append(
f"{db_name}.{coll_name} (index {coll_idx / coll_data:.1f}x data)"
)
# Emit checks
if large_docs:
_add(
results,
"Performance Efficiency",
"PERF1",
f"{len(large_docs)} collection(s) with avg doc size > 8 KB",
"warn",
", ".join(large_docs[:5]),
)
else:
_add(
results,
"Performance Efficiency",
"PERF1",
"All collections have avg doc size < 8 KB",
"pass",
)
if redundant > 0:
_add(
results,
"Performance Efficiency",
"PERF1b",
f"{redundant} redundant index(es) (prefix subsets)",
"warn",
)
else:
_add(results, "Performance Efficiency", "PERF1b", "No redundant indexes detected", "pass")
if low_cardinality > 0:
_add(
results,
"Performance Efficiency",
"PERF1c",
f"{low_cardinality} low cardinality index(es)",
"warn",
", ".join(low_card_names[:5]),
)
else:
_add(
results,
"Performance Efficiency",
"PERF1c",
"No low cardinality indexes detected",
"pass",
)
if unused_indexes > 0:
_add(
results,
"Cost Optimization",
"COST3",
f"{unused_indexes} unused index(es) of {total_indexes} total",
"warn",
"Remove unused indexes to reduce write overhead and storage",
)
else:
_add(
results,
"Cost Optimization",
"COST3",
f"No unused indexes ({total_indexes} total)",
"pass",
)
if ttl_colls:
_add(
results,
"Cost Optimization",
"COST4",
f"TTL indexes on {len(ttl_colls)} collection(s)",
"pass",
", ".join(ttl_colls[:5]),
)
else:
_add(
results,
"Cost Optimization",
"COST4",
"No TTL indexes found",
"warn",
"Consider TTL indexes for automatic data expiration",
)
if total_data_size > 0:
ratio = total_index_size / total_data_size * 100
_add(
results,
"Performance Efficiency",
"PERF8",
f"Index-to-data ratio: {ratio:.0f}%",
"warn" if ratio > 50 else "pass",
"Indexes exceed 50% of data size" if ratio > 50 else "",
)
if bloated_colls:
_add(
results,
"Performance Efficiency",
"PERF9",
f"{len(bloated_colls)} collection(s) with >30% storage bloat",
"warn",
"Run compact command. " + ", ".join(bloated_colls[:5]),
)
else:
_add(results, "Performance Efficiency", "PERF9", "No significant storage bloat", "pass")
if over_indexed_colls:
_add(
results,
"Performance Efficiency",
"PERF10",
f"{len(over_indexed_colls)} collection(s) with >10 indexes",
"warn",
", ".join(over_indexed_colls[:5]),
)
else:
_add(results, "Performance Efficiency", "PERF10", "No over-indexed collections", "pass")
if compression_disabled:
_add(
results,
"Sustainability",
"SUST2",
f"Compression disabled on {len(compression_disabled)} collection(s)",
"warn",
", ".join(compression_disabled[:5]),
)
else:
_add(results, "Sustainability", "SUST2", "Compression enabled on all collections", "pass")
if collscan_candidates:
_add(
results,
"Performance Efficiency",
"PERF15",
f"{len(collscan_candidates)} large collection(s) with no secondary indexes",
"warn",
", ".join(collscan_candidates[:5]),
)
else:
_add(
results,
"Performance Efficiency",
"PERF15",
"All large collections have secondary indexes",
"pass",
)
if write_amp_colls:
_add(
results,
"Performance Efficiency",
"PERF16",
f"{len(write_amp_colls)} collection(s) with index size > 2x data",
"warn",
", ".join(write_amp_colls[:5]),
)
else:
_add(
results, "Performance Efficiency", "PERF16", "No excessive index-to-data ratio", "pass"
)
return results
# ---------------------------------------------------------------------------
# Report generation
# ---------------------------------------------------------------------------
def generate_report(results, cluster_id, output_dir):
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# JSON output
json_path = output_dir / "wa_review_results.json"
with open(json_path, "w") as f:
json.dump(
{
"cluster": cluster_id,
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"checks": results,
},
f,
indent=2,
)
# Markdown summary
md_path = output_dir / "wa_review_report.md"
pillars: dict[str, list] = {}
for r in results:
pillars.setdefault(r["pillar"], []).append(r)
counts = {"pass": 0, "warn": 0, "fail": 0, "info": 0}
for r in results:
counts[r["status"]] = counts.get(r["status"], 0) + 1
lines = [
f"# Well-Architected Review: {cluster_id}\n",
f"**Date:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}\n",
f"**Summary:** {counts['pass']} pass, {counts['warn']} warnings, "
f"{counts['fail']} failures, {counts['info']} info\n",
]
status_icon = {"pass": "[PASS]", "warn": "[WARN]", "fail": "[FAIL]", "info": "[INFO]"}
for pillar in [
"Reliability",
"Security",
"Operational Excellence",
"Cost Optimization",
"Performance Efficiency",
"Sustainability",
]:
checks = pillars.get(pillar, [])
if not checks:
continue
lines.append(f"\n## {pillar}\n")
lines.append("| Status | Check | Detail |")
lines.append("|--------|-------|--------|")
for c in checks:
icon = status_icon.get(c["status"], c["status"])
label = c.get("label", "").replace("|", "/")
detail = c.get("detail", "").replace("|", "/")
lines.append(f"| {icon} | {label} | {detail} |")
with open(md_path, "w") as f:
f.write("\n".join(lines) + "\n")
return json_path, md_path
# ---------------------------------------------------------------------------
# Collect database stats via pymongo (--uri)
# ---------------------------------------------------------------------------
def _redact_uri_creds(msg):
"""Strip any embedded mongodb credentials (user:pass@) from an error
message so they are never printed to stdout/logs."""
return re.sub(r"://[^/@\s]*@", "://<redacted>@", str(msg))
def collect_db_stats(uri, tls_ca_file=None, tls_allow_invalid_certs=False):
"""Connect to DocumentDB/MongoDB, collect collStats + indexStats for all
collections across all databases. Returns analysis_data dict compatible
with run_db_checks()."""
try:
import pymongo
except ImportError:
print("ERROR: pymongo required for --uri. Install with: pip install pymongo")
sys.exit(1)
client = pymongo.MongoClient(
uri,
serverSelectionTimeoutMS=10000,
tlsAllowInvalidCertificates=tls_allow_invalid_certs,
**({"tlsCAFile": tls_ca_file} if tls_ca_file else {}),
)
analysis: dict[str, dict] = {}
skip_dbs = {"admin", "local", "config"}
try:
for db_name in client.list_database_names():
if db_name in skip_dbs:
continue
db = client[db_name]
collections: dict[str, dict] = {}
for coll_name in db.list_collection_names():
if coll_name.startswith("system."):
continue
try:
stats = db.command("collStats", coll_name)
idx_stats = list(db[coll_name].aggregate([{"$indexStats": {}}]))
indexes: list[dict] = []
raw_indexes = list(db[coll_name].list_indexes())
idx_usage = {s["name"]: s.get("accesses", {}).get("ops", 0) for s in idx_stats}
for idx in raw_indexes:
name = idx["name"]
key = idx.get("key", {})
size = stats.get("indexSizes", {}).get(name, 0)
ops = idx_usage.get(name, 0)
ordered_fields = list(key.keys())
entry = {
"name": name,
"fields": key,
"ordered_fields": ordered_fields,
"size": size,
"usage": {"ops": ops, "potential_unused": ops == 0 and name != "_id_"},
"cardinality": {"is_low": False},
}
if idx.get("expireAfterSeconds") is not None:
entry["expireAfterSeconds"] = idx["expireAfterSeconds"]
indexes.append(entry)
comp_enabled = False
comp_info = stats.get("compression", {})
comp_enabled = comp_info.get("enabled", False) or comp_info.get("enable", False)
data_size = stats.get("size", 0)
storage_size = stats.get("storageSize", 0)
unused_bytes = (
max(0, storage_size - data_size) if storage_size > data_size else 0
)
unused_pct = (unused_bytes / storage_size * 100) if storage_size > 0 else 0
collections[coll_name] = {
"count": stats.get("count", 0),
"size": data_size,
"storageSize": storage_size,
"avgObjSize": stats.get("avgObjSize", 0),
"totalIndexSize": stats.get("totalIndexSize", 0),
"indexes": indexes,
"compression": {"enabled": comp_enabled},
"unusedStorageSize": {
"unusedBytes": unused_bytes,
"unusedPercent": unused_pct,
},
}
except Exception as e:
collections[coll_name] = {"error": str(e)}
if collections:
analysis[db_name] = collections
except Exception as e:
print(f" ERROR: Cannot connect to database: {_redact_uri_creds(e)}")
return {}
finally:
client.close()
return analysis
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(description="DocumentDB Well-Architected Review")
parser.add_argument("--cluster-id", required=True, help="DocumentDB cluster identifier")
parser.add_argument("--region", required=True, help="AWS region")
parser.add_argument(
"--uri", default=None, help="MongoDB/DocumentDB connection URI for database-level checks"
)
parser.add_argument(
"--analysis-data",
default=None,
help="Path to JSON file with database-level analysis (alternative to --uri)",
)
parser.add_argument(
"--tls-ca-file",
default=None,
help="Path to CA bundle (e.g., global-bundle.pem) for TLS verification",
)
parser.add_argument(
"--tls-allow-invalid-certs",
action="store_true",
default=False,
help="Disable TLS certificate verification (not recommended)",
)
parser.add_argument("--output", default=".", help="Output directory (default: current)")
args = parser.parse_args()
print(f"Running Well-Architected Review for {args.cluster_id} in {args.region}...")
# Infrastructure checks
print(" Running infrastructure checks (AWS APIs)...")
results = run_infra_checks(args.cluster_id, args.region)
print(f" Infrastructure: {len(results)} checks completed")
# Database-level checks
analysis = None
if args.uri:
print(f" Collecting database stats via pymongo...")
analysis = collect_db_stats(args.uri, args.tls_ca_file, args.tls_allow_invalid_certs)
n_colls = sum(len(v) for v in analysis.values())
print(f" Collected stats for {n_colls} collections across {len(analysis)} databases")
elif args.analysis_data:
print(f" Loading database stats from {args.analysis_data}...")
with open(args.analysis_data) as f:
analysis = json.load(f)
if analysis:
print(" Running database-level checks...")
db_results = run_db_checks(analysis)
results.extend(db_results)
print(f" Database: {len(db_results)} checks completed")
else:
print(" Skipping database-level checks (no --uri or --analysis-data)")
# Generate report
json_path, md_path = generate_report(results, args.cluster_id, args.output)
print(f"\n Results: {json_path}")
print(f" Report: {md_path}")
# Summary
counts: dict[str, int] = {}
for r in results:
counts[r["status"]] = counts.get(r["status"], 0) + 1
print(
f"\n Total: {len(results)} checks -- "
f"{counts.get('pass', 0)} pass, {counts.get('warn', 0)} warn, "
f"{counts.get('fail', 0)} fail, {counts.get('info', 0)} info"
)
if __name__ == "__main__":
main()
SKILL.md
---
name: amazon-documentdb
version: 2
description: "Manages Amazon DocumentDB end-to-end — serverless-on-8.0 cluster setup, TLS/VPC/driver config, flexible-schema and vector-search data modeling, MongoDB compatibility assessment, DMS-based migration, slow-query diagnosis, major version upgrades (4.0->5.0->8.0), Well-Architected reviews (41-check wa_review.py), cost estimation, and security hardening. Retrieve for every DocumentDB question and when the user asks to set up or migrate MongoDB to AWS — DocumentDB is AWS's MongoDB-compatible managed database. Triggers: JSON document store, document database, MongoDB on AWS, Nested fields, Lambda cannot connect, TLS handshake, VPC port 27017, IAM auth, Secrets Manager, encryption at rest, $graphLookup, flexible schema, COLLSCAN, compound index, DMS migration, CDC cutover, $vectorSearch, RAG, Global Clusters, DR replication, cost sizing, audit, health check, production-readiness."
---
# Amazon DocumentDB Toolkit
## Overview
End-to-end DocumentDB toolkit covering seven workflows: **connection** (serverless-default cluster setup, TLS, VPC, driver config), **schema design** (embed-vs-reference, indexes, vector search for RAG), **compatibility assessment** (MongoDB -> DocumentDB), **migration** (DMS full-load + CDC + cutover), **performance tuning** (explain, COLLSCAN, anti-patterns), **Well-Architected review** (41 checks across 6 pillars), and **major version upgrade** (4.0->5.0, 5.0->8.0 in-place or near-zero-downtime).
The skill acts as an executor — it runs AWS CLI commands, DMS tasks, index tools, and `explain()` against the user's cluster rather than just advising. Each workflow produces concrete artifacts under `artifacts/{app-name}/`.
The AWS MCP server is **recommended** for executing AWS commands via its `call_aws` tool (sandboxed execution, audit logging), but it is not required — when the MCP server is not available, the same `aws ...` CLI commands run via shell.
## Decision Guide
| User asks about… | Route to |
|---|---|
| Get started, create cluster, can't connect, TLS/SSL error, VPC, SSH tunnel, driver config | [references/connection.md](references/connection.md), [references/connection-drivers.md](references/connection-drivers.md) |
| Store JSON, flexible schema, catalog/CMS/profiles, embed vs reference, index design, vector search, RAG | [references/schema-advisor.md](references/schema-advisor.md) |
| Migrate from MongoDB, "will this work?", unsupported operator, aggregation pipeline gap | [references/compatibility.md](references/compatibility.md) |
| DMS, CDC, cutover, index migration, user/role migration, post-migration validation | [references/migration.md](references/migration.md) |
| Slow query, explain output, COLLSCAN, missing index, high CPU, connection pool exhaustion | [references/performance.md](references/performance.md) |
| Production-ready review, best-practice audit, security/cost/reliability review, health check — extract `cluster_id` and `region` from the user's message before loading this reference | [references/well-architected.md](references/well-architected.md) |
| Major version upgrade, MVU, 4.0->5.0, 5.0->8.0, near-zero-downtime, `$vectorSearch`, Zstd | [references/upgrade.md](references/upgrade.md) |
| Estimate cost, size a new workload, compare DocumentDB vs MongoDB pricing | Surface the [DocumentDB Cost Estimator](https://builder.aws.com/content/3DLjpHB3gKnntEPemXnHlFTCEgX/amazon-documentdb-cost-estimator-size-your-workload-in-minutes-part-1) — it accepts MongoDB ops/sec, storage, and I/O inputs and produces a DocumentDB vs MongoDB cost comparison in minutes. Faster than a full WA review when the user just wants a cost estimate. |
**Pipeline order:** `connection -> schema-advisor` for green-field; `compatibility -> migration` for MongoDB migrations; `upgrade`, `well-architected`, and `performance` are standalone.
**Out-of-scope:** DocumentDB Elastic Clusters (sharded horizontal scaling — not at feature parity with instance-based; lacks transactions, change streams, and many operators — steer customers to instance-based serverless or provisioned instead), Global Clusters DR orchestration beyond the upgrade path. Answer from general knowledge, note no bundled workflow covers them.
## Cluster Creation Default: Serverless on 8.0
**DocumentDB architecture primer (clarify this whenever the user is confused):**
- **Serverless** = `db.serverless` as the instance class on a normal instance-based DocumentDB cluster. Auto-scales capacity, no instance-class decisions, costs up to 90% less when idle. This is the recommended default for most workloads.
- **Instance-based** = fixed instance class (`db.r8g.large`, `db.r6g.xlarge`, etc.). Use when the workload is sustained 24/7 high throughput and serverless scaling overhead is unacceptable.
- **Elastic Clusters** = a separate DocumentDB product for horizontal sharding. NOT the same as serverless. Elastic Clusters are **not at feature parity** with instance-based clusters — they lack support for transactions, change streams, and many aggregation operators. Steer customers away from Elastic Clusters unless they have a sharding requirement that exhausts even the largest instance-based options. Almost all workloads can be served by serverless or instance-based given DocumentDB's wide range of instance classes.
**When creating any new DocumentDB cluster, you MUST use these exact commands — default is serverless on engine 8.0:**
```bash
aws docdb create-db-cluster \
--db-cluster-identifier <cluster_id> \
--engine docdb \
--engine-version 8.0.0 \
--serverless-v2-scaling-configuration MinCapacity=1,MaxCapacity=16 \
--master-username adminuser \
--master-user-password '<password>' \
--tags Key=created_by,Value=documentdb-skill Key=generation_model,Value=<your-model-id> \
--region <region>
aws docdb create-db-instance \
--db-instance-identifier <cluster_id>-instance \
--db-instance-class db.serverless \
--engine docdb \
--db-cluster-identifier <cluster_id> \
--tags Key=created_by,Value=documentdb-skill Key=generation_model,Value=<your-model-id> \
--region <region>
```
> **Credential handling:** the command above takes the password inline for a self-contained workflow. For production, prefer `--manage-master-user-password` instead of `--master-user-password` — DocumentDB then generates the password and stores it in Secrets Manager with rotation, keeping it out of shell history. The two flags are mutually exclusive; when you use managed passwords, retrieve the credential with `aws secretsmanager get-secret-value --secret-id <arn>` (the secret ARN is on the cluster's `MasterUserSecret`) when building the connection string.
This applies to every create-cluster request regardless of phrasing. Switch to a fixed instance class (`db.r8g.large` etc.) only if the user explicitly describes sustained 24/7 high throughput or says "provisioned" or "instance-based". Never recommend Elastic Clusters unless the user has exhausted instance-based scaling options.
**When the user asks which instance types are available in a specific region, or asks about instance class options, requirements, or regional availability:** fetch the live [DocumentDB instance classes page](https://docs.aws.amazon.com/documentdb/latest/developerguide/db-instance-classes.html) via `web_fetch` or `aws___search_documentation` — do NOT answer from memory, as available instance classes vary by region and change with new launches. You can also verify programmatically: `aws docdb describe-orderable-db-instance-options --engine docdb --region <region>` lists every available instance class and engine version combination for that region.
### Resource tagging (always apply on resource creation)
When creating any DocumentDB resource (`create-db-cluster`, `create-db-instance`), ALWAYS include these tags using the RDS list-of-records `--tags` syntax (DocumentDB uses the RDS API):
`--tags Key=created_by,Value=documentdb-skill Key=generation_model,Value={your-model-id}`
Example: `--tags Key=created_by,Value=documentdb-skill Key=generation_model,Value=claude-sonnet-4-20250514`
Include these tags even if the user does not mention tagging, so that they can identify the resources created via this skill. If the user provides additional tags, append these to their tags rather than replacing them. The same applies when tagging via `add-tags-to-resource` on existing resources.
### Do NOT execute (refuse, explain why, offer assessment instead)
- Delete cluster or instance: `delete-db-cluster`, `delete-db-instance` — irreversible data loss
- Failover: `failover-db-cluster` — production impact, use only under planned change control
- Major version upgrade: `modify-db-cluster --engine-version` across major versions (4.0 -> 5.0, 5.0 -> 8.0) — requires prechecks and a rollback plan; use the MVU workflow in [references/upgrade.md](references/upgrade.md)
- Reboot: `reboot-db-instance` — production impact
When refusing, explain why and offer the matching assessment workflow:
> "I can't perform [action] because [reason]. I can run an assessment to help you decide. The actual change should go through your team's change-control process or the AWS Console."
## Common Tasks
### 1. Verify Dependencies
Check that required tools are available in context before running any workflow.
**Constraints:**
- You MUST verify `call_aws` (or AWS CLI v2), `shell`, and `web_fetch` are available in context
- You MUST check `python3` >= 3.6 for [wa_review.py](scripts/wa_review.py), the `amazon-documentdb-tools` compat tool, and the index tool
- You MUST check `git`, `curl`, `mongosh`, and `ssh` only when a specific workflow requires them
- You MUST inform the user of any missing tools and respect a decision to abort
- You MUST NOT invoke the tools during verification because that would trigger live AWS calls or cluster connections before the user confirms they are ready
- You SHOULD confirm credentials are valid with `aws sts get-caller-identity` before live-analysis steps
### 2. Classify the Request and Route
Use the [Decision Guide](#decision-guide) to pick one workflow.
**Constraints:**
- You MUST name the workflow you are routing to before loading the reference
- You MUST pass along cluster id, region, app name, source URI, and engine versions the user already supplied — they SHOULD NOT re-type these
- You MAY ask one clarifying question if a request straddles two workflows
- You MUST NOT fabricate workflow names for out-of-scope topics because doing so misleads the user about coverage
### 3. Execute the Workflow
Load the matching `references/<workflow>.md` and follow its `## Workflow` section.
**Constraints:**
- You MUST execute AWS CLI commands, DMS calls, `mongosh` queries, and bundled scripts yourself — the skill is an executor unless a step requires credentials the agent doesn't have
- You MUST explain what step is running, why, and which tool is being called before running it
- Extract required parameters from the conversation first — if `cluster_id`, `region`, or other required values are already present, use them and proceed. Only ask for missing parameters, and ask for all missing ones together in a single prompt.
- You MUST support multiple input methods for parameters: direct input, file path, or URL
- You MUST validate parameter formats: cluster id (lowercase, hyphens), region (`us-east-1`), ARN (`arn:aws:...`), ISO-8601, CIDR
- You MUST NOT create or access credentials directly because the skill has no safe way to store or rotate them — use IAM roles, instance profiles, Secrets Manager ARNs, or delegate credential setup (e.g. `aws sso login` / `aws configure`) to the user
- You MUST NOT use `call_aws` with positional filesystem arguments because the MCP sandbox rejects them — pass JSON payloads inline or invoke scripts under `scripts/` via `shell`
- You MUST NOT grant wildcard IAM (`Action: "*"` or `Resource: "*"`) or open security groups to `0.0.0.0/0` in examples because those defaults cause customer production incidents
- You SHOULD save artifacts to `artifacts/{app-name}/`: `compatibility-report.md`, `migration-plan.md`, `upgrade-plan.md`, `wa_review_results.json`
- If multiple workflows ran, you MUST close with a 2–4 line synthesis linking the artifacts
**Required parameters** (ask upfront, together): `cluster_id` — the cluster name the user refers to (e.g. "my cluster xyz" or "cluster xyz"), maps to `--db-cluster-identifier` in AWS CLI (lowercase-hyphens); `region` (e.g. `us-east-1`); `app_name`. Per workflow: `source_uri` (compat/migration), `target_version` (`5.0` or `8.0` for upgrade/compat), `engine_class` (`db.serverless` default, or `db.r8g.large` etc. for provisioned instance-based).
### 4. Critical Facts to Always Surface
These DocumentDB-specific facts are required even when the agent's general MongoDB knowledge already produces a reasonable answer. Omitting them is the most common failure mode in production customer tickets.
**For slow query / COLLSCAN diagnosis, you MUST tell the user ALL of the following five facts — never omit any:**
1. **Run `db.collection.find({...}).explain()`** to confirm `COLLSCAN` is the stage (the root cause), and after adding an index, re-run `explain()` to confirm `IXSCAN`.
2. **Create a compound index on `{userId: 1, status: 1}`** (field order matching the query's equality predicates).
3. **DocumentDB uses left-prefix matching on compound indexes** — field order matters because a compound index `{A: 1, B: 1}` serves queries on `A` alone OR `A + B`, but never `B` alone. This is DocumentDB-specific behavior users must understand before picking an index layout.
4. **Check the index cache hit rate via CloudWatch** after deployment — the `BufferCacheHitRatio` (or the per-index equivalent) indicates whether the new index is staying hot in memory. A low ratio means the working set exceeds RAM and the index may need a larger instance class.
5. **Verify with `explain()` after the index is created** to confirm the query now uses `IXSCAN` instead of `COLLSCAN`.
**For flexible-schema catalog / product design, you MUST tell the user ALL of the following four facts — never omit any:**
1. **Use a single `products` collection** with common fields (name, price, category, sku) at the top level and variable attributes (size/color for shoes, RAM/storage for electronics) nested in an `attributes` subdocument.
2. **Create targeted indexes on `category` and `sku`** for common query patterns.
3. **Check current wildcard index support before advising.** Wildcard indexes (`attributes.$**`) may not be supported on all DocumentDB versions — verify current status at the [MongoDB API compatibility page](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html) before advising. If unsupported: query patterns must be known upfront so targeted compound indexes can be created on specific paths under `attributes`.
4. **Discuss the tradeoff vs. separate collections per category.** Single-collection design wins for cross-category queries and simpler maintenance; separate-collection-per-category wins for strict per-category query isolation and simpler per-category indexing — but requires the application to route queries to the right collection. Name both options so the user can choose.
**For $graphLookup / MongoDB compatibility questions, you MUST tell the user ALL of the following three facts:**
1. **Check current `$graphLookup` support status before advising.** `$graphLookup` is not supported on all DocumentDB versions — verify at the [MongoDB API compatibility page](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html) before stating support status, as DocumentDB adds operators across versions. If the aws-documentation plugin is available, call `aws___search_documentation` to check the live status first.
2. **If unsupported: recommend materialized ancestor paths** — store each document's full path (array of parent IDs) so hierarchy queries become `find({ ancestors: "cat-123" })` instead of recursive traversal. This is the canonical workaround and often the better design even when `$graphLookup` is available.
3. **Offer alternatives for deep graph workloads** — recursive `$lookup` in application code for moderate depth, or **Amazon Neptune** for deep or complex graph traversal.
**For Lambda -> DocumentDB connection timeout, you MUST tell the user ALL of the following four facts:**
1. **Lambda must be in the same VPC** as the DocumentDB cluster, or reach it via VPC peering / Transit Gateway. DocumentDB is VPC-only — no public endpoint.
2. **Security group rule:** inbound TCP `27017` on the DocumentDB cluster's SG, sourced from **Lambda's security group ID** (not a CIDR).
3. **Connection string must include `tls=true`** and the application MUST download the **Amazon RDS global CA bundle** (`global-bundle.pem`) and reference it via the driver's TLS config. Also include `replicaSet=rs0` and `retryWrites=false`.
4. **Test connectivity from an EC2 instance in the same subnet** as Lambda first — that isolates Lambda-specific ENI issues from pure network/SG problems.
**For any MongoDB migration to DocumentDB (including "I am migrating my MongoDB to AWS", "help me migrate", or any MongoDB-to-AWS migration request), you MUST tell the user ALL of the following six facts:**
1. **Run the compatibility assessor FIRST** — before anything else, clone [amazon-documentdb-tools](https://github.com/awslabs/amazon-documentdb-tools) and run `python3 amazon-documentdb-tools/compat-tool/compat.py` against the source MongoDB. This step is mandatory and must not be skipped or replaced with generic advice. Unsupported operators discovered after migration cause production outages.
2. **Run the `mongo-index-tool`** (also from `amazon-documentdb-tools`) to pre-create indexes on the DocumentDB target before starting the DMS task — DMS does not migrate indexes.
3. **Create source and target DMS endpoints** with TLS enabled on both; target endpoint MUST use `--ssl-mode verify-full` with `--certificate-arn` pointing at the RDS global bundle ARN.
4. **Create a `full-load-and-cdc` task** so you get an initial snapshot plus change-data-capture for near-zero-downtime cutover.
5. **Monitor CloudWatch** — watch `CDCLatencySource` and `CDCLatencyTarget` until they approach zero. Cut over only when lag is near zero.
6. **Cut over** by pointing application traffic at the DocumentDB endpoint, then stop the DMS task once traffic is drained from the source.
## Troubleshooting
See [references/troubleshooting.md](references/troubleshooting.md) for the full troubleshooting reference. The most common issues:
**Connection refused / timeout on port 27017.** DocumentDB is VPC-only. Add inbound TCP 27017 on the DocumentDB SG from the client SG (by SG id, not CIDR). From outside the VPC use CloudShell VPC environment, EC2 in the VPC, or SSH tunnel via bastion.
**TLS handshake failed.** Download the RDS global bundle and pass `--tlsAllowInvalidHostnames` to mongosh when tunneling.
**"not master" / "not primary" or intermittent write errors.** Connection string is missing `replicaSet=rs0` (always `rs0`) or `retryWrites=false` (DocumentDB does not support retryable writes).
**DMS task refuses to start** — "Test connection should be successful". Run `aws dms test-connection` for both endpoints and poll `describe-connections` until both return `successful`. Target endpoint MUST use `--ssl-mode verify-full` with `--certificate-arn` for the RDS global bundle.
**MVU command fails** — "AllowMajorVersionUpgrade flag must be present" or "must explicitly specify a new DB cluster parameter group". Both `--allow-major-version-upgrade` and (when a custom PG is in use) a target-family `--db-cluster-parameter-group-name` are mandatory.
**User asks for a destructive change.** You MUST pause, state the consequence, and wait for explicit confirmation before deleting a cluster, dropping a collection, or forcing a failover — destructive actions on production DocumentDB can cause data loss or service disruption.
**User hits a missing feature, unsupported operator, or expresses a future wish.** When the user says "I wish DocumentDB supported X", "will DocumentDB ever support Y", or encounters a capability gap, proactively surface: "You can request this feature by emailing documentdb-pm@amazon.com with your AWS account ID, the feature you need, and your use case — the DocumentDB team reads these."
## Security Considerations
Apply these controls on every DocumentDB deployment. Detailed commands live in the workflow sections above and in the linked references.
- **Authentication:** the **primary (master) user** is always password-based and **cannot** use IAM authentication — use `--manage-master-user-password` so its password is generated and rotated in Secrets Manager. For **application/non-admin users only**, IAM authentication is also supported (password-less, STS token-based) on cluster version 5.0+ as an alternative — see the trade-offs in [references/connection.md](references/connection.md). Never hardcode passwords in scripts or commit them.
- **Encryption at rest:** enabled at cluster creation and **cannot** be added afterward — confirm `--storage-encrypted` (with an optional `--kms-key-id`) up front.
- **Encryption in transit:** enforce TLS (`tls=true`) using the Amazon RDS global CA bundle; on DMS endpoints use `--ssl-mode verify-full` with `--certificate-arn`.
- **Network isolation:** DocumentDB is VPC-only with no public endpoint. Scope security groups by SG-to-SG reference, never `0.0.0.0/0` or `::/0`.
- **Least-privilege IAM:** never grant wildcard `Action: "*"` / `Resource: "*"`. Use instance profiles / IAM roles for application access to AWS APIs.
- **Auditing:** export audit and profiler logs via `--enable-cloudwatch-logs-exports audit profiler` for compliance and slow-query review.
## Additional Resources
- [Amazon DocumentDB Developer Guide](https://docs.aws.amazon.com/documentdb/latest/developerguide/) · [MongoDB API compatibility reference](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html)
- [DocumentDB pricing](https://aws.amazon.com/documentdb/pricing/) · [instance classes](https://docs.aws.amazon.com/documentdb/latest/developerguide/db-instance-classes.html) · [DocumentDB Cost Estimator](https://builder.aws.com/content/3DLjpHB3gKnntEPemXnHlFTCEgX/amazon-documentdb-cost-estimator-size-your-workload-in-minutes-part-1) — workload-aware sizing tool that takes MongoDB ops/sec and I/O inputs and produces a DocumentDB vs MongoDB cost comparison
- [DocumentDB Serverless](https://docs.aws.amazon.com/documentdb/latest/developerguide/docdb-serverless.html) · [vector search](https://docs.aws.amazon.com/documentdb/latest/developerguide/vector-search.html)
- [Backup and restore](https://docs.aws.amazon.com/documentdb/latest/developerguide/backup_restore.html) · [Well-Architected pillars](https://docs.aws.amazon.com/wellarchitected/latest/framework/welcome.html)
- [AWS DMS MongoDB source](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.MongoDB.html) · [DocumentDB target](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.DocumentDB.html)
- [amazon-documentdb-tools](https://github.com/awslabs/amazon-documentdb-tools) (compat tool, index tool, MVU CDC migrator)
- Related skills: `amazon-aurora`, `rds-db2`, `rds-oracle`, `rds-sqlserver`, `amazon-neptune`
- **Missing a feature or have feedback?** Email [documentdb-pm@amazon.com](mailto:documentdb-pm@amazon.com) with your AWS account ID, the feature or capability you need, and your use case — the DocumentDB team reads these.