find_stray_ec2.py

import datetime
from operator import itemgetter
from pprint import pprint

import boto3


def chunks(lst, n):
    """Yield successive n-sized chunks from lst."""
    for i in range(0, len(lst), n):
        yield lst[i:i + n]


client = boto3.client('ec2')

response = client.describe_instances(MaxResults=9999)
ami_names = {}
instances = []
for reservations in response['Reservations']:
    for instance in reservations['Instances']:
        # print(instance)
        state = instance['State']['Name']
        if state == 'terminated':
            continue
        start_at: datetime.datetime = instance['LaunchTime']
        start_at_kst = start_at.astimezone(datetime.timezone(datetime.timedelta(hours=9)))
        tags = {i['Key']: i['Value'] for i in instance.get('Tags', [])}
        instance_name = tags.get('Name', '')
        image_id = instance['ImageId']
        if image_id not in ami_names:
            name = client.describe_images(ImageIds=[image_id, ], )['Images'][0]['Name']
            ami_names[image_id] = name

        instances.append(
            {'id': instance['InstanceId'], 'name': instance_name, 'ami_name': ami_names[image_id], 'state': state,
             'start_at': start_at_kst}
        )

eb_instances = list(filter(lambda e: e['ami_name'].startswith('aws-elasticbeanstalk'), instances))
# pprint(eb_instances)
instance_asg = {}
if eb_instances:
    client = boto3.client('autoscaling')
    for instance_chunks in chunks(eb_instances, 50):
        asgs = client.describe_auto_scaling_instances(InstanceIds=list(map(itemgetter('id'), instance_chunks)))
        for asg in asgs['AutoScalingInstances']:
            instance_asg[asg['InstanceId']] = asg['AutoScalingGroupName']

none_eb_instances = []
for instance in eb_instances:
    if instance['id'] not in instance_asg:
        none_eb_instances.append(instance)

none_eb_instances.sort(key=itemgetter('state'))
pprint(none_eb_instances)

import orjson

print(orjson.dumps(none_eb_instances).decode())