import boto3 import botocore.exceptions from datetime import datetime, timedelta from pprint import pprint import csv STATISTICS_PERIOD_DAYS = 90 METRIC_PERIOD_SECONDS = 60 MIN_THROUGHPUT = 125 MAX_THROUGHPUT = 1000 MIN_IOPS = 3000 MAX_IOPS = 16000 RESERVATION_MODIFIER = 1.1 def get_ebs_combined_metric_sum_max( client, volume_id, start_time, end_time, combined_metrics ): namespace = "AWS/EBS" base_metrics_queries = [] metric_ids = [] for metric in combined_metrics: metric_id = "c" + metric metric_ids.append(metric_id) base_metrics_queries.append( { "Id": metric_id, "MetricStat": { "Metric": { "Namespace": namespace, "MetricName": metric, "Dimensions": [ {"Name": "VolumeId", "Value": volume_id}, ], }, "Period": METRIC_PERIOD_SECONDS, "Stat": "Sum", }, "ReturnData": False, } ) expression = "(" + " + ".join(metric_ids) + ")" aggregated_metric = { "Id": "combined_sum", "Expression": expression, "Label": "combined_sum", "ReturnData": True, } response = client.get_metric_data( MetricDataQueries=base_metrics_queries + [aggregated_metric], StartTime=start_time, EndTime=end_time, ) max_value = 0 if "Values" in response["MetricDataResults"][0]: if len(response["MetricDataResults"][0]["Values"]) > 0: max_value = max( [ metric_sum / METRIC_PERIOD_SECONDS for metric_sum in response["MetricDataResults"][0]["Values"] ] ) return max_value def get_max_thr_iops_metrics(client, volume_id, start_time, end_time): throughput_metrics = [ "VolumeReadBytes", "VolumeWriteBytes", ] iops_metrics = [ "VolumeReadOps", "VolumeWriteOps", ] result = {} throughput_bytes = get_ebs_combined_metric_sum_max( client, volume_id, start_time, end_time, throughput_metrics ) result["Throughput"] = throughput_bytes / (1024 * 1024) result["Iops"] = get_ebs_combined_metric_sum_max( client, volume_id, start_time, end_time, iops_metrics ) return result def get_limited_values(value, min, max, modifier): limited_value = value * modifier if limited_value < min: limited_value = min elif limited_value > max: limited_value = max return int(limited_value) def get_optimizer_volume_data(client, volume_arn): response = client.get_ebs_volume_recommendations(volumeArns=[volume_arn]) if len(response["volumeRecommendations"]) > 0: if response["volumeRecommendations"][0]["finding"] == "NotOptimized": return response["volumeRecommendations"][0]["volumeRecommendationOptions"][ 0 ]["configuration"] return {} def get_ec2_instance_name(client, id): response = client.describe_instances(InstanceIds=[id]) tags = response["Reservations"][0]["Instances"][0]["Tags"] for tag in tags: if tag["Key"] == "Name": return tag["Value"] return "" def get_csv_row(client, volume_desc, volume_update, region): instance_id = "" instance_name = "" if "Attachments" in volume_desc: if len(volume_desc["Attachments"]) > 0: instance_id = volume_desc["Attachments"][0]["InstanceId"] instance_name = get_ec2_instance_name(client, instance_id) return { "VolumeId": volume_update["VolumeId"], "CurrentIops": volume_update["Iops"]["Defined"], "CurrentThrougput": volume_update["Throughput"]["Defined"], "NewIops": int(volume_update["Iops"]["Recommended"]), "NewThrougput": int(volume_update["Throughput"]["Recommended"]), "InstanceId": instance_id, "InstanceName": instance_name, "Region": region, } def get_volume_recommendations(volume_desc, region, start_time, end_time): cw_client = boto3.client("cloudwatch", region_name=region) optimizer_client = boto3.client("compute-optimizer", region_name=region) iops = volume_desc["Iops"] if "Iops" in v else MIN_IOPS throughput = volume_desc["Throughput"] if "Throughput" in v else MIN_THROUGHPUT metrics = get_max_thr_iops_metrics( cw_client, volume_desc["VolumeId"], start_time, end_time ) recommended_thr = get_limited_values( metrics["Throughput"], MIN_THROUGHPUT, MAX_THROUGHPUT, RESERVATION_MODIFIER, ) recommended_iops = get_limited_values( metrics["Iops"], MIN_IOPS, MAX_IOPS, RESERVATION_MODIFIER, ) volume_arn = "arn:aws:ec2:{}:{}:volume/{}".format( region, account_id, volume_desc["VolumeId"] ) optimizer_figures = get_optimizer_volume_data(optimizer_client, volume_arn) # If Compute Optimizer values are greater, take them as the recommendation. if len(optimizer_figures) > 0: if optimizer_figures["volumeBaselineIOPS"] != recommended_iops: print( f"{v['VolumeId']} | IOPS: current {iops} | optimizer {optimizer_figures['volumeBaselineIOPS']} | script {recommended_iops} " ) if optimizer_figures["volumeBaselineThroughput"] != recommended_thr: print( f"{v['VolumeId']} | Throughput: current {throughput} | optimizer {optimizer_figures['volumeBaselineThroughput']} | script {recommended_thr}" ) if optimizer_figures["volumeBaselineIOPS"] > recommended_iops: recommended_iops = optimizer_figures["volumeBaselineIOPS"] if optimizer_figures["volumeBaselineThroughput"] > recommended_thr: recommended_thr = optimizer_figures["volumeBaselineThroughput"] change_thr = (recommended_thr * 100 / throughput) - 100 change_iops = (recommended_iops * 100 / iops) - 100 if change_iops > 0 or change_thr > 0: pprint(f"{volume_desc['VolumeId']} | increase in parameters") if recommended_thr == throughput and recommended_iops == iops: return {} volume_update = { "VolumeId": volume_desc["VolumeId"], "Region": region, "Throughput": { "Defined": throughput, "Measured": metrics["Throughput"], "Recommended": recommended_thr, "Change": change_thr, }, "Iops": { "Defined": iops, "Measured": metrics["Iops"], "Recommended": recommended_iops, "Change": change_iops, }, "Optimizer": optimizer_figures, } return volume_update if __name__ == "__main__": regions = [ "us-east-1", "us-west-2", ] dry_run = True export_to_csv = True account_id = boto3.client("sts").get_caller_identity().get("Account") exported_rows = [] for region in regions: ec2_client = boto3.client("ec2", region_name=region) cw_client = boto3.client("cloudwatch", region_name=region) optimizer_client = boto3.client("compute-optimizer", region_name=region) response = ec2_client.describe_volumes() volumes = response["Volumes"] while "NextToken" in response: response = ec2_client.describe_volumes(NextToken=response["NextToken"]) volumes.extend(response["Volumes"]) end_time = datetime.utcnow() start_time = end_time - timedelta(days=STATISTICS_PERIOD_DAYS) for v in response["Volumes"]: if v["VolumeType"] == "gp3": volume_update = get_volume_recommendations( v, region, start_time, end_time ) if len(volume_update) > 0: # pprint(volume_update) if export_to_csv: exported_rows.append( get_csv_row(ec2_client, v, volume_update, region) ) try: modify = ec2_client.modify_volume( VolumeId=volume_update["VolumeId"], VolumeType="gp3", Iops=int(volume_update["Iops"]["Recommended"]), Throughput=int(volume_update["Throughput"]["Recommended"]), DryRun=dry_run, ) except botocore.exceptions.ClientError as error: if error.response["Error"]["Code"] == "DryRunOperation": print(f"{volume_update['VolumeId']} DryRun operation") else: raise error if export_to_csv: keys = exported_rows[0].keys() with open("volume_updates.csv", "w", newline="") as output_file: dict_writer = csv.DictWriter(output_file, keys) dict_writer.writeheader() dict_writer.writerows(exported_rows)