Removes a broken terraform lock state

terralock.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
# To fix this issue in terraform:
#    Failed to unlock state: failed to retrieve lock info: unexpected end of JSON input
# Python does what Terraform cannot do!

"""Delete terraform lock
Usage:
    terralock.py --list --table TABLE_NAME
    terralock.py --delete --table TABLE_NAME [-l LOCK_ID, --confirm YES]

Arguments:
    YES                               Enter 'yes' to confirm
    LOCK_ID                           Lock ID to find
    TABLE_NAME                        Table name for your state file

Options:
    -h                                show this message
    -l, --lockid LOCK_ID              lock id to delete
    -c YES, --confirm yes             lock id to confirm delete on
    -t TABLE_NAME, --table TABLE_NAME table name of your state file
"""

from __future__ import print_function # Python 2/3 compatible
from boto3.dynamodb.conditions import Key, Attr
from docopt import docopt
from os import environ
import boto3
import decimal
import json
import json

# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, decimal.Decimal):
            return str(o)
        return super(DecimalEncoder, self).default(o)

def print_arguements(args):
    """
    prints a pretty format version of arguements
    """
    for key, val in args.items():
        print('     {:<10s}: {:<10s}'.format(key, str(val)))

def get_arg_option(args):
    for key, value in args.items():
        if (key != '--force' and key.startswith('--') and
                isinstance(value, bool) and value):
            return key.replace('-', '')

def scan_table_allpages(dynamodb_resource, table_name, filter_key=None, filter_value=None):
    """
    Perform a scan operation on table. Can specify filter_key (col name) and its value to be filtered. This gets all pages of results.
    Returns list of items.
    """
    table = dynamodb_resource.Table(table_name)

    if filter_key and filter_value:
        filtering_exp = Key(filter_key).eq(filter_value)
        response = table.scan(FilterExpression=filtering_exp)
    else:
        response = table.scan()

    items = response['Items']
    while True:
        if response.get('LastEvaluatedKey'):
            response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
            items += response['Items']
        else:
            break

    return items

def get_table_metadata(table_name):
    """
    Get some metadata about chosen table.
    """
    table = dynamodb_resource.Table(table_name)

    return {
        'num_items': table.item_count,
        'primary_key_name': table.key_schema[0],
        'status': table.table_status,
        'bytes_size': table.table_size_bytes,
        'global_secondary_indices': table.global_secondary_indexes
    }

def delete_item(dynamodb_resource, table_name, pk_name, pk_value):
    """
    Delete an item (row) in table from its primary key.
    """
    table = dynamodb_resource.Table(table_name)
    response = table.delete_item(Key={pk_name: pk_value})

    return

class TerralockCommand(object):
    """Terralock object command
    """

    def __init__(self, args):
        self.args = args
        default_region = 'us-east-1'
        if not environ.get('AWS_DEFAULT_REGION'):
            print("Environment AWS_DEFAULT_REGION not found, using default ({})".format(default_region))
        self.region_name = environ.get('AWS_DEFAULT_REGION', 'us-east-1')
        default_endpoint = 'https://dynamodb-fips.us-east-1.amazonaws.com'
        if not environ.get('DYNAMO_ENDPOINT'):
            print("Environment DYNAMO_ENDPOINT not found, using default ({})".format(default_endpoint))
        self.endpoint_url = environ.get('DYNAMO_ENDPOINT', 'https://dynamodb-fips.us-east-1.amazonaws.com')

    def list(self):
        """
        list all of the objects in this table
        """
        table_name  = self.args['--table']
        dynamodb_resource = boto3.resource('dynamodb', region_name=self.region_name, endpoint_url=self.endpoint_url)
        #table = dynamodb_resource.Table(table_name)
        results = scan_table_allpages(dynamodb_resource, table_name)
        print(json.dumps(results, indent=5))
        print("Found {len} objects in `{table}'".format(len=len(results), table=table_name))

    def delete(self):
        """
        delete a terraform lock in the dynamodb table
        """
        table_name  = self.args['--table']
        lock_id  = self.args['--lockid']
        confirm  = self.args['--confirm']

        perishable_results = list()
        dynamodb_resource = boto3.resource('dynamodb', region_name=self.region_name, endpoint_url=self.endpoint_url)

        if not confirm:
            all_items = scan_table_allpages(dynamodb_resource, table_name)
            found_perishable_item = False
            for key in all_items:
                keyinfo = key.get('Info')
                if keyinfo:
                    if lock_id in keyinfo:
                        found_perishable_item = True
                        table_lock_id = key.get('LockID')
                        perishable_results.append(key)
            if found_perishable_item:
                print("Found {} items we can delete. Please review and add --confirm to your command")
                print(json.dumps(perishable_results, indent=5))
            else:
                print("Nothing to delete matching your terraform lock ID on table `{}'".format(table_name))
        elif confirm == "yes":
            all_items = scan_table_allpages(dynamodb_resource, table_name)
            table_lock_id = None
            for key in all_items:
                keyinfo = key.get('Info')
                if keyinfo:
                    if lock_id in keyinfo:
                        table_lock_id = key.get('LockID')
            if not table_lock_id:
                print("Couldn't find that lock ID!")
                exit(1)
            delete_item(dynamodb_resource, table_name, 'LockID', table_lock_id)
        else:
            print("--confirm must equal `yes'")
            exit(1)

def main():
    """Parse the CLI"""
    arguments = docopt(__doc__)

    cmd = TerralockCommand(arguments)
    method = get_arg_option(arguments)
    getattr(cmd, method)()


# This is the standard boilerplate that calls the main() function.
if __name__ == '__main__':
    main()