aws-tag-a-day

A tool for simplifying multiple people fixing tags by proposing a tag a day for resources with those missing tags.

Installation

pip install aws-tag-a-day

Quickstart

# Generate configuration file
mkdir -p ~/.config/tagaday/
cat > ~/.config/tagaday/config.yml <<EOY
dynamodb-table-name: 'tag-proposals'
dynamodb-table-region: 'eu-west-2'
services:
- rds
- ec2
- s3
- emr
regions:
- us-east-1
- us-west-1
- eu-west-1
- ap-southeast-1
- ap-northeast-1
required-tags:
- Project
- Owner
- Name
- Service
- Availability
EOY

# Create dynamodb table defined in the above config file.
# If the table exists already, the utility will not overwrite it.
tag-a-day-initialise

# Start proposing tags
tag-a-day

Motivation

There are a lot of tools for handling batch tagging, but not many tools for aiding in filling in empty tags on a large scale.

The suggested workflow is:

1. Create configuration file.
2. Create a DynamoDB table by running tag-a-day-initialise. This is not a destructive operation, and existing tables will not be modified or removed. The utility will throw an error if the table already exists.
3. Have 1 or more people start using tag-a-day.
4. Reconcile any divergent or duplicate tags by running tag-a-day-reconcile, and discuss any duplicate tagging suggestions.
5. Apply the tags to the resources.

NOTE: Steps 4 and 5 are not yet implemented by this utility.

Reference

Supported Services

• emr
• ec2
• s3
• rds

Configuration File & CLI options

Extending

aws-tag-a-day is built on a plugin architecture, using entry_point in setuptools.

To add more TagHandlers, you can either add classes to this repo, or create a new python package with its own setup.py, and hook into the plugin architecture using the tag_a_day.tag_handlers entrypoint.

1. Create a new class, inheriting from tag_a_day.services.service.Service

   from tag_a_day.services.service import Service
   
   class CustomTagHandler(Service): pass
   

2. Set a unique name for the Tag handler

   from tag_a_day.services.service import Service
   
   class CustomTagHandler(Service):
      name='custom_handler'
   

3. Create two stub methods, resources and handler matching the signatures below:

   from tag_a_day.services.service import Service
   
   class VpcTagHandler(Service):
      name='ec2_vpc'
   
      def resources(self, session):
        pass
   
      def handler(self, resource, expected_tags, region, session, cache, proposals):
        pass
   

4. Implement resources(...) to return an iterable. If using boto3 resources, this should look like:

     def resources(self, session):
       return session.resource('ec2').vpc.all()
   

   If using boto3 client, don't forget to implement pagination, and should look like:

     def resources(self, session):
       ec2 = session.client('ec2')
       paginator = ec2.get_paginator('describe_vpcs')
   
       for page in paginator.paginate():
         for vpc in page:
           yield vpc 
   

5. Implement handle(...) to yield a payload describing the tag proposal (example is using boto3.resources):

     def handle(self, vpc, expected_tags, region, session, cache, proposals):
       # This boilerplate logic will handle checking the tags which have already been
       # evaluated for this user.
       evaluated_tags = self._progress.evaluated_tags(vpc.vpc_id)
       vpc_info, missing_tags = \
           self._build_tag_sets(expected_tags, evaluated_tags, vpc.tags)
   
       # Check if the user has proposed values for all the missing tags 
       if self._progress.has_finished(vpc.vpc_id, expected_tags):
           # Print a skip message
           self._skip(vpc.vpc_id)
           return
   
       if any(missing_tags):
         # Print information about this resource, which could be useful 
         # to provide context around tagging.
         self._print_table(
           ("VpcID", vpc.vpc_id),
           *vpc_info
         )
   
         # Allow the user to skip auditing this resource       
         if self._user_skip():
           return
   
         # Build our user prompt to ask for new tags
         tag_prompt = self._build_tag_prompt(missing_tags)
         for tag_key in missing_tags:
           # Yield a proposal for a new tag key/value pair for the given
           # resource id.
           yield {
             'resource_id': vpc.vpc_id,
             'tag_key': tag_key,
             'tag_value': tag_prompt(tag_key),
           }
   

6. Finally, add the entrypoint by extending setup.py:

   from setuptools import setup
   
   setup(
     ...,
     entry_points={
       'tag_a_day.tag_handlers': [
         'vpc = my_package.my_module:VPCTagHandler',
       ],
     }
   )