aws-count-tag-names.py

#!/usr/bin/env python
"""
Using boto3, scan all AWS resources in the current account, and produce a report
detailing all of the distinct tag names and the number of resources having each
one.

If you have ideas for improvements, or want the latest version, it's at:
<https://github.com/jantman/misc-scripts/blob/master/aws-count-tag-names.py>

Copyright 2016 Jason Antman <jason@jasonantman.com> <http://www.jasonantman.com>
Free for any use provided that patches are submitted back to me.

REQUIREMENTS:

* boto3
* texttable

    pip install boto3 texttable

CHANGELOG:
2017-06-27 Jason Antman <jason@jasonantman.com>:
  - initial version of script
"""

import sys
import argparse
import logging
from boto3 import resource, client
from collections import defaultdict
from texttable import Texttable

FORMAT = "[%(asctime)s %(levelname)s] %(message)s"
logging.basicConfig(level=logging.WARNING, format=FORMAT)
logger = logging.getLogger()

# suppress boto3 internal logging below WARNING level
boto3_log = logging.getLogger("boto3")
boto3_log.setLevel(logging.WARNING)
boto3_log.propagate = True

# suppress botocore internal logging below WARNING level
botocore_log = logging.getLogger("botocore")
botocore_log.setLevel(logging.WARNING)
botocore_log.propagate = True


class TagCounter(object):

    def __init__(self):
        self.tags = defaultdict(int)
        self.res_count = 0

    def print_table(self):
        print("Count of resource tags for %d EC2 Instances, Images, Security "
              "Groups, Snapshots, Volumes, and ASGs\n" % self.res_count)
        t = Texttable()
        t.set_cols_dtype(['t', 'i'])
        rows = [['Tag Name', 'Count']]
        for k in sorted(self.tags.keys(), key=lambda s: s.lower()):
            rows.append([k, self.tags[k]])
        t.add_rows(rows)
        print(t.draw() + "\n")
        print(
            "Generated by: https://github.com/jantman/misc-scripts/blob/"
            "master/aws-count-tag-names.py\n")

    def run(self):
        logger.debug('Getting regions...')
        regions = self.get_region_names()
        logger.debug('Regions: %s', regions)
        for r in regions:
            self.tags_for_region(r)

    def get_region_names(self):
        conn = client('ec2')
        res = conn.describe_regions()
        regions = []
        for r in res['Regions']:
            regions.append(r['RegionName'])
        return regions

    def tags_for_region(self, region_name):
        logger.info('Getting tags for region: %s', region_name)
        logger.debug('Querying EC2 Instances...')
        res = resource('ec2', region_name=region_name)
        for i in res.instances.all():
            self.res_count += 1
            if i.tags is None:
                continue
            for t in i.tags:
                self.tags[t['Key']] += 1
        logger.debug('Querying EC2 Images (AMIs)...')
        for i in res.images.all():
            self.res_count += 1
            if i.tags is None:
                continue
            for t in i.tags:
                self.tags[t['Key']] += 1
        logger.debug('Querying EC2 Security Groups...')
        for i in res.security_groups.all():
            self.res_count += 1
            if i.tags is None:
                continue
            for t in i.tags:
                self.tags[t['Key']] += 1
        logger.debug('Querying EC2 Snapshots...')
        for i in res.snapshots.all():
            self.res_count += 1
            if i.tags is None:
                continue
            for t in i.tags:
                self.tags[t['Key']] += 1
        logger.debug('Querying EC2 Volumes...')
        for i in res.volumes.all():
            self.res_count += 1
            if i.tags is None:
                continue
            for t in i.tags:
                self.tags[t['Key']] += 1
        logger.debug('Querying ASGs...')
        cli = client('autoscaling', region_name=region_name)
        p = cli.get_paginator('describe_auto_scaling_groups')
        for resp in p.paginate():
            for asg in resp['AutoScalingGroups']:
                self.res_count += 1
                for t in asg['Tags']:
                    self.tags[t['Key']] += 1
        logger.info('Done with region.')


def parse_args(argv):
    """
    parse arguments/options

    this uses the new argparse module instead of optparse
    see: <https://docs.python.org/2/library/argparse.html>
    """
    p = argparse.ArgumentParser(description='Count distinct AWS tags')
    p.add_argument('-v', '--verbose', dest='verbose', action='count', default=0,
                   help='verbose output. specify twice for debug-level output.')
    args = p.parse_args(argv)

    return args


def set_log_info():
    """set logger level to INFO"""
    set_log_level_format(logging.INFO,
                         '%(asctime)s %(levelname)s:%(name)s:%(message)s')


def set_log_debug():
    """set logger level to DEBUG, and debug-level output format"""
    set_log_level_format(
        logging.DEBUG,
        "%(asctime)s [%(levelname)s %(filename)s:%(lineno)s - "
        "%(name)s.%(funcName)s() ] %(message)s"
    )


def set_log_level_format(level, format):
    """
    Set logger level and format.

    :param level: logging level; see the :py:mod:`logging` constants.
    :type level: int
    :param format: logging formatter format string
    :type format: str
    """
    formatter = logging.Formatter(fmt=format)
    logger.handlers[0].setFormatter(formatter)
    logger.setLevel(level)

if __name__ == "__main__":
    args = parse_args(sys.argv[1:])

    # set logging level
    if args.verbose > 1:
        set_log_debug()
    elif args.verbose == 1:
        set_log_info()

    script = TagCounter()
    script.run()
    script.print_table()