CloudWatch CPU/Disk IOPs dashboard JSON generator for ASG-backed EC2 ECS clusters

cluster_dash.py

import boto3
import json

# assumes ASG has tag Name="ECS Cluster - cluster_name"
cluster_name = "some-clsuter"

asgs = [
  {
    "name": asg["AutoScalingGroupName"],
    "instance_type": asg["Instances"][0]["InstanceType"],
  }
  for page in boto3.client("autoscaling").get_paginator('describe_auto_scaling_groups').paginate()
  for asg in page['AutoScalingGroups']
  if len(asg["Instances"]) > 0 \
    and any(tag["Key"] == "Name" for tag in asg["Tags"]) \
    and [tag for tag in asg["Tags"] if tag["Key"] == "Name"][0]["Value"].split(" - ")[-1] == cluster_name
]

print("CPU Usage")
print(json.dumps({
    "metrics": [["AWS/EC2", "CPUUtilization", "AutoScalingGroupName", asg["name"], {"label": asg["instance_type"]}] for asg in asgs],
    "view": "timeSeries",
    "stacked": False,
    "region": "us-east-1",
    "stat": "p95",
    "period": 900
}))

id1 = 1
id2 = 1
metrics = []
for m in asgs:
  label = m["instance_type"]
  asg_name = m["name"]
  metrics.append(
    [ {"expression": f"(m{id1}+m{id1+1})/m{id1+2}/60", "label": f"{label}", "id": f"e{id2}"} ]
  )
  id2 +=1
  metrics.append(
    [ "AWS/EC2", "EBSReadOps", "AutoScalingGroupName", asg_name, {"id": f"m{id1}", "visible": False} ]
  )
  metrics.append(
    [ "AWS/EC2", "EBSWriteOps", "AutoScalingGroupName", asg_name, {"id": f"m{id1+1}", "visible": False} ]
  )
  metrics.append(
    [ "AWS/EC2", "EBSWriteOps", "AutoScalingGroupName", asg_name, {"id": f"m{id1+2}", "visible": False, "stat": "SampleCount"} ]
  )
  id1 += 3

print("Disk Usage")
print(json.dumps({"metrics": metrics, "view": "timeSeries", "stacked": False, "region": "us-east-1", "stat": "Sum", "period": 60}))