Instantly share code, notes, and snippets.

Embed
What would you like to do?
Ersatz Cost Allocation Tags
import csv
import io
import tempfile
import zipfile
import boto3
import yaml
from botocore.exceptions import ClientError
BUCKET = COST ALLOCATION REPORT BUCKET
ACCOUNT = AWS ACCOUNT NUMBER
def get_ec_clients_and_resources():
response = boto3.client('ec2').describe_regions()
for region in response['Regions']:
region_name_ = region['RegionName']
regions.append(region_name_)
botos[region_name_] = {}
botos[region_name_]['client'] = boto3.client('ec2', region_name=region_name_)
botos[region_name_]['resource'] = boto3.resource('ec2', region_name=region_name_)
def load_department_info():
user_departments = {}
yml = yaml.load(open('users_in_dept.yml'))
for dept in yml.keys():
for entry in yml[dept]:
if isinstance(entry, str):
user_departments[entry] = dept
return user_departments
def extract_iam_user(creator):
if creator.startswith('IAMUser'):
return creator.split(':')[2]
else:
return creator
def tag(instance, key, value):
instance.create_tags(
DryRun=False,
Tags=[
{
'Key': key,
'Value': str(value)
}
]
)
def get_department_for_user(user):
if user in department_names:
department_ = department_names[user]
elif "AssumedRole" in user:
department_ = None
else:
raise Exception("USER NOT IN DEPARTMENT: {}".format(user))
return department_
def apply_tags(availability_zone, cost_allocation_tag, ec2_instance_id):
region = availability_zone[:-1]
ec2 = botos[region]['resource']
instance = ec2.Instance(ec2_instance_id)
try:
user = extract_iam_user(cost_allocation_tag)
state = instance.state
state_name = state['Name']
if state_name == 'running' or state_name == 'stopped':
tag(instance, 'ersatz:createdBy', cost_allocation_tag)
tag(instance, 'ersatz:iam_username', user)
tag(instance, 'ersatz:department', get_department_for_user(user))
print("{} {}".format(availability_zone, ec2_instance_id))
except ClientError as e:
message = str(e)
if " does not exist" not in message:
print("Unexpected error:", str(e))
except AttributeError as ae:
if not "has no attribute 'get'" in str(ae):
print("Unexpected error: {}".format(ae))
department_names = load_department_info()
botos = {}
regions = []
processed = {}
def discover_instances(botos):
for key in botos:
client = botos[key]['client']
response = client.describe_instances()
reservations = response['Reservations']
for reservation in reservations:
for instance in reservation['Instances']:
instance_id_ = instance['InstanceId']
if 'Tags' in instance:
for tag in instance['Tags']:
if tag['Key'] == 'ersatz:createdBy':
value_ = tag['Value']
processed[instance_id_] = value_
def lambda_handler(event, context):
get_ec_clients_and_resources()
discover_instances(botos)
client = boto3.client('s3')
response = client.list_objects_v2(Bucket=BUCKET,
Prefix="{}-aws-billing-detailed-line-items-with-resources-and-tags".format(ACCOUNT))
key_ = response['Contents'][-1]['Key']
csv_file = key_[:-4]
csv_zip = tempfile.mkstemp()[1] + 'csv.zip'
client.download_file(BUCKET, key_, csv_zip)
with zipfile.ZipFile(csv_zip) as myzip:
with myzip.open(csv_file) as myfile:
for row in csv.DictReader(io.TextIOWrapper(myfile, encoding='UTF-8', newline=''),
delimiter=',',
quotechar='"'):
cost_allocation_tag = row['aws:createdBy']
zone_ = row["AvailabilityZone"]
id_ = row["ResourceId"]
linked_account_id_ = row["LinkedAccountId"]
if id_ and not id_ in processed:
if cost_allocation_tag and zone_ and (linked_account_id_ == ACCOUNT) and id_.startswith("i-"):
apply_tags(zone_, cost_allocation_tag, id_)
processed[id_] = cost_allocation_tag
11259: # Marketing
- bob.smith@inittech.com
11260: # Engineering
- johnny.appleseed@inittech.com
- bobby.schmitt@inittech.com
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment