crumbled.py -- Sets a single target group ARN in an ALB listener default action.

.gitignore

.venv/
*.txt

.pre-commit-config.yaml

repos:
  - repo: local
    hooks:
      - id: black
        name: black
        language: system
        entry: black
        types: [python]
      - id: flake8
        name: flake8
        language: system
        entry: flake8
        types: [python]

.python-version

3.6.9

crumbled.py

#!/usr/bin/env python3
#
"""
Sets a single target group ARN in an ALB listener default action.

This disgusting hack was necessary to facilitate the migration of 
Web from EC2 to Fargate due to idiosynchrasies in Terraform.

See: https://jira.benchling.team/browse/BNCH-24083
"""
import argparse
import logging
import subprocess
import re
import json

import boto3


logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


def parse_args():
    """
    Define an argument parser and return the parsed arguments
    """
    parser = argparse.ArgumentParser(
        prog="crumbled", description="works magic with alb listener default actions"
    )
    parser.add_argument("--name", help="the name of the deploy to patch", required=True)
    parser.add_argument(
        "--dry-run", help="do a dry-run", default=False, action="store_true"
    )
    parser.add_argument(
        "--verbose",
        help="increase logging verbosity",
        default=False,
        action="store_true",
    )
    return parser.parse_args()


def parse_arn(stream, target):
    """
    Parse out an arn from running `terraform state show ...`.

    Since the output of `terraform state show` embeds ASCII escape
    characters, we have to strip them out.
    """
    regex = re.compile(
        r"^\s+arn\s+= \"(arn:aws:elasticloadbalancing:.+:.+:{}.+)\"".format(target)
    )
    ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")

    matches = []
    for line in stream.split("\n"):
        line = ansi_escape.sub("", line)
        match = re.findall(regex, line)
        if len(match) == 1:
            matches.append(match[0])

    if len(matches) > 1:
        raise RuntimeError(
            "Too many matches in input data: {}. Expected 1".format(len(matches))
        )

    return matches[0]


def main():
    args = parse_args()

    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)

    logger.warn(
        "If this script seems to hang, it may be because `aws-okta` is asking you for an auth token."
        " This script has no logic to handle that case interactively!"
    )

    logger.info("You selected {}".format(args.name))

    # Get the aws_lb_listener
    cmd = [
        "terraform",
        "state",
        "show",
        "module.{}.aws_lb_listener.web_https".format(args.name),
    ]

    logger.info("Running command -> `{}`".format(" ".join(cmd)))
    proc = subprocess.run(cmd, stdout=subprocess.PIPE, check=True)
    logger.debug("\n{}".format(proc.stdout))

    arn_lb_listener = parse_arn(proc.stdout.decode(), "listener")
    logger.info("lb listener arn is {}".format(arn_lb_listener))

    # Get the lb_target_group
    cmd = [
        "terraform",
        "state",
        "show",
        "module.{}.aws_lb_target_group.web".format(args.name),
    ]

    logger.info("Running command -> `{}`".format(" ".join(cmd)))
    proc = subprocess.run(cmd, stdout=subprocess.PIPE, check=True)
    logger.debug("\n{}".format(proc.stdout.decode()))

    arn_target_group = parse_arn(proc.stdout.decode(), "targetgroup")
    logger.info("lb target group arn is {}".format(arn_target_group))

    if args.dry_run:
        return

    # Modify the listener
    client = boto3.client("elbv2")

    response = client.modify_listener(
        ListenerArn=arn_lb_listener,
        DefaultActions=[{"Type": "forward", "TargetGroupArn": arn_target_group}],
    )
    print(json.dumps(response))


if __name__ == "__main__":
    main()

# EOF

requirements.txt

appdirs==1.4.4
attrs==20.3.0
black==19.3b0
blessings==1.7
boto3==1.17.26
botocore==1.20.26
bpython==0.21
certifi==2020.12.5
chardet==4.0.0
click==7.1.2
curtsies==0.3.5
cwcwidth==0.1.4
greenlet==1.0.0
idna==2.10
importlib-metadata==3.7.2
Jinja2==2.11.3
jmespath==0.10.0
jsonschema==3.2.0
MarkupSafe==1.1.1
openapi-spec-validator==0.2.9
Pygments==2.8.1
pyrsistent==0.17.3
python-dateutil==2.8.1
pyxdg==0.27
PyYAML==5.4.1
requests==2.25.1
s3transfer==0.3.4
six==1.15.0
toml==0.10.2
typing-extensions==3.7.4.3
urllib3==1.26.3
zipp==3.4.1

setup.cfg

[flake8]

ignore =
    Q000,
    C0103,
    C0111,
    C0301,
    C0321,
    # Whitespace before ':'
    E203,
    # Expected 1 blank line before a nested definition, found 0
    E306,
    # Module level import not at top of file
    E402,
    # Line too long (disabling since Black handles it for us)
    E501,
    # Do not assign a lambda expression, use a def
    E731,
    # No illegible variable names: I, O, and l (disabling since IDEs render them distinctly enough)
    E741,
    F403,
    R0903,
    R0904,
    R0911,
    R0912,
    R0913,
    R0914,
    R0915,
    # print(...) function & reserved word "print" - See discussion at https://git.io/fjVE2 for context.
    T003,
    T101,
    W0142,
    # Line break occurred before a binary operator (superseded by W504)
    W503,
    X100