aws_lambda_builders/workflow.py

"""
Implementation of a base workflow
"""

import functools
import logging
import os
from collections import namedtuple
from enum import Enum
from typing import Optional

from aws_lambda_builders.actions import ActionFailedError
from aws_lambda_builders.architecture import X86_64
from aws_lambda_builders.binary_path import BinaryPath
from aws_lambda_builders.exceptions import (
    MisMatchRuntimeError,
    RuntimeValidatorError,
    WorkflowFailedError,
    WorkflowUnknownError,
)
from aws_lambda_builders.path_resolver import PathResolver
from aws_lambda_builders.registry import DEFAULT_REGISTRY
from aws_lambda_builders.validator import RuntimeValidator

LOG = logging.getLogger(__name__)


# Named tuple to express the capabilities supported by the builder.
# ``Language`` is the programming language. Ex: Python
# ``LangageFramework`` is the framework of particular language. Ex: PIP
# ``ApplicationFramework`` is the specific application framework used to write the code. Ex: Chalice
Capability = namedtuple("Capability", ["language", "dependency_manager", "application_framework"])


class BuildMode(object):
    DEBUG = "debug"
    RELEASE = "release"


class BuildDirectory(Enum):
    SCRATCH = "scratch"
    ARTIFACTS = "artifacts"
    SOURCE = "source"


class BuildInSourceSupport(Enum):
    """
    Enum to define a workflow's support for building in source.
    """

    # can't build in source directory (e.g. only able to build in temporary or artifacts directories)
    NOT_SUPPORTED = [False]

    # can build in source directory but not required to
    OPTIONALLY_SUPPORTED = [False, True]

    # only able to build in source directory and not somewhere else
    EXCLUSIVELY_SUPPORTED = [True]


# TODO: Move sanitize out to its own class.
def sanitize(func):  # pylint: disable=too-many-statements
    """
    sanitize the executable path of the runtime specified by validating it.
    :param func: Workflow's run method is sanitized
    """

    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):  # pylint: disable=too-many-statements
        valid_paths = {}
        invalid_paths = {}
        validation_errors = []
        # NOTE: we need to access binaries to get paths and resolvers, before validating.
        for binary, binary_checker in self.binaries.items():
            invalid_paths[binary] = []
            try:
                exec_paths = (
                    binary_checker.resolver.exec_paths
                    if not binary_checker.path_provided
                    else binary_checker.binary_path
                )
            except ValueError as ex:
                raise WorkflowFailedError(workflow_name=self.NAME, action_name="Resolver", reason=str(ex))
            for executable_path in exec_paths:
                try:
                    valid_path = binary_checker.validator.validate(executable_path)
                    if valid_path:
                        valid_paths[binary] = valid_path
                except MisMatchRuntimeError as ex:
                    LOG.debug("Invalid executable for %s at %s", binary, executable_path, exc_info=str(ex))
                    invalid_paths[binary].append(executable_path)

                except RuntimeValidatorError as ex:
                    LOG.debug("Runtime validation error for %s", binary, exc_info=str(ex))
                    if str(ex) not in validation_errors:
                        validation_errors.append(str(ex))

                if valid_paths.get(binary, None):
                    binary_checker.binary_path = valid_paths[binary]
                    break
        if validation_errors:
            raise WorkflowFailedError(
                workflow_name=self.NAME, action_name="Validation", reason="\n".join(validation_errors)
            )

        if len(self.binaries) != len(valid_paths):
            validation_failed_binaries = set(self.binaries.keys()).difference(valid_paths.keys())
            for validation_failed_binary in validation_failed_binaries:
                message = "Binary validation failed for {0}, searched for {0} in following locations  : {1} which did not satisfy constraints for runtime: {2}. Do you have {0} for runtime: {2} on your PATH?".format(
                    validation_failed_binary, invalid_paths[validation_failed_binary], self.runtime
                )
                validation_errors.append(message)
            raise WorkflowFailedError(
                workflow_name=self.NAME, action_name="Validation", reason="\n".join(validation_errors)
            )
        func(self, *args, **kwargs)

    return wrapper


