Skip to content

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()
@jutler

This comment has been minimized.

Copy link

jutler commented Apr 9, 2019

@gmr this is awesome, just the thing i was looking for. I'm very new to python so still learning. Just 2 things i wanted to ask. Is there a way to include Elastic IP's in this code? Also how could you implement a dryrun in this so we can see what changes would me made without actually making them in the first instance?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.