Setting the file. One moment. 63.7
Troubleshooting
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
Reference
scripts/wa_review.py
Python·1,341 lines·47 KB
17import argparse
18import json
19import re
20import sys
21from datetime import datetime, timedelta, timezone
22from pathlib import Path
23
24try:
25 import boto3
26except ImportError:
27 print("ERROR: boto3 required. Install with: pip install boto3")
28 sys.exit(1)
29
30# ---------------------------------------------------------------------------
31# Constants
32# ---------------------------------------------------------------------------
33CONN_LIMITS = {
34 "db.t3.medium": 1000,
35 "db.t4g.medium": 1000,
36 "db.r5.large": 3400,
37 "db.r6g.large": 3400,
38 "db.r6gd.large": 3400,
39 "db.r8g.large": 3400,
40 "db.r5.xlarge": 7000,
41 "db.r6g.xlarge": 7000,
42 "db.r6gd.xlarge": 7000,
43 "db.r8g.xlarge": 7000,
44 "db.r5.2xlarge": 14200,
45 "db.r6g.2xlarge": 14200,
46 "db.r6gd.2xlarge": 14200,
47 "db.r8g.2xlarge": 14200,
48 "db.r5.4xlarge": 28400,
49 "db.r6g.4xlarge": 28400,
50 "db.r6gd.4xlarge": 28400,
51 "db.r8g.4xlarge": 28400,
52 "db.r5.8xlarge": 60000,
53 "db.r6g.8xlarge": 60000,
54 "db.r6gd.8xlarge": 60000,
55 "db.r8g.8xlarge": 60000,
56 "db.r5.12xlarge": 60000,
57 "db.r6g.12xlarge": 60000,
58 "db.r6gd.12xlarge": 60000,
59 "db.r8g.12xlarge": 60000,
60 "db.r5.16xlarge": 60000,
61 "db.r6g.16xlarge": 60000,
62 "db.r6gd.16xlarge": 60000,
63 "db.r8g.16xlarge": 60000,
64 "db.r5.24xlarge": 60000,
65}
66INSTANCE_RAM_GIB = {
67 "db.t3.medium": 4,
68 "db.t4g.medium": 4,
69 "db.r5.large": 16,
70 "db.r6g.large": 16,
71 "db.r6gd.large": 16,
72 "db.r8g.large": 16,
73 "db.r5.xlarge": 32,
74 "db.r6g.xlarge": 32,
75 "db.r6gd.xlarge": 32,
76 "db.r8g.xlarge": 32,
77 "db.r5.2xlarge": 64,
78 "db.r6g.2xlarge": 64,
79 "db.r6gd.2xlarge": 64,
80 "db.r8g.2xlarge": 64,
81 "db.r5.4xlarge": 128,
82 "db.r6g.4xlarge": 128,
83 "db.r6gd.4xlarge": 128,
84 "db.r8g.4xlarge": 128,
85 "db.r5.8xlarge": 256,
86 "db.r6g.8xlarge": 256,
87 "db.r6gd.8xlarge": 256,
88 "db.r8g.8xlarge": 256,
89 "db.r5.12xlarge": 384,
90 "db.r6g.12xlarge": 384,
91 "db.r6gd.12xlarge": 384,
92 "db.r8g.12xlarge": 384,
93 "db.r5.16xlarge": 512,
94 "db.r6g.16xlarge": 512,
95 "db.r6gd.16xlarge": 512,
96 "db.r8g.16xlarge": 512,
97 "db.r5.24xlarge": 768,
98}
99GRAVITON_FAMILIES = ("r6g", "r7g", "r8g", "t4g", "r6gd")
100
101
102def _add(results, pillar, check_id, label, status, detail=""):
103 results.append(
104 {"pillar": pillar, "id": check_id, "label": label, "status": status, "detail": detail}
105 )
106
107
108# ---------------------------------------------------------------------------
109# Infrastructure checks (boto3)
110# ---------------------------------------------------------------------------
111def run_infra_checks(cluster_id, region):
112 results: list[dict] = []
113 docdb = boto3.client("docdb", region_name=region)
114 cw = boto3.client("cloudwatch", region_name=region)
115 ec2 = boto3.client("ec2", region_name=region)
116
117 try:
118 cl = docdb.describe_db_clusters(DBClusterIdentifier=cluster_id)["DBClusters"][0]
119 except Exception as e:
120 _add(results, "Other", "ERR", f"Cannot describe cluster: {e}", "fail")
121 return results
122
123 try:
124 insts = docdb.describe_db_instances(
125 Filters=[{"Name": "db-cluster-id", "Values": [cluster_id]}]
126 )["DBInstances"]
127 except Exception as e:
128 insts = []
129 _add(results, "Other", "ERR", f"Cannot describe instances: {e}", "fail")
130
131 # -- RELIABILITY --------------------------------------------------------
132 retention = cl.get("BackupRetentionPeriod", 1)
133 _add(
134 results,
135 "Reliability",
136 "REL1",
137 f"Backup retention period ({retention} days)",
138 "pass" if retention >= 7 else "warn" if retention >= 3 else "fail",
139 "Recommended: 7+ days for production" if retention < 7 else "",
140 )
141
142 del_prot = cl.get("DeletionProtection", False)
143 _add(
144 results,
145 "Reliability",
146 "REL2",
147 f"Deletion protection ({'enabled' if del_prot else 'disabled'})",
148 "pass" if del_prot else "fail",
149 "" if del_prot else "Enable deletion protection for production clusters",
150 )
151
152 n_inst = len(insts)
153 _add(
154 results,
155 "Reliability",
156 "REL5a",
157 f"Instance count ({n_inst})",
158 "pass" if n_inst >= 2 else "fail",
159 "Minimum 2 instances required for auto failover" if n_inst < 2 else "",
160 )
161
162 azs = {i.get("AvailabilityZone", "") for i in insts}
163 _add(
164 results,
165 "Reliability",
166 "REL5b",
167 f"Instances across {len(azs)} AZ(s)",
168 "pass" if len(azs) >= 2 else "fail",
169 "Single AZ -- no failover protection" if len(azs) < 2 else "",
170 )
171
172 engine_ver = cl.get("EngineVersion", "unknown")
173 major = engine_ver.split(".")[0] if engine_ver != "unknown" else ""
174 if major in ("3", "4"):
175 _add(
176 results,
177 "Reliability",
178 "REL6",
179 f"Engine version {engine_ver} (approaching or past end-of-life)",
180 "fail",
181 "Upgrade to DocumentDB 5.0 or 8.0",
182 )
183 elif major == "5":
184 _add(
185 results,
186 "Reliability",
187 "REL6",
188 f"Engine version {engine_ver}",
189 "pass",
190 "Consider upgrading to 8.0 for Zstandard compression and Query Planner v3",
191 )
192 else:
193 _add(results, "Reliability", "REL6", f"Engine version {engine_ver}", "pass")
194
195 # Recent failover events (14 days) — paginated
196 try:
197 evt_end = datetime.now(timezone.utc)
198 evt_start = evt_end - timedelta(days=13)
199 events_list = []
200 paginator = docdb.get_paginator("describe_events")
201 for page in paginator.paginate(
202 SourceIdentifier=cluster_id,
203 SourceType="db-cluster",
204 StartTime=evt_start,
205 EndTime=evt_end,
206 ):
207 events_list.extend(page.get("Events", []))
208 failover_events = [
209 e
210 for e in events_list
211 if "failover" in e.get("Message", "").lower()
212 or "failover" in ",".join(e.get("EventCategories", [])).lower()
213 ]
214 if failover_events:
215 _add(
216 results,
217 "Reliability",
218 "REL7",
219 f"{len(failover_events)} failover event(s) in last 13 days",
220 "warn",
221 f"Most recent: {failover_events[-1].get('Message', '')[:120]}",
222 )
223 else:
224 _add(results, "Reliability", "REL7", "No failover events in last 13 days", "pass")
225 except Exception as e:
226 _add(results, "Reliability", "REL7", f"Cannot check events: {e}", "warn")
227
228 # -- SECURITY -----------------------------------------------------------
229 encrypted = cl.get("StorageEncrypted", False)
230 _add(
231 results,
232 "Security",
233 "SEC1a",
234 f"Encryption at rest ({'enabled' if encrypted else 'disabled'})",
235 "pass" if encrypted else "fail",
236 "" if encrypted else "Enable encryption at rest (requires new cluster)",
237 )
238
239 tls_val = "unknown"
240 try:
241 pg_name = cl.get("DBClusterParameterGroup", "")
242 if pg_name:
243 params = docdb.describe_db_cluster_parameters(DBClusterParameterGroupName=pg_name).get(
244 "Parameters", []
245 )
246 for p in params:
247 if p.get("ParameterName") == "tls":
248 tls_val = p.get("ParameterValue", "enabled")
249 elif p.get("ParameterName") == "tls_version":
250 tv = p.get("ParameterValue", "")
251 if tv and "1.2" in tv and "1.0" not in tv and "1.1" not in tv:
252 _add(results, "Security", "SEC6", f"TLS minimum version: {tv}", "pass")
253 elif tv:
254 _add(
255 results,
256 "Security",
257 "SEC6",
258 f"TLS minimum version: {tv}",
259 "warn",
260 "Set tls_version to TLSv1.2 to disable older protocols",
261 )
262 except Exception as e:
263 _add(results, "Security", "SEC6", f"Cannot check TLS parameters: {e}", "warn")
264 # Ensure SEC6 is always present
265 if not any(r["id"] == "SEC6" for r in results):
266 _add(
267 results,
268 "Security",
269 "SEC6",
270 "TLS minimum version: unknown",
271 "warn",
272 "Could not determine tls_version parameter",
273 )
274 _add(
275 results,
276 "Security",
277 "SEC1b",
278 f"TLS ({tls_val})",
279 "pass" if tls_val == "enabled" else "warn" if tls_val == "unknown" else "fail",
280 (
281 "Could not determine TLS status"
282 if tls_val == "unknown"
283 else ("" if tls_val == "enabled" else "TLS should be enabled")
284 ),
285 )
286
287 # Security groups
288 sg_open = False
289 sg_checked = 0
290 for vsg in cl.get("VpcSecurityGroups", []):
291 sg_id = vsg.get("VpcSecurityGroupId", "")
292 if not sg_id:
293 continue
294 try:
295 sg_detail = ec2.describe_security_groups(GroupIds=[sg_id])["SecurityGroups"][0]
296 sg_checked += 1
297 for rule in sg_detail.get("IpPermissions", []):
298 for ip_range in rule.get("IpRanges", []):
299 if ip_range.get("CidrIp") == "0.0.0.0/0":
300 sg_open = True
301 _add(
302 results,
303 "Security",
304 "SEC2",
305 f"Security group {sg_id} open to 0.0.0.0/0",
306 "fail",
307 "Restrict to specific CIDR ranges",
308 )
309 for ip_range in rule.get("Ipv6Ranges", []):
310 if ip_range.get("CidrIpv6") == "::/0":
311 sg_open = True
312 _add(
313 results,
314 "Security",
315 "SEC2",
316 f"Security group {sg_id} open to ::/0",
317 "fail",
318 "Restrict to specific CIDR ranges",
319 )
320 except Exception as e:
321 _add(results, "Security", "SEC2", f"Cannot check SG {sg_id}: {e}", "warn")
322 if not sg_open and sg_checked > 0:
323 _add(
324 results,
325 "Security",
326 "SEC2",
327 f"Security groups properly restricted ({sg_checked} checked)",
328 "pass",
329 )
330
331 logs = cl.get("EnabledCloudwatchLogsExports", [])
332 audit_enabled = "audit" in logs
333 profiler_enabled = "profiler" in logs
334 _add(
335 results,
336 "Security",
337 "SEC5",
338 f"Audit logging ({'enabled' if audit_enabled else 'disabled'})",
339 "pass" if audit_enabled else "warn",
340 "" if audit_enabled else "Enable audit logging for compliance",
341 )
342
343 # Secrets Manager
344 try:
345 sm = boto3.client("secretsmanager", region_name=region)
346 found_secret = False
347 for page in sm.get_paginator("list_secrets").paginate():
348 for s in page.get("SecretList", []):
349 name = (s.get("Name", "") or "").lower()
350 desc = (s.get("Description", "") or "").lower()
351 if cluster_id.lower() in name or cluster_id.lower() in desc:
352 found_secret = True
353 break
354 if found_secret:
355 break
356 _add(
357 results,
358 "Security",
359 "SEC3",
360 f"Secrets Manager {'references' if found_secret else 'does not reference'} this cluster",
361 "pass" if found_secret else "warn",
362 "" if found_secret else "Store credentials in Secrets Manager",
363 )
364 except Exception as e:
365 _add(results, "Security", "SEC3", f"Cannot check Secrets Manager: {e}", "warn")
366
367 # -- OPERATIONAL EXCELLENCE ---------------------------------------------
368 sg_name = cl.get("DBSubnetGroup", "")
369 try:
370 if sg_name:
371 sg = docdb.describe_db_subnet_groups(DBSubnetGroupName=sg_name)["DBSubnetGroups"][0]
372 sg_azs = {s["SubnetAvailabilityZone"]["Name"] for s in sg.get("Subnets", [])}
373 _add(
374 results,
375 "Operational Excellence",
376 "OPS2",
377 f"Subnet group spans {len(sg_azs)} AZ(s)",
378 "pass" if len(sg_azs) >= 3 else "warn",
379 "Recommended: 3 AZs for failover flexibility" if len(sg_azs) < 3 else "",
380 )
381 except Exception as e:
382 _add(results, "Operational Excellence", "OPS2", f"Cannot check subnet group: {e}", "warn")
383
384 _add(
385 results,
386 "Operational Excellence",
387 "OPS5a",
388 f"Profiler logging ({'enabled' if profiler_enabled else 'disabled'})",
389 "pass" if profiler_enabled else "warn",
390 "" if profiler_enabled else "Enable profiler for slow query analysis",
391 )
392
393 pg_name = cl.get("DBClusterParameterGroup", "")
394 _add(
395 results,
396 "Operational Excellence",
397 "OPS5c",
398 f"Parameter group: {pg_name}",
399 "warn" if pg_name.startswith("default.") else "pass",
400 (
401 "Use a custom parameter group for workload-specific tuning"
402 if pg_name.startswith("default.")
403 else ""
404 ),
405 )
406
407 _add(
408 results,
409 "Operational Excellence",
410 "OPS7",
411 f"Maintenance window: {cl.get('PreferredMaintenanceWindow', 'not set')}",
412 "info",
413 "Verify this window aligns with your lowest-traffic period",
414 )
415
416 try:
417 n_alarms = len(cw.describe_alarms(AlarmNamePrefix=cluster_id).get("MetricAlarms", []))
418 _add(
419 results,
420 "Operational Excellence",
421 "OPS5b",
422 f"CloudWatch alarms ({n_alarms} configured)",
423 "pass" if n_alarms >= 3 else "warn" if n_alarms > 0 else "fail",
424 (
425 "Recommended: alarms for CPU, FreeableMemory, DatabaseConnections"
426 if n_alarms < 3
427 else ""
428 ),
429 )
430 except Exception as e:
431 _add(results, "Operational Excellence", "OPS5b", f"Cannot check alarms: {e}", "warn")
432
433 # -- COST OPTIMIZATION --------------------------------------------------
434 try:
435 n_tags = len(
436 docdb.list_tags_for_resource(ResourceName=cl["DBClusterArn"]).get("TagList", [])
437 )
438 _add(
439 results,
440 "Cost Optimization",
441 "COST6",
442 f"Cost allocation tags ({n_tags} tags)",
443 "pass" if n_tags >= 2 else "warn",
444 "Add cost allocation tags for expense tracking" if n_tags < 2 else "",
445 )
446 except Exception as e:
447 _add(results, "Cost Optimization", "COST6", f"Cannot check tags: {e}", "warn")
448
449 storage_type = cl.get("StorageType", "standard")
450 _add(
451 results,
452 "Cost Optimization",
453 "COST7",
454 f"Storage type: {storage_type}",
455 "info",
456 (
457 "Evaluate I/O-Optimized for write-heavy workloads"
458 if storage_type != "iopt1"
459 else "I/O-Optimized active -- no per-I/O charges"
460 ),
461 )
462
463 # -- PER-INSTANCE CHECKS ------------------------------------------------
464 end = datetime.now(timezone.utc)
465 start = end - timedelta(days=7)
466
467 for inst in insts:
468 iid = inst["DBInstanceIdentifier"]
469 itype = inst["DBInstanceClass"]
470 dim = [{"Name": "DBInstanceIdentifier", "Value": iid}]
471 is_writer = inst.get("IsClusterWriter", False)
472 family = itype.replace("db.", "").split(".")[0] if itype.startswith("db.") else ""
473
474 # CPU — use hourly Maximum for P95 to capture peak usage within each hour
475 try:
476 raw_dps = cw.get_metric_statistics(
477 Namespace="AWS/DocDB",
478 MetricName="CPUUtilization",
479 Dimensions=dim,
480 StartTime=start,
481 EndTime=end,
482 Period=3600,
483 Statistics=["Average", "Maximum"],
484 ).get("Datapoints", [])
485 if raw_dps:
486 avg_cpu = sum(d["Average"] for d in raw_dps) / len(raw_dps)
487 max_vals = sorted(d["Maximum"] for d in raw_dps)
488 p95_cpu = max_vals[int(len(max_vals) * 0.95)]
489 _add(
490 results,
491 "Cost Optimization",
492 "COST1",
493 f"CPU for {iid} (avg {avg_cpu:.1f}%, P95 {p95_cpu:.1f}%)",
494 "warn" if p95_cpu < 10 else "pass",
495 f"Instance {itype} may be oversized" if p95_cpu < 10 else "",
496 )
497 except Exception as e:
498 _add(results, "Cost Optimization", "COST1", f"Cannot check CPU for {iid}: {e}", "warn")
499
500 # Graviton
501 _add(
502 results,
503 "Sustainability",
504 "SUST1",
505 f"{iid} {'uses' if family in GRAVITON_FAMILIES else 'does not use'} Graviton ({itype})",
506 "pass" if family in GRAVITON_FAMILIES else "warn",
507 (
508 ""
509 if family in GRAVITON_FAMILIES
510 else "Migrate to Graviton (r6g/r8g) for better price-performance"
511 ),
512 )
513
514 # Buffer cache hit ratio
515 try:
516 dps = [
517 d["Average"]
518 for d in cw.get_metric_statistics(
519 Namespace="AWS/DocDB",
520 MetricName="BufferCacheHitRatio",
521 Dimensions=dim,
522 StartTime=start,
523 EndTime=end,
524 Period=3600,
525 Statistics=["Average"],
526 ).get("Datapoints", [])
527 ]
528 if dps:
529 avg_cache = sum(dps) / len(dps)
530 _add(
531 results,
532 "Performance Efficiency",
533 "PERF6",
534 f"Buffer cache hit ratio for {iid} ({avg_cache:.1f}%)",
535 "pass" if avg_cache >= 99 else "warn" if avg_cache >= 95 else "fail",
536 "Working set may not fit in memory" if avg_cache < 95 else "",
537 )
538 except Exception as e:
539 _add(
540 results,
541 "Performance Efficiency",
542 "PERF6",
543 f"Cannot check cache for {iid}: {e}",
544 "warn",
545 )
546
547 # Connections vs limits
548 try:
549 dps = [
550 d["Maximum"]
551 for d in cw.get_metric_statistics(
552 Namespace="AWS/DocDB",
553 MetricName="DatabaseConnections",
554 Dimensions=dim,
555 StartTime=start,
556 EndTime=end,
557 Period=3600,
558 Statistics=["Maximum"],
559 ).get("Datapoints", [])
560 ]
561 limit = CONN_LIMITS.get(itype, 0)
562 if dps and limit:
563 max_conn = max(dps)
564 pct = max_conn / limit * 100
565 _add(
566 results,
567 "Performance Efficiency",
568 "PERF5",
569 f"Peak connections for {iid} ({int(max_conn)}/{limit} = {pct:.0f}%)",
570 "pass" if pct < 70 else "warn" if pct < 90 else "fail",
571 "Consider upsizing or connection pooling" if pct >= 70 else "",
572 )
573 elif dps:
574 _add(
575 results,
576 "Performance Efficiency",
577 "PERF5",
578 f"Connection limit unknown for {iid} ({itype})",
579 "warn",
580 "Instance type not in lookup table",
581 )
582 except Exception as e:
583 _add(
584 results,
585 "Performance Efficiency",
586 "PERF5",
587 f"Cannot check connections for {iid}: {e}",
588 "warn",
589 )
590
591 # Idle reader detection
592 if not is_writer:
593 try:
594 conn_dps = [
595 d["Average"]
596 for d in cw.get_metric_statistics(
597 Namespace="AWS/DocDB",
598 MetricName="DatabaseConnections",
599 Dimensions=dim,
600 StartTime=start,
601 EndTime=end,
602 Period=3600,
603 Statistics=["Average"],
604 ).get("Datapoints", [])
605 ]
606 io_dps = [
607 d["Average"]
608 for d in cw.get_metric_statistics(
609 Namespace="AWS/DocDB",
610 MetricName="ReadIOPS",
611 Dimensions=dim,
612 StartTime=start,
613 EndTime=end,
614 Period=3600,
615 Statistics=["Average"],
616 ).get("Datapoints", [])
617 ]
618 avg_conn = sum(conn_dps) / len(conn_dps) if conn_dps else 0
619 avg_iops = sum(io_dps) / len(io_dps) if io_dps else 0
620 if avg_conn < 2 and avg_iops < 5:
621 _add(
622 results,
623 "Cost Optimization",
624 "COST9",
625 f"Reader {iid} appears idle (avg {avg_conn:.0f} conn, {avg_iops:.0f} ReadIOPS)",
626 "warn",
627 "Consider removing this replica to reduce cost",
628 )
629 else:
630 _add(
631 results,
632 "Cost Optimization",
633 "COST9",
634 f"Reader {iid} is active (avg {avg_conn:.0f} conn, {avg_iops:.0f} ReadIOPS)",
635 "pass",
636 )
637 except Exception as e:
638 _add(
639 results, "Cost Optimization", "COST9", f"Cannot check reader {iid}: {e}", "warn"
640 )
641
642 # FreeableMemory
643 ram_gib = INSTANCE_RAM_GIB.get(itype, 0)
644 if ram_gib:
645 try:
646 dps = [
647 d["Minimum"]
648 for d in cw.get_metric_statistics(
649 Namespace="AWS/DocDB",
650 MetricName="FreeableMemory",
651 Dimensions=dim,
652 StartTime=start,
653 EndTime=end,
654 Period=3600,
655 Statistics=["Minimum"],
656 ).get("Datapoints", [])
657 ]
658 if dps:
659 min_free = min(dps)
660 free_pct = min_free / (ram_gib * 1024**3) * 100
661 _add(
662 results,
663 "Performance Efficiency",
664 "PERF11",
665 f"FreeableMemory min for {iid}: {min_free / (1024**3):.1f} GiB ({free_pct:.0f}%)",
666 "fail" if free_pct < 5 else "warn" if free_pct < 10 else "pass",
667 "Instance under memory pressure" if free_pct < 10 else "",
668 )
669 except Exception as e:
670 _add(
671 results,
672 "Performance Efficiency",
673 "PERF11",
674 f"Cannot check memory for {iid}: {e}",
675 "warn",
676 )
677 else:
678 _add(
679 results,
680 "Performance Efficiency",
681 "PERF11",
682 f"Unknown instance type {itype} -- cannot check FreeableMemory",
683 "warn",
684 )
685
686 # SwapUsage
687 try:
688 dps = [
689 d["Maximum"]
690 for d in cw.get_metric_statistics(
691 Namespace="AWS/DocDB",
692 MetricName="SwapUsage",
693 Dimensions=dim,
694 StartTime=start,
695 EndTime=end,
696 Period=3600,
697 Statistics=["Maximum"],
698 ).get("Datapoints", [])
699 ]
700 if dps and max(dps) > 0:
701 _add(
702 results,
703 "Performance Efficiency",
704 "PERF12",
705 f"SwapUsage max for {iid}: {max(dps) / (1024**2):.0f} MB",
706 "fail",
707 "Instance is swapping -- critically undersized",
708 )
709 elif dps:
710 _add(results, "Performance Efficiency", "PERF12", f"No swap on {iid}", "pass")
711 except Exception as e:
712 _add(
713 results,
714 "Performance Efficiency",
715 "PERF12",
716 f"Cannot check swap for {iid}: {e}",
717 "warn",
718 )
719
720 # DiskQueueDepth
721 try:
722 dps = [
723 d["Average"]
724 for d in cw.get_metric_statistics(
725 Namespace="AWS/DocDB",
726 MetricName="DiskQueueDepth",
727 Dimensions=dim,
728 StartTime=start,
729 EndTime=end,
730 Period=3600,
731 Statistics=["Average"],
732 ).get("Datapoints", [])
733 ]
734 if dps:
735 avg_dqd = sum(dps) / len(dps)
736 _add(
737 results,
738 "Performance Efficiency",
739 "PERF13",
740 f"DiskQueueDepth avg for {iid}: {avg_dqd:.1f}",
741 "warn" if avg_dqd > 5 else "pass",
742 "I/O backing up -- evaluate I/O-Optimized or upsizing" if avg_dqd > 5 else "",
743 )
744 except Exception as e:
745 _add(
746 results,
747 "Performance Efficiency",
748 "PERF13",
749 f"Cannot check DiskQueueDepth for {iid}: {e}",
750 "warn",
751 )
752
753 # IndexBufferCacheHitRatio
754 try:
755 dps = [
756 d["Average"]
757 for d in cw.get_metric_statistics(
758 Namespace="AWS/DocDB",
759 MetricName="IndexBufferCacheHitRatio",
760 Dimensions=dim,
761 StartTime=start,
762 EndTime=end,
763 Period=3600,
764 Statistics=["Average"],
765 ).get("Datapoints", [])
766 ]
767 if dps:
768 avg_idx = sum(dps) / len(dps)
769 _add(
770 results,
771 "Performance Efficiency",
772 "PERF14",
773 f"IndexBufferCacheHitRatio for {iid}: {avg_idx:.1f}%",
774 "pass" if avg_idx >= 99 else "warn" if avg_idx >= 95 else "fail",
775 "Indexes do not fit in memory" if avg_idx < 95 else "",
776 )
777 except Exception as e:
778 _add(
779 results,
780 "Performance Efficiency",
781 "PERF14",
782 f"Cannot check index cache for {iid}: {e}",
783 "warn",
784 )
785
786 # DatabaseCursorsTimedOut
787 try:
788 dps = [
789 d["Sum"]
790 for d in cw.get_metric_statistics(
791 Namespace="AWS/DocDB",
792 MetricName="DatabaseCursorsTimedOut",
793 Dimensions=dim,
794 StartTime=start,
795 EndTime=end,
796 Period=86400,
797 Statistics=["Sum"],
798 ).get("Datapoints", [])
799 ]
800 total = sum(dps) if dps else 0
801 if total > 0:
802 _add(
803 results,
804 "Reliability",
805 "REL8",
806 f"{int(total)} cursor(s) timed out on {iid} in last 7 days",
807 "warn",
808 "Application may not be closing cursors properly",
809 )
810 else:
811 _add(results, "Reliability", "REL8", f"No cursor timeouts on {iid}", "pass")
812 except Exception as e:
813 _add(results, "Reliability", "REL8", f"Cannot check cursors for {iid}: {e}", "warn")
814
815 # AvailableMVCCIds (writer only)
816 if is_writer:
817 try:
818 dps = [
819 d["Minimum"]
820 for d in cw.get_metric_statistics(
821 Namespace="AWS/DocDB",
822 MetricName="AvailableMVCCIds",
823 Dimensions=dim,
824 StartTime=start,
825 EndTime=end,
826 Period=3600,
827 Statistics=["Minimum"],
828 ).get("Datapoints", [])
829 ]
830 if dps:
831 min_mvcc = min(dps)
832 pct = min_mvcc / 1_400_000_000 * 100
833 _add(
834 results,
835 "Reliability",
836 "REL9",
837 f"AvailableMVCCIds min: {min_mvcc:,.0f} ({pct:.0f}%)",
838 "fail" if pct < 25 else "warn" if pct < 50 else "pass",
839 (
840 "MVCC ID exhaustion risk -- investigate long-running transactions"
841 if pct < 50
842 else ""
843 ),
844 )
845 except Exception as e:
846 _add(results, "Reliability", "REL9", f"Cannot check MVCCIds: {e}", "warn")
847
848 return results
849
850
851# ---------------------------------------------------------------------------
852# Database-level checks (from pre-collected analysis JSON)
853# ---------------------------------------------------------------------------
854def run_db_checks(analysis_data):
855 results: list[dict] = []
856 if not analysis_data:
857 return results
858
859 total_indexes = 0
860 unused_indexes = 0
861 redundant = 0
862 low_cardinality = 0
863 low_card_names = []
864 large_docs = []
865 ttl_colls = []
866 total_data_size = 0
867 total_index_size = 0
868 total_unused_bytes = 0
869 bloated_colls = []
870 over_indexed_colls = []
871 compression_disabled = []
872 collscan_candidates = []
873 write_amp_colls = []
874
875 for db_name, collections in analysis_data.items():
876 if not isinstance(collections, dict):
877 continue
878 for coll_name, stats in collections.items():
879 if not isinstance(stats, dict) or "error" in stats:
880 continue
881 indexes = stats.get("indexes", [])
882 total_indexes += len(indexes)
883
884 for idx in indexes:
885 if idx.get("usage", {}).get("potential_unused"):
886 unused_indexes += 1
887 if idx.get("cardinality", {}).get("is_low"):
888 low_cardinality += 1
889 low_card_names.append(f"{db_name}.{coll_name}.{idx['name']}")
890 if idx.get("expireAfterSeconds") is not None:
891 if f"{db_name}.{coll_name}" not in ttl_colls:
892 ttl_colls.append(f"{db_name}.{coll_name}")
893
894 avg_obj = stats.get("avgObjSize", 0)
895 if avg_obj > 8192:
896 large_docs.append(f"{db_name}.{coll_name} ({avg_obj:,} bytes)")
897
898 # Redundant indexes (prefix subset)
899 ordered = [tuple(idx.get("ordered_fields", [])) for idx in indexes]
900 for i, a in enumerate(ordered):
901 for j, b in enumerate(ordered):
902 if i != j and len(a) > 0 and len(a) < len(b) and b[: len(a)] == a:
903 redundant += 1
904 break
905
906 total_data_size += stats.get("size", 0)
907 for idx in indexes:
908 total_index_size += idx.get("size", 0)
909
910 unused_info = stats.get("unusedStorageSize", {})
911 unused_pct = unused_info.get("unusedPercent", 0.0)
912 total_unused_bytes += unused_info.get("unusedBytes", 0)
913 if unused_pct > 30:
914 bloated_colls.append(f"{db_name}.{coll_name} ({unused_pct:.0f}%)")
915
916 if len(indexes) > 10:
917 over_indexed_colls.append(f"{db_name}.{coll_name} ({len(indexes)} indexes)")
918
919 comp = stats.get("compression", {})
920 if not comp.get("enabled", False):
921 compression_disabled.append(f"{db_name}.{coll_name}")
922
923 doc_count = stats.get("count", 0)
924 non_id = [idx for idx in indexes if idx.get("name") != "_id_"]
925 if doc_count > 100000 and len(non_id) == 0:
926 collscan_candidates.append(f"{db_name}.{coll_name} ({doc_count:,} docs)")
927
928 coll_data = stats.get("size", 0)
929 coll_idx = sum(idx.get("size", 0) for idx in indexes)
930 if coll_data > 0 and coll_idx > 2 * coll_data:
931 write_amp_colls.append(
932 f"{db_name}.{coll_name} (index {coll_idx / coll_data:.1f}x data)"
933 )
934
935 # Emit checks
936 if large_docs:
937 _add(
938 results,
939 "Performance Efficiency",
940 "PERF1",
941 f"{len(large_docs)} collection(s) with avg doc size > 8 KB",
942 "warn",
943 ", ".join(large_docs[:5]),
944 )
945 else:
946 _add(
947 results,
948 "Performance Efficiency",
949 "PERF1",
950 "All collections have avg doc size < 8 KB",
951 "pass",
952 )
953
954 if redundant > 0:
955 _add(
956 results,
957 "Performance Efficiency",
958 "PERF1b",
959 f"{redundant} redundant index(es) (prefix subsets)",
960 "warn",
961 )
962 else:
963 _add(results, "Performance Efficiency", "PERF1b", "No redundant indexes detected", "pass")
964
965 if low_cardinality > 0:
966 _add(
967 results,
968 "Performance Efficiency",
969 "PERF1c",
970 f"{low_cardinality} low cardinality index(es)",
971 "warn",
972 ", ".join(low_card_names[:5]),
973 )
974 else:
975 _add(
976 results,
977 "Performance Efficiency",
978 "PERF1c",
979 "No low cardinality indexes detected",
980 "pass",
981 )
982
983 if unused_indexes > 0:
984 _add(
985 results,
986 "Cost Optimization",
987 "COST3",
988 f"{unused_indexes} unused index(es) of {total_indexes} total",
989 "warn",
990 "Remove unused indexes to reduce write overhead and storage",
991 )
992 else:
993 _add(
994 results,
995 "Cost Optimization",
996 "COST3",
997 f"No unused indexes ({total_indexes} total)",
998 "pass",
999 )
1000
1001 if ttl_colls:
1002 _add(
1003 results,
1004 "Cost Optimization",
1005 "COST4",
1006 f"TTL indexes on {len(ttl_colls)} collection(s)",
1007 "pass",
1008 ", ".join(ttl_colls[:5]),
1009 )
1010 else:
1011 _add(
1012 results,
1013 "Cost Optimization",
1014 "COST4",
1015 "No TTL indexes found",
1016 "warn",
1017 "Consider TTL indexes for automatic data expiration",
1018 )
1019
1020 if total_data_size > 0:
1021 ratio = total_index_size / total_data_size * 100
1022 _add(
1023 results,
1024 "Performance Efficiency",
1025 "PERF8",
1026 f"Index-to-data ratio: {ratio:.0f}%",
1027 "warn" if ratio > 50 else "pass",
1028 "Indexes exceed 50% of data size" if ratio > 50 else "",
1029 )
1030
1031 if bloated_colls:
1032 _add(
1033 results,
1034 "Performance Efficiency",
1035 "PERF9",
1036 f"{len(bloated_colls)} collection(s) with >30% storage bloat",
1037 "warn",
1038 "Run compact command. " + ", ".join(bloated_colls[:5]),
1039 )
1040 else:
1041 _add(results, "Performance Efficiency", "PERF9", "No significant storage bloat", "pass")
1042
1043 if over_indexed_colls:
1044 _add(
1045 results,
1046 "Performance Efficiency",
1047 "PERF10",
1048 f"{len(over_indexed_colls)} collection(s) with >10 indexes",
1049 "warn",
1050 ", ".join(over_indexed_colls[:5]),
1051 )
1052 else:
1053 _add(results, "Performance Efficiency", "PERF10", "No over-indexed collections", "pass")
1054
1055 if compression_disabled:
1056 _add(
1057 results,
1058 "Sustainability",
1059 "SUST2",
1060 f"Compression disabled on {len(compression_disabled)} collection(s)",
1061 "warn",
1062 ", ".join(compression_disabled[:5]),
1063 )
1064 else:
1065 _add(results, "Sustainability", "SUST2", "Compression enabled on all collections", "pass")
1066
1067 if collscan_candidates:
1068 _add(
1069 results,
1070 "Performance Efficiency",
1071 "PERF15",
1072 f"{len(collscan_candidates)} large collection(s) with no secondary indexes",
1073 "warn",
1074 ", ".join(collscan_candidates[:5]),
1075 )
1076 else:
1077 _add(
1078 results,
1079 "Performance Efficiency",
1080 "PERF15",
1081 "All large collections have secondary indexes",
1082 "pass",
1083 )
1084
1085 if write_amp_colls:
1086 _add(
1087 results,
1088 "Performance Efficiency",
1089 "PERF16",
1090 f"{len(write_amp_colls)} collection(s) with index size > 2x data",
1091 "warn",
1092 ", ".join(write_amp_colls[:5]),
1093 )
1094 else:
1095 _add(
1096 results, "Performance Efficiency", "PERF16", "No excessive index-to-data ratio", "pass"
1097 )
1098
1099 return results
1100
1101
1102# ---------------------------------------------------------------------------
1103# Report generation
1104# ---------------------------------------------------------------------------
1105def generate_report(results, cluster_id, output_dir):
1106 output_dir = Path(output_dir)
1107 output_dir.mkdir(parents=True, exist_ok=True)
1108
1109 # JSON output
1110 json_path = output_dir / "wa_review_results.json"
1111 with open(json_path, "w") as f:
1112 json.dump(
1113 {
1114 "cluster": cluster_id,
1115 "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
1116 "checks": results,
1117 },
1118 f,
1119 indent=2,
1120 )
1121
1122 # Markdown summary
1123 md_path = output_dir / "wa_review_report.md"
1124 pillars: dict[str, list] = {}
1125 for r in results:
1126 pillars.setdefault(r["pillar"], []).append(r)
1127
1128 counts = {"pass": 0, "warn": 0, "fail": 0, "info": 0}
1129 for r in results:
1130 counts[r["status"]] = counts.get(r["status"], 0) + 1
1131
1132 lines = [
1133 f"# Well-Architected Review: {cluster_id}\n",
1134 f"**Date:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}\n",
1135 f"**Summary:** {counts['pass']} pass, {counts['warn']} warnings, "
1136 f"{counts['fail']} failures, {counts['info']} info\n",
1137 ]
1138
1139 status_icon = {"pass": "[PASS]", "warn": "[WARN]", "fail": "[FAIL]", "info": "[INFO]"}
1140 for pillar in [
1141 "Reliability",
1142 "Security",
1143 "Operational Excellence",
1144 "Cost Optimization",
1145 "Performance Efficiency",
1146 "Sustainability",
1147 ]:
1148 checks = pillars.get(pillar, [])
1149 if not checks:
1150 continue
1151 lines.append(f"\n## {pillar}\n")
1152 lines.append("| Status | Check | Detail |")
1153 lines.append("|--------|-------|--------|")
1154 for c in checks:
1155 icon = status_icon.get(c["status"], c["status"])
1156 label = c.get("label", "").replace("|", "/")
1157 detail = c.get("detail", "").replace("|", "/")
1158 lines.append(f"| {icon} | {label} | {detail} |")
1159
1160 with open(md_path, "w") as f:
1161 f.write("\n".join(lines) + "\n")
1162
1163 return json_path, md_path
1164
1165
1166# ---------------------------------------------------------------------------
1167# Collect database stats via pymongo (--uri)
1168# ---------------------------------------------------------------------------
1169def _redact_uri_creds(msg):
1170 """Strip any embedded mongodb credentials (user:pass@) from an error
1171 message so they are never printed to stdout/logs."""
1172 return re.sub(r"://[^/@\s]*@", "://<redacted>@", str(msg))
1173
1174
1175def collect_db_stats(uri, tls_ca_file=None, tls_allow_invalid_certs=False):
1176 """Connect to DocumentDB/MongoDB, collect collStats + indexStats for all
1177 collections across all databases. Returns analysis_data dict compatible
1178 with run_db_checks()."""
1179 try:
1180 import pymongo
1181 except ImportError:
1182 print("ERROR: pymongo required for --uri. Install with: pip install pymongo")
1183 sys.exit(1)
1184
1185 client = pymongo.MongoClient(
1186 uri,
1187 serverSelectionTimeoutMS=10000,
1188 tlsAllowInvalidCertificates=tls_allow_invalid_certs,
1189 **({"tlsCAFile": tls_ca_file} if tls_ca_file else {}),
1190 )
1191 analysis: dict[str, dict] = {}
1192 skip_dbs = {"admin", "local", "config"}
1193
1194 try:
1195 for db_name in client.list_database_names():
1196 if db_name in skip_dbs:
1197 continue
1198 db = client[db_name]
1199 collections: dict[str, dict] = {}
1200 for coll_name in db.list_collection_names():
1201 if coll_name.startswith("system."):
1202 continue
1203 try:
1204 stats = db.command("collStats", coll_name)
1205 idx_stats = list(db[coll_name].aggregate([{"$indexStats": {}}]))
1206
1207 indexes: list[dict] = []
1208 raw_indexes = list(db[coll_name].list_indexes())
1209 idx_usage = {s["name"]: s.get("accesses", {}).get("ops", 0) for s in idx_stats}
1210
1211 for idx in raw_indexes:
1212 name = idx["name"]
1213 key = idx.get("key", {})
1214 size = stats.get("indexSizes", {}).get(name, 0)
1215 ops = idx_usage.get(name, 0)
1216 ordered_fields = list(key.keys())
1217
1218 entry = {
1219 "name": name,
1220 "fields": key,
1221 "ordered_fields": ordered_fields,
1222 "size": size,
1223 "usage": {"ops": ops, "potential_unused": ops == 0 and name != "_id_"},
1224 "cardinality": {"is_low": False},
1225 }
1226 if idx.get("expireAfterSeconds") is not None:
1227 entry["expireAfterSeconds"] = idx["expireAfterSeconds"]
1228 indexes.append(entry)
1229
1230 comp_enabled = False
1231 comp_info = stats.get("compression", {})
1232 comp_enabled = comp_info.get("enabled", False) or comp_info.get("enable", False)
1233
1234 data_size = stats.get("size", 0)
1235 storage_size = stats.get("storageSize", 0)
1236 unused_bytes = (
1237 max(0, storage_size - data_size) if storage_size > data_size else 0
1238 )
1239 unused_pct = (unused_bytes / storage_size * 100) if storage_size > 0 else 0
1240
1241 collections[coll_name] = {
1242 "count": stats.get("count", 0),
1243 "size": data_size,
1244 "storageSize": storage_size,
1245 "avgObjSize": stats.get("avgObjSize", 0),
1246 "totalIndexSize": stats.get("totalIndexSize", 0),
1247 "indexes": indexes,
1248 "compression": {"enabled": comp_enabled},
1249 "unusedStorageSize": {
1250 "unusedBytes": unused_bytes,
1251 "unusedPercent": unused_pct,
1252 },
1253 }
1254 except Exception as e:
1255 collections[coll_name] = {"error": str(e)}
1256
1257 if collections:
1258 analysis[db_name] = collections
1259 except Exception as e:
1260 print(f" ERROR: Cannot connect to database: {_redact_uri_creds(e)}")
1261 return {}
1262 finally:
1263 client.close()
1264
1265 return analysis
1266
1267
1268# ---------------------------------------------------------------------------
1269# Main
1270# ---------------------------------------------------------------------------
1271def main():
1272 parser = argparse.ArgumentParser(description="DocumentDB Well-Architected Review")
1273 parser.add_argument("--cluster-id", required=True, help="DocumentDB cluster identifier")
1274 parser.add_argument("--region", required=True, help="AWS region")
1275 parser.add_argument(
1276 "--uri", default=None, help="MongoDB/DocumentDB connection URI for database-level checks"
1277 )
1278 parser.add_argument(
1279 "--analysis-data",
1280 default=None,
1281 help="Path to JSON file with database-level analysis (alternative to --uri)",
1282 )
1283 parser.add_argument(
1284 "--tls-ca-file",
1285 default=None,
1286 help="Path to CA bundle (e.g., global-bundle.pem) for TLS verification",
1287 )
1288 parser.add_argument(
1289 "--tls-allow-invalid-certs",
1290 action="store_true",
1291 default=False,
1292 help="Disable TLS certificate verification (not recommended)",
1293 )
1294 parser.add_argument("--output", default=".", help="Output directory (default: current)")
1295 args = parser.parse_args()
1296
1297 print(f"Running Well-Architected Review for {args.cluster_id} in {args.region}...")
1298
1299 # Infrastructure checks
1300 print(" Running infrastructure checks (AWS APIs)...")
1301 results = run_infra_checks(args.cluster_id, args.region)
1302 print(f" Infrastructure: {len(results)} checks completed")
1303
1304 # Database-level checks
1305 analysis = None
1306 if args.uri:
1307 print(f" Collecting database stats via pymongo...")
1308 analysis = collect_db_stats(args.uri, args.tls_ca_file, args.tls_allow_invalid_certs)
1309 n_colls = sum(len(v) for v in analysis.values())
1310 print(f" Collected stats for {n_colls} collections across {len(analysis)} databases")
1311 elif args.analysis_data:
1312 print(f" Loading database stats from {args.analysis_data}...")
1313 with open(args.analysis_data) as f:
1314 analysis = json.load(f)
1315
1316 if analysis:
1317 print(" Running database-level checks...")
1318 db_results = run_db_checks(analysis)
1319 results.extend(db_results)
1320 print(f" Database: {len(db_results)} checks completed")
1321 else:
1322 print(" Skipping database-level checks (no --uri or --analysis-data)")
1323
1324 # Generate report
1325 json_path, md_path = generate_report(results, args.cluster_id, args.output)
1326 print(f"\n Results: {json_path}")
1327 print(f" Report: {md_path}")
1328
1329 # Summary
1330 counts: dict[str, int] = {}
1331 for r in results:
1332 counts[r["status"]] = counts.get(r["status"], 0) + 1
1333 print(
1334 f"\n Total: {len(results)} checks -- "
1335 f"{counts.get('pass', 0)} pass, {counts.get('warn', 0)} warn, "
1336 f"{counts.get('fail', 0)} fail, {counts.get('info', 0)} info"
1337 )
1338
1339
1340if __name__ == "__main__":
1341 main()