class _WorkflowMetaClass(type):
    """
    A metaclass that maintains the registry of loaded builders
    """

    def __new__(mcs, name, bases, class_dict):
        """
        Add the builder to registry when loading the class
        """

        cls = type.__new__(mcs, name, bases, class_dict)

        # We don't want to register the base classes, so we simply return here.
        # Also, skip further steps if the class is marked for testing
        if cls.__name__ == "BaseWorkflow" or cls.__TESTING__:
            return cls

        # Validate class variables

        # All classes must provide a name
        if not isinstance(cls.NAME, str):
            raise ValueError("Workflow must provide a valid name")

        # All workflows must express their capabilities
        if not isinstance(cls.CAPABILITY, Capability):
            raise ValueError("Workflow '{}' must register valid capabilities".format(cls.NAME))

        # All workflows must define supported values for build in source
        if not isinstance(cls.BUILD_IN_SOURCE_SUPPORT, BuildInSourceSupport):
            raise ValueError("Workflow '{}' must define supported values for build in source".format(cls.NAME))

        # All workflows must define default build directory
        if not isinstance(cls.DEFAULT_BUILD_DIR, BuildDirectory):
            raise ValueError("Workflow '{}' must define default build directory".format(cls.NAME))

        LOG.debug("Registering workflow '%s' with capability '%s'", cls.NAME, cls.CAPABILITY)
        DEFAULT_REGISTRY[cls.CAPABILITY] = cls

        return cls


class BaseWorkflow(object, metaclass=_WorkflowMetaClass):
    """
    Default implementation of the builder workflow. It provides several useful capabilities out-of-box that help
    minimize the scope of build actions.
    """

    # Set this property if you are in the process of testing a workflow class. This will prevent the class from
    # being added to registry.
    __TESTING__ = False

    NAME = None

    # Capabilities supported by this builder. Must be an instance of `Capability` named tuple
    CAPABILITY = None

    # Optional list of manifests file/folder names supported by this workflow.
    SUPPORTED_MANIFESTS = []

    # Support for building in source, each workflow should define this.
    BUILD_IN_SOURCE_SUPPORT = None

    # The directory where the workflow builds/installs by default, each workflow should define this.
    DEFAULT_BUILD_DIR = None

    def __init__(
        self,
        source_dir,
        artifacts_dir,
        scratch_dir,
        manifest_path,
        runtime=None,
        executable_search_paths=None,
        optimizations=None,
        options=None,
        mode=BuildMode.RELEASE,
        download_dependencies=True,
        dependencies_dir=None,
        combine_dependencies=True,
        architecture=X86_64,
        is_building_layer=False,
        experimental_flags=None,
        build_in_source=None,
        unpatched_runtime=None,
    ):
        # pylint: disable-msg=too-many-locals
        """
        Initialize the builder with given arguments. These arguments together form the "public API" that each
        build action must support at the minimum.

        Parameters
        ----------
        source_dir : str
            Path to a folder containing the source code
        artifacts_dir : str
            Path to a folder where the built artifacts should be placed
        scratch_dir : str
            Path to a directory that the workflow can use as scratch space. Workflows are expected to use this directory
            to write temporary files instead of ``/tmp`` or other OS-specific temp directories.
        manifest_path : str
            Path to the dependency manifest
        runtime : str, optional
            Optional, name of the AWS Lambda runtime that you are building for. This is sent to the builder for
            informational purposes, by default None
        executable_search_paths : list, optional
            Additional list of paths to search for executables required by the workflow, by default None
        optimizations : dict, optional
            dictionary of optimization flags to pass to the build action. **Not supported**, by default None
        options : dict, optional
            dictionary of options ot pass to build action. **Not supported**., by default None
        mode : str, optional
            Mode the build should produce, by default BuildMode.RELEASE
        download_dependencies: bool, optional
            Should download dependencies when building
        dependencies_dir : str, optional
            Path to folder the dependencies should be downloaded to
        combine_dependencies: bool, optional
            This flag will only be used if dependency_folder is specified. False will not copy dependencies
            from dependency_folder into build folder
        architecture : str, optional
            Architecture type either arm64 or x86_64 for which the build will be based on in AWS lambda, by default X86_64

        is_building_layer: bool, optional
            Boolean flag which will be set True if current build operation is being executed for layers

        experimental_flags: list, optional
            List of strings, which will indicate enabled experimental flags for the current build session

        build_in_source: Optional[bool]
            Optional, will execute the build operation in the source directory if True.
        """

        self.source_dir = source_dir
        self.artifacts_dir = artifacts_dir
        self.scratch_dir = scratch_dir
        self.manifest_path = manifest_path
        self.runtime = runtime
        self.optimizations = optimizations
        self.options = options
        self.executable_search_paths = executable_search_paths
        self.mode = mode
        self.download_dependencies = download_dependencies
        self.dependencies_dir = dependencies_dir
        self.combine_dependencies = combine_dependencies
        self.architecture = architecture
        self.is_building_layer = is_building_layer
        self.unpatched_runtime = unpatched_runtime
        self.experimental_flags = experimental_flags if experimental_flags else []

        # this represents where the build/install happens, not the final output directory (that's the artifacts_dir)
        self.build_dir = self._select_build_dir(build_in_source)

        # Actions are registered by the subclasses as they seem fit
        self.actions = []
        self._binaries = {}

    def _select_build_dir(self, build_in_source: Optional[bool]) -> str:
        """
        Returns the build directory for the workflow.
        """

        should_build_in_source = build_in_source
        if build_in_source not in self.BUILD_IN_SOURCE_SUPPORT.value:
            # assign default value
            should_build_in_source = self.DEFAULT_BUILD_DIR == BuildDirectory.SOURCE

            # only show warning if an unsupported value was explicitly passed in
            if build_in_source is not None:
                LOG.warning(
                    'Workflow %s does not support value "%s" for building in source. Using default value "%s".',
                    self.NAME,
                    build_in_source,
                    should_build_in_source,
                )

        build_directory_mapping = {
            BuildDirectory.SCRATCH: self.scratch_dir,
            BuildDirectory.ARTIFACTS: self.artifacts_dir,
            BuildDirectory.SOURCE: self.source_dir,
        }

        return self.source_dir if should_build_in_source else build_directory_mapping.get(self.DEFAULT_BUILD_DIR)

    def is_supported(self):
        """
        Is the given manifest supported? If the workflow exposes no manifests names, then we it is assumed that
        we don't have a restriction
        """

        if self.SUPPORTED_MANIFESTS:
            return os.path.basename(self.manifest_path) in self.SUPPORTED_MANIFESTS

        return True

    def get_resolvers(self):
        """
        Non specialized path resolver that just returns the list of executable for the runtime on the path.
        """
        return [
            PathResolver(
                runtime=self.runtime,
                binary=self.CAPABILITY.language,
                executable_search_paths=self.executable_search_paths,
            )
        ]

    def get_validators(self):
        """
        No-op validator that does not validate the runtime_path.
        """
        return [RuntimeValidator(runtime=self.runtime, architecture=self.architecture)]

    @property
    def binaries(self):
        if not self._binaries:
            resolvers = self.get_resolvers()
            validators = self.get_validators()
            self._binaries = {
                resolver.binary: BinaryPath(resolver=resolver, validator=validator, binary=resolver.binary)
                for resolver, validator in zip(resolvers, validators)
            }
        return self._binaries

    @binaries.setter
    def binaries(self, binaries):
        self._binaries = binaries

    @sanitize
    def run(self):
        """
        Actually perform the build by executing registered actions.

        :raises WorkflowFailedError: If the workflow does not contain any actions or if one of the actions ran into
            an error

        :raises WorkflowUnknownError: If one of the actions in the workflow raised an unhandled exception
        """

        LOG.debug("Running workflow '%s'", self.NAME)

        if not self.actions:
            raise WorkflowFailedError(
                workflow_name=self.NAME, action_name=None, reason="Workflow does not have any actions registered"
            )

        for action in self.actions:
            action_info = "{}:{}".format(self.NAME, action.NAME)
            function_name = ""
            if self.options and "build_logical_id" in self.options:
                function_name = "{}:".format(self.options["build_logical_id"])
            LOG.info("%s Running %s", function_name, action_info)

            try:
                action.execute()

                LOG.debug("%s succeeded", action_info)

            except ActionFailedError as ex:
                LOG.debug("%s failed", action_info, exc_info=ex)

                raise WorkflowFailedError(workflow_name=self.NAME, action_name=action.NAME, reason=str(ex))
            except Exception as ex:
                LOG.debug("%s raised unhandled exception", action_info, exc_info=ex)

                raise WorkflowUnknownError(workflow_name=self.NAME, action_name=action.NAME, reason=str(ex))

    def __repr__(self):
        """
        Pretty prints information about this workflow.

        Sample output:
            Workflow=MyWorkflow
            Actions=
                Name=Action1, Purpose=COPY_SOURCE, Description=Copies source code
                Name=Action2, Purpose=RESOLVE_DEPENDENCIES, Description=Resolves dependencies
                Name=Action3, Purpose=COMPILE_SOURCE, Description=Compiles code
        """
        return "Workflow={}\nActions=\n\t{}".format(self.NAME, "\n\t".join(map(str, self.actions)))