AWS Reserved Instances by nornalized units to know how many you have total and how many you have reserved

check_aws_ri.py

#!/usr/bin/python
import argparse
import logging
from pprint import pformat

import boto3

logging.basicConfig(level="INFO")
logger = logging.getLogger('ec2-check-normalized-units')


class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'


NORMALIZED_UNITS = {
    "nano": 0.25,
    "micro": 0.5,
    "small": 1,
    "medium": 2,
    "large": 4,
    "xlarge": 8,
    "2xlarge": 16,
    "3xlarge": 24,
    "4xlarge": 32,
    "6xlarge": 48,
    "8xlarge": 64,
    "9xlarge": 72,
    "10xlarge": 80,
    "12xlarge": 96,
    "16xlarge": 128,
    "18xlarge": 144,
    "24xlarge": 192,
    "32xlarge": 256
}


def get_per_class_normalized_units(instance_report):
    normalized_units = {}
    for instance_type, instance_count in instance_report.items():
        instance_split = instance_type.split('.')
        instance_family = instance_split[0]
        instance_size = instance_split[1]
        norm_unit = convert_to_normalized_units(instance_size, instance_count)
        normalized_units[instance_family] = normalized_units.get(instance_family, 0) + norm_unit
    return normalized_units


def convert_to_normalized_units(instance_size, instance_count):
    return NORMALIZED_UNITS[instance_size] * instance_count


def get_ec2_instances(ec2_client):
    ec2_instances = ec2_client.describe_instances()
    running_ec2_instances = {}
    running_ec2_spot_instances = {}
    for reservation in ec2_instances['Reservations']:
        for instance in reservation['Instances']:
            if instance['State']['Name'] != "running":
                logger.debug("Disqualifying instance %s: not running\n" % (instance['InstanceId']))
            elif "InstanceLifecycle" in instance:
                if instance['InstanceLifecycle'] == "spot":
                    instance_type = instance['InstanceType']
                    running_ec2_spot_instances[instance_type] = running_ec2_spot_instances.get(instance_type, 0) + 1
            else:
                instance_type = instance['InstanceType']
                logger.debug("Running instance: {}\n".format(pformat(instance)))
                running_ec2_instances[instance_type] = running_ec2_instances.get(instance_type, 0) + 1
    return running_ec2_instances, running_ec2_spot_instances


def get_ec2_reserved_instances(ec2_client):
    ec2_reserved_instances = {}
    ec2_reservations = ec2_client.describe_reserved_instances()

    for ri in ec2_reservations['ReservedInstances']:
        if ri['State'] != "active":
            logger.debug("Excluding reserved instances %s: no longer active\n" % (ri['ReservedInstancesId']))
        else:
            instance_type = ri['InstanceType']
            instance_signature = (instance_type)
            ec2_reserved_instances[instance_signature] = ec2_reserved_instances.get(instance_signature,
                                                                                    0) + ri['InstanceCount']
    ec2_reserved = dict(ec2_reserved_instances)
    return ec2_reserved


def main():
    parser = argparse.ArgumentParser(description='Cross reference existing ec2 reservations to current instances.')
    parser.add_argument('--log', default="INFO", help='Change log level (default: WARN)')
    parser.add_argument('--region', default='us-east-1', help='AWS Region to connect to')

    args = parser.parse_args()

    logger.debug("boto version = %s", boto3.__version__)
    ec2_client = boto3.client('ec2', region_name=args.region)
    running_ec2, running_spot = get_ec2_instances(ec2_client)
    logger.debug("\nRunning instances: {}\n".format(pformat(running_ec2)))
    logger.debug("\nRunning SPOT instances: {}\n".format(pformat(running_spot)))
    reserved_ec2 = get_ec2_reserved_instances(ec2_client)

    logger.debug("\nReserved instances: {}".format(pformat(reserved_ec2)))

    nu_running = get_per_class_normalized_units(running_ec2)
    nu_reserved = get_per_class_normalized_units(reserved_ec2)
    logger.debug(bcolors.HEADER + "Normalized running instances" + bcolors.ENDC)
    logger.debug(pformat(nu_running))

    logger.debug(bcolors.HEADER + "Normalized reserved instances" + bcolors.ENDC)
    logger.debug(pformat(nu_reserved))

    missing_nu = {key: nu_running[key] - nu_reserved.get(key, 0) for key in nu_running.keys()}

    output = "\n" + bcolors.BOLD + bcolors.HEADER + \
             "Required instance normalized units for 100% reservation" + \
             bcolors.ENDC + "\n"
    for family, miss_units in missing_nu.items():
        if miss_units > 0:
            color = bcolors.WARNING
            miss_units = str(miss_units) + bcolors.OKGREEN + "   (" + str(
                int(miss_units / 4)) + "*large instance)" + bcolors.ENDC
        else:
            color = bcolors.OKBLUE
        output += bcolors.BOLD + '[' + family + ']   --- \t' + bcolors.ENDC + color + str(
            miss_units) + bcolors.ENDC + '\n'
    logger.info(output)


if __name__ == "__main__":
    main()