"""CFNgin entrypoint."""
import logging
import os
import re
import sys
from yaml.constructor import ConstructorError
from runway._logging import PrefixAdaptor
from runway.util import MutableMap, SafeHaven, cached_property
from .actions import build, destroy, diff
from .config import render_parse_load as load_config
from .context import Context as CFNginContext
from .environment import parse_environment
from .providers.aws.default import ProviderBuilder
# explicitly name logger so its not redundant
LOGGER = logging.getLogger("runway.cfngin")
[docs]class CFNgin(object):
"""Control CFNgin.
Attributes:
EXCLUDE_REGEX (str): Regex used to exclude YAML files when searching
for config files.
EXCLUDE_LIST (str): Global list of YAML file names to exclude when
searching for config files.
concurrency (int): Max number of CFNgin stacks that can be deployed
concurrently. If the value is ``0``, will be constrained based on
the underlying graph.
interactive (bool): Wether or not to prompt the user before taking
action.
parameters (MutableMap): Combination of the parameters provided when
initalizing the class and any environment files that are found.
recreate_failed (bool): Destroy and re-create stacks that are stuck in
a failed state from an initial deployment when updating.
region (str): The AWS region where CFNgin is currently being executed.
sys_path (str): Working directory.
tail (bool): Wether or not to display all CloudFormation events in the
terminal.
"""
EXCLUDE_REGEX = r"runway(\..*)?\.(yml|yaml)"
EXCLUDE_LIST = ["bitbucket-pipelines.yml", "buildspec.yml", "docker-compose.yml"]
def __init__(self, ctx, parameters=None, sys_path=None):
"""Instantiate class.
Args:
ctx (runway.context.Context): Runway context object.
parameters (Optional[Dict[str. Any]]): Parameters from Runway.
sys_path (Optional[str]): Working directory.
"""
self.__ctx = ctx
self._env_file_name = None
self.concurrency = ctx.env.max_concurrent_cfngin_stacks
self.interactive = ctx.is_interactive
self.parameters = MutableMap()
self.recreate_failed = ctx.is_noninteractive
self.region = ctx.env_region
self.sys_path = sys_path or os.getcwd()
self.tail = bool(ctx.env.debug or ctx.env.verbose)
self.parameters.update(self.env_file)
if parameters:
LOGGER.debug("adding Runway parameters to CFNgin parameters")
self.parameters.update(parameters)
self._inject_common_parameters()
@cached_property
def env_file(self):
"""Contents of a CFNgin environment file.
Returns:
MutableMap
"""
result = {}
supported_names = [
"{}.env".format(self.__ctx.env_name),
"{}-{}.env".format(self.__ctx.env_name, self.region),
]
for _, file_name in enumerate(supported_names):
file_path = os.path.join(self.sys_path, file_name)
if os.path.isfile(file_path):
LOGGER.info("found environment file: %s", file_path)
self._env_file_name = file_path
with open(file_path, "r") as file_:
result.update(parse_environment(file_.read()))
return MutableMap(**result)
[docs] def deploy(self, force=False, sys_path=None):
"""Run the CFNgin deploy action.
Args:
force (bool): Explicitly enable the action even if an environment
file is not found.
sys_path (Optional[str]): Explicitly define a path to work in.
If not provided, ``self.sys_path`` is used.
"""
if self.should_skip(force):
return
if not sys_path:
sys_path = self.sys_path
config_file_names = self.find_config_files(sys_path=sys_path)
with SafeHaven(
environ=self.__ctx.env_vars, sys_modules_exclude=["awacs", "troposphere"]
):
for config_name in config_file_names:
logger = PrefixAdaptor(os.path.basename(config_name), LOGGER)
logger.notice("deploy (in progress)")
with SafeHaven(
argv=["stacker", "build", config_name],
sys_modules_exclude=["awacs", "troposphere"],
):
ctx = self.load(config_name)
action = build.Action(
context=ctx,
provider_builder=self._get_provider_builder(
ctx.config.service_role
),
)
action.execute(concurrency=self.concurrency, tail=self.tail)
logger.success("deploy (complete)")
[docs] def destroy(self, force=False, sys_path=None):
"""Run the CFNgin destroy action.
Args:
force (bool): Explicitly enable the action even if an environment
file is not found.
sys_path (Optional[str]): Explicitly define a path to work in.
If not provided, ``self.sys_path`` is used.
"""
if self.should_skip(force):
return
if not sys_path:
sys_path = self.sys_path
config_file_names = self.find_config_files(sys_path=sys_path)
# destroy should run in reverse to handle dependencies
config_file_names.reverse()
with SafeHaven(environ=self.__ctx.env_vars):
for config_name in config_file_names:
logger = PrefixAdaptor(os.path.basename(config_name), LOGGER)
logger.notice("destroy (in progress)")
with SafeHaven(argv=["stacker", "destroy", config_name]):
ctx = self.load(config_name)
action = destroy.Action(
context=ctx,
provider_builder=self._get_provider_builder(
ctx.config.service_role
),
)
action.execute(
concurrency=self.concurrency, force=True, tail=self.tail
)
logger.success("destroy (complete)")
[docs] def load(self, config_path):
"""Load a CFNgin config into a context object.
Args:
config_path (str): Valid path to a CFNgin config file.
Returns:
:class:`runway.cfngin.context.Context`
"""
LOGGER.debug("loading CFNgin config: %s", os.path.basename(config_path))
try:
config = self._get_config(config_path)
return self._get_context(config, config_path)
except ConstructorError as err:
if err.problem.startswith(
"could not determine a constructor " "for the tag '!"
):
LOGGER.error(
'"%s" is located in the module\'s root directory '
"and appears to be a CloudFormation template; "
"please move CloudFormation templates to a subdirectory",
config_path,
)
sys.exit(1)
raise
[docs] def plan(self, force=False, sys_path=None):
"""Run the CFNgin plan action.
Args:
force (bool): Explicitly enable the action even if an environment
file is not found.
sys_path (Optional[str]): Explicitly define a path to work in.
If not provided, ``self.sys_path`` is used.
"""
if self.should_skip(force):
return
if not sys_path:
sys_path = self.sys_path
config_file_names = self.find_config_files(sys_path=sys_path)
with SafeHaven(environ=self.__ctx.env_vars):
for config_name in config_file_names:
logger = PrefixAdaptor(os.path.basename(config_name), LOGGER)
logger.notice("plan (in progress)")
with SafeHaven(argv=["stacker", "diff", config_name]):
ctx = self.load(config_name)
action = diff.Action(
context=ctx,
provider_builder=self._get_provider_builder(
ctx.config.service_role
),
)
action.execute()
logger.success("plan (complete)")
[docs] def should_skip(self, force=False):
"""Determine if action should be taken or not.
Args:
force (bool): If ``True``, will always return ``False`` meaning
the action should not be skipped.
Returns:
bool: Skip action or not.
"""
if force or self.env_file:
return False
LOGGER.info("skipped; no parameters and environment file not found")
return True
def _get_config(self, file_path, validate=True):
"""Initialize a CFNgin config object from a file.
Args:
file_path (str): Path to the config file to load.
validate (bool): Validate the loaded config.
Returns:
:class:`runway.cfngin.config.Config`
"""
with open(file_path, "r") as file_:
raw_config = file_.read()
return load_config(raw_config, self.parameters, validate)
def _get_context(self, config, config_path):
"""Initialize a CFNgin context object.
Args:
config (:class:`runway.cfngin.config.Config): CFNgin config object.
config_path (str): Path to the config file that was provided.
Returns:
:class:`runway.cfngin.context.Context`
"""
return CFNginContext(
boto3_credentials=self.__ctx.boto3_credentials,
config=config,
config_path=config_path,
environment=self.parameters,
force_stacks=[], # placeholder
region=self.region,
stack_names=[], # placeholder
)
def _get_provider_builder(self, service_role=None):
"""Initialize provider builder.
Args:
service_role (Optional[str]): CloudFormation service role.
Returns:
ProviderBuilder
"""
if self.interactive:
LOGGER.verbose("using interactive AWS provider mode")
else:
LOGGER.verbose("using default AWS provider mode")
return ProviderBuilder(
interactive=self.interactive,
recreate_failed=self.recreate_failed,
region=self.region,
service_role=service_role,
)
def _inject_common_parameters(self):
"""Add common parameters if they don't already exist.
Adding these commonly used parameters will remove the need to add
lookup support (mainly for environment variable lookups) in places
such as ``cfngin_bucket``.
Injected Parameters
~~~~~~~~~~~~~~~~~~~
**environment (str)**
Taken from the ``DEPLOY_ENVIRONMENT`` environment variable. This
will the be current Runway environment being processed.
**region (str)**
Taken from the ``AWS_REGION`` environment variable. This will be
the current region being deployed to.
"""
if not self.parameters.get("environment"):
self.parameters["environment"] = self.__ctx.env_name
if not self.parameters.get("region"):
self.parameters["region"] = self.region
[docs] @classmethod
def find_config_files(cls, exclude=None, sys_path=None):
"""Find CFNgin config files.
Args:
exclude (Optional[List[str]]): List of file names to exclude. This
list is appended to the global exclude list.
sys_path (Optional[str]): Explicitly define a path to search for
config files.
Returns:
List[str]: Path to config files that were found.
"""
if not sys_path:
sys_path = os.getcwd()
elif os.path.isfile(sys_path):
return [sys_path]
exclude = exclude or []
result = []
exclude.extend(cls.EXCLUDE_LIST)
for root, _dirs, files in os.walk(sys_path):
for name in files:
if re.match(cls.EXCLUDE_REGEX, name) or (
name in exclude or name.startswith(".")
):
# Hidden files (e.g. .gitlab-ci.yml), Runway configs,
# and docker-compose files definitely aren't stacker
# config files
continue
if os.path.splitext(name)[-1] in [".yaml", ".yml"]:
result.append(os.path.join(root, name))
break # only need top level files
result.sort()
return result