Instantly share code, notes, and snippets.

Embed
What would you like to do?
A lambda function that will copy EC2 tags to all related Volumes and Network Interfaces. A full writeup can be found on my site http://mlapida.com/thoughts/tagging-and-snapshotting-with-lambda
from __future__ import print_function
import json
import boto3
import logging
#setup simple logging for INFO
logger = logging.getLogger()
logger.setLevel(logging.ERROR)
#define the connection region
ec2 = boto3.resource('ec2', region_name="us-west-2")
#Set this to True if you don't want the function to perform any actions
debugMode = False
def lambda_handler(event, context):
#List all EC2 instances
base = ec2.instances.all()
#loop through by running instances
for instance in base:
#Tag the Volumes
for vol in instance.volumes.all():
#print(vol.attachments[0]['Device'])
if debugMode == True:
print("[DEBUG] " + str(vol))
tag_cleanup(instance, vol.attachments[0]['Device'])
else:
tag = vol.create_tags(Tags=tag_cleanup(instance, vol.attachments[0]['Device']))
print("[INFO]: " + str(tag))
#Tag the Network Interfaces
for eni in instance.network_interfaces:
#print(eni.attachment['DeviceIndex'])
if debugMode == True:
print("[DEBUG] " + str(eni))
tag_cleanup(instance, "eth"+str(eni.attachment['DeviceIndex']))
else:
tag = eni.create_tags(Tags=tag_cleanup(instance, "eth"+str(eni.attachment['DeviceIndex'])))
print("[INFO]: " + str(tag))
#------------- Functions ------------------
#returns the type of configuration that was performed
def tag_cleanup(instance, detail):
tempTags=[]
v={}
for t in instance.tags:
#pull the name tag
if t['Key'] == 'Name':
v['Value'] = t['Value'] + " - " + str(detail)
v['Key'] = 'Name'
tempTags.append(v)
#Set the important tags that should be written here
elif t['Key'] == 'Application Owner':
print("[INFO]: Application Owner Tag " + str(t))
tempTags.append(t)
elif t['Key'] == 'Cost Center':
print("[INFO]: Cost Center Tag " + str(t))
tempTags.append(t)
elif t['Key'] == 'Date Created':
print("[INFO]: Date Created Tag " + str(t))
tempTags.append(t)
elif t['Key'] == 'Requestor':
print("[INFO]: Requestor Tag " + str(t))
tempTags.append(t)
elif t['Key'] == 'System Owner':
print("[INFO]: System Owner Tag " + str(t))
tempTags.append(t)
else:
print("[INFO]: Skip Tag - " + str(t))
print("[INFO] " + str(tempTags))
return(tempTags)
@n2taylor

This comment has been minimized.

Copy link

n2taylor commented May 31, 2018

I borrowed this code. Thanks. Here's a condensed version for py3.6 with less logging:

import boto3


def lambda_handler(event, context):
    is_test = context.function_name == 'test'  # this value is injected by SAM local
    instances = boto3.resource('ec2').instances.all()

    copyable_tag_keys = ["Application Owner", "Cost Center", "Date Created", "Requestor", "System Owner"]

    for instance in instances:
        copyable_tags = [t for t in instance.tags
                         if t["Key"] in copyable_tag_keys] if instance.tags else []
        if not copyable_tags:
            continue

        # Tag the EBS Volumes
        print(f"{instance.instance_id}: {instance.tags}")
        for vol in instance.volumes.all():
            print(f"{vol.attachments[0]['Device']}: {copyable_tags}")
            if not is_test:
                vol.create_tags(Tags=copyable_tags)

        # Tag the Elastic Network Interfaces
        for eni in instance.network_interfaces:
            print(f"eth{str(eni.attachment['DeviceIndex'])}: {copyable_tags}")
            if not is_test:
                eni.create_tags(Tags=copyable_tags)
@gmr

This comment has been minimized.

Copy link

gmr commented Jan 8, 2019

And another variation adding elb and elbv2 eni tagging. I took out the lambda bits, but it'd be easy to add them back.

import boto3

COPYABLE = ["Service", "Environment", "Team", "Name"]


def ec2():
    print('Processing EC2 Instances')

    instances = boto3.resource('ec2').instances.all()
    for instance in instances:
        tags = [t for t in instance.tags or [] if t['Key'] in COPYABLE]
        if not tags:
            continue

        # Tag the EBS Volumes
        for vol in instance.volumes.all():
            print('Updating tags for {}'.format(vol.id))
            vol.create_tags(Tags=tags)

        # Tag the Elastic Network Interfaces
        for eni in instance.network_interfaces:
            print('Updating tags for {}'.format(eni.id))
            eni.create_tags(Tags=tags)


def elb():
    print('Processing ELB Instances')

    def filter(i):
        return (i.get('RequesterId') == 'amazon-elb' and
                i['Description'].startswith('ELB') and
                '/' not in i['Description'])

    tags = _get_elb_tags('elb')
    for interface in _network_interfaces(filter):
        name = interface['Description'].split(' ')[1]
        if name not in tags:
            continue
        _tag_network_interface(interface['NetworkInterfaceId'], tags[name])


def elbv2():
    print('Processing ELBv2 Instances')

    def filter(i):
        return (i.get('RequesterId') == 'amazon-elb' and
                i['Description'].startswith('ELB') and
                '/' in i['Description'])

    tags = _get_elb_tags('elbv2')
    for interface in _network_interfaces(filter):
        name = interface['Description'].split('/')[1]
        if name not in tags:
            continue
        _tag_network_interface(interface['NetworkInterfaceId'], tags[name])


def _get_elb_tags(name='elb'):
    if name == 'elb':
        page_name = 'LoadBalancerDescriptions'
        key = 'LoadBalancerName'
        kwname = 'LoadBalancerNames'
    elif name == 'elbv2':
        page_name = 'LoadBalancers'
        key = 'LoadBalancerArn'
        kwname = 'ResourceArns'
    else:
        raise ValueError('Invalid name: {}'.format(name))

    tags = {}
    client = boto3.client(name)
    paginator = client.get_paginator('describe_load_balancers')
    for page in paginator.paginate():
        for lb in page[page_name]:
            response = client.describe_tags(**{kwname: [lb[key]]})
            lb_tags = [item for sublist in
                       [r.get('Tags', []) for r in response['TagDescriptions']]
                       for item in sublist]
            tags[lb['LoadBalancerName']] = [t for t in lb_tags if
                                            t['Key'] in COPYABLE]
            tags[lb['LoadBalancerName']].append(
                {'Key': 'Name', 'Value': lb['LoadBalancerName']})
    return tags


def _network_interfaces(filter=None):
    client = boto3.client('ec2')
    paginator = client.get_paginator('describe_network_interfaces')
    for page in paginator.paginate():
        for interface in page['NetworkInterfaces']:
            if filter and not filter(interface):
                continue
            yield interface


def _tag_network_interface(eni_id, tags):
    print('Updating tags for {}'.format(eni_id))
    ec2 = boto3.resource('ec2')
    eni = ec2.NetworkInterface(eni_id)
    eni.create_tags(Tags=tags)


def main():
    ec2()
    elb()
    elbv2()


if __name__ == '__main__':
    main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment