Query A100 AWS nodes CPU utilization

query_utilization.py

import boto3
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional

cloudwatch = boto3.client("cloudwatch")
ec2 = boto3.resource("ec2")

def ec2_get_instances(filter_name, filter_value):
    return ec2.instances.filter(Filters=[{'Name': filter_name,
                                          'Values': [filter_value]}])


def ec2_instances_of_type(instance_type='t4g.2xlarge'):
    return ec2_get_instances('instance-type', instance_type)


def get_instance_name(instance) -> Optional[str]:
    if instance.tags is None:
        return None
    for tag in instance.tags:
        if tag['Key'] == 'Name':
            return tag['Value']
    return None

def get_cpu_utilization(instance_id = "i-02e8329783a0e77b5",
                        period=4*3600,
                        delta=timedelta(days=7)):
    now = datetime.now().replace(microsecond=0, second=0, minute=0)
    response = cloudwatch.get_metric_statistics(
                        Namespace='AWS/EC2',
                MetricName='CPUUtilization',
                Dimensions=[
                    {
                    'Name': 'InstanceId',
                    'Value': instance_id,
                    },
                ],
                StartTime=now - delta,
                EndTime=now,
                Period=period,
                Statistics=[
                    'Average',
                ],
                Unit='Percent'
            )
    return pd.DataFrame({"timestamp": [pd.to_datetime(x["Timestamp"]) for x in response["Datapoints"]],
                         f"percent-{instance_id}": [x["Average"] for x in response["Datapoints"]]}).sort_values("timestamp", ignore_index=True)


def main() -> None:
    p24d_instances = ec2_instances_of_type("p4d.24xlarge")

    combined = None
    for instance in p24d_instances:
        print(instance.id, get_instance_name(instance))
        df = get_cpu_utilization(instance.id)
        df.to_csv(f"a100-utilization-{instance.id}.csv")
        combined = combined.merge(df,how="outer") if combined is not None else df
        print(df)
    combined.to_csv("a100-utilization-combined.csv")

if __name__ == "__main__":
    main()