""" Context object used by build command """ import logging import os import pathlib import shutil from typing import Dict, List, Optional, Tuple import click from samcli.commands._utils.constants import DEFAULT_BUILD_DIR from samcli.commands._utils.experimental import ExperimentalFlag, prompt_experimental from samcli.commands._utils.template import ( get_template_data, move_template, ) from samcli.commands.build.exceptions import InvalidBuildDirException, MissingBuildMethodException from samcli.commands.build.utils import MountMode, prompt_user_to_enable_mount_with_write_if_needed from samcli.commands.exceptions import UserException from samcli.lib.bootstrap.nested_stack.nested_stack_manager import NestedStackManager from samcli.lib.build.app_builder import ( ApplicationBuilder, ApplicationBuildResult, BuildError, UnsupportedBuilderLibraryVersionError, ) from samcli.lib.build.build_graph import DEFAULT_DEPENDENCIES_DIR from samcli.lib.build.bundler import EsbuildBundlerManager from samcli.lib.build.exceptions import ( BuildInsideContainerError, InvalidBuildGraphException, ) from samcli.lib.build.workflow_config import UnsupportedRuntimeException from samcli.lib.intrinsic_resolver.intrinsics_symbol_table import IntrinsicsSymbolTable from samcli.lib.providers.provider import LayerVersion, ResourcesToBuildCollector, Stack from samcli.lib.providers.sam_api_provider import SamApiProvider from samcli.lib.providers.sam_function_provider import SamFunctionProvider from samcli.lib.providers.sam_layer_provider import SamLayerProvider from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider from samcli.lib.telemetry.event import EventName, EventTracker, UsedFeature from samcli.lib.utils.osutils import BUILD_DIR_PERMISSIONS from samcli.local.docker.manager import ContainerManager from samcli.local.lambdafn.exceptions import ( FunctionNotFound, ResourceNotFound, ) LOG = logging.getLogger(__name__) class BuildContext: def __init__( self, resource_identifier: Optional[str], template_file: str, base_dir: Optional[str], build_dir: str, cache_dir: str, cached: bool, parallel: bool, mode: Optional[str], manifest_path: Optional[str] = None, clean: bool = False, use_container: bool = False, # pylint: disable=fixme # FIXME: parameter_overrides is never None, we should change this to "dict" from Optional[dict] # See samcli/commands/_utils/options.py:251 for its all possible values parameter_overrides: Optional[dict] = None, docker_network: Optional[str] = None, skip_pull_image: bool = False, container_env_var: Optional[dict] = None, container_env_var_file: Optional[str] = None, build_images: Optional[dict] = None, excluded_resources: Optional[Tuple[str, ...]] = None, aws_region: Optional[str] = None, create_auto_dependency_layer: bool = False, stack_name: Optional[str] = None, print_success_message: bool = True, locate_layer_nested: bool = False, hook_name: Optional[str] = None, build_in_source: Optional[bool] = None, mount_with: str = MountMode.READ.value, mount_symlinks: Optional[bool] = False, ) -> None: """ Initialize the class Parameters ---------- resource_identifier: Optional[str] The unique identifier of the resource template_file: str Path to the template for building base_dir : str Path to a folder. Use this folder as the root to resolve relative source code paths against build_dir : str Path to the directory where we will be storing built artifacts cache_dir : str Path to a the directory where we will be caching built artifacts cached: Optional. Set to True to build each function with cache to improve performance parallel : bool Optional. Set to True to build each function in parallel to improve performance mode : str Optional, name of the build mode to use ex: 'debug' manifest_path : Optional[str] Optional path to manifest file to replace the default one clean: bool Clear the build directory before building use_container: bool Build inside container parameter_overrides: Optional[dict] Optional dictionary of values for SAM template parameters that might want to get substituted within the template docker_network: Optional[str] Docker network to run the container in. skip_pull_image: bool Whether we should pull new Docker container image or not container_env_var: Optional[dict] An optional dictionary of environment variables to pass to the container container_env_var_file: Optional[dict] An optional path to file that contains environment variables to pass to the container build_images: Optional[dict] An optional dictionary of build images to be used for building functions aws_region: Optional[str] Aws region code create_auto_dependency_layer: bool Create auto dependency layer for accelerate feature stack_name: Optional[str] Original stack name, which is used to generate layer name for accelerate feature print_success_message: bool Print successful message locate_layer_nested: bool Locate layer to its actual, worked with nested stack hook_name: Optional[str] Name of the hook package build_in_source: Optional[bool] Set to True to build in the source directory. mount_with: Mount mode of source code directory when building inside container, READ ONLY by default mount_symlinks Optional[bool]: Indicates if symlinks should be mounted inside the container """ self._resource_identifier = resource_identifier self._template_file = template_file self._base_dir = base_dir # Note(xinhol): use_raw_codeuri is temporary to fix a bug, and will be removed for a permanent solution. self._use_raw_codeuri = bool(self._base_dir) self._build_dir = build_dir self._cache_dir = cache_dir self._parallel = parallel self._manifest_path = manifest_path self._clean = clean self._use_container = use_container self._parameter_overrides = parameter_overrides # Override certain CloudFormation pseudo-parameters based on values provided by customer self._global_parameter_overrides: Optional[Dict] = None if aws_region: self._global_parameter_overrides = {IntrinsicsSymbolTable.AWS_REGION: aws_region} self._docker_network = docker_network self._skip_pull_image = skip_pull_image self._mode = mode self._cached = cached self._container_env_var = container_env_var self._container_env_var_file = container_env_var_file self._build_images = build_images self._exclude = excluded_resources self._create_auto_dependency_layer = create_auto_dependency_layer self._stack_name = stack_name self._print_success_message = print_success_message self._function_provider: Optional[SamFunctionProvider] = None self._layer_provider: Optional[SamLayerProvider] = None self._container_manager: Optional[ContainerManager] = None self._stacks: List[Stack] = [] self._locate_layer_nested = locate_layer_nested self._hook_name = hook_name self._build_in_source = build_in_source self._build_result: Optional[ApplicationBuildResult] = None self._mount_with = MountMode(mount_with) self._mount_symlinks = mount_symlinks def __enter__(self) -> "BuildContext": self.set_up() return self def set_up(self) -> None: """Set up class members used for building This should be called each time before run() if stacks are changed.""" self._stacks, remote_stack_full_paths = SamLocalStackProvider.get_stacks( self._template_file, parameter_overrides=self._parameter_overrides, global_parameter_overrides=self._global_parameter_overrides, ) if remote_stack_full_paths: LOG.warning( "Below nested stacks(s) specify non-local URL(s), which are unsupported:\n%s\n" "Skipping building resources inside these nested stacks.", "\n".join([f"- {full_path}" for full_path in remote_stack_full_paths]), ) # Note(xinhol): self._use_raw_codeuri is added temporarily to fix issue #2717 # when base_dir is provided, codeuri should not be resolved based on template file path. # we will refactor to make all path resolution inside providers intead of in multiple places self._function_provider = SamFunctionProvider( self.stacks, self._use_raw_codeuri, locate_layer_nested=self._locate_layer_nested ) self._layer_provider = SamLayerProvider(self.stacks, self._use_raw_codeuri) if not self._base_dir: # Base directory, if not provided, is the directory containing the template self._base_dir = str(pathlib.Path(self._template_file).resolve().parent) self._build_dir = self._setup_build_dir(self._build_dir, self._clean) if self._cached: cache_path = pathlib.Path(self._cache_dir) cache_path.mkdir(mode=BUILD_DIR_PERMISSIONS, parents=True, exist_ok=True) self._cache_dir = str(cache_path.resolve()) dependencies_path = pathlib.Path(DEFAULT_DEPENDENCIES_DIR) dependencies_path.mkdir(mode=BUILD_DIR_PERMISSIONS, parents=True, exist_ok=True) if self._use_container: self._container_manager = ContainerManager( docker_network_id=self._docker_network, skip_pull_image=self._skip_pull_image ) def __exit__(self, *args): pass def get_resources_to_build(self): return self.resources_to_build def run(self) -> None: """Runs the building process by creating an ApplicationBuilder.""" if self._is_sam_template(): SamApiProvider.check_implicit_api_resource_ids(self.stacks) self._stacks = self._handle_build_pre_processing() caught_exception: Optional[Exception] = None try: # boolean value indicates if mount with write or not, defaults to READ ONLY mount_with_write = False if self._use_container: if self._mount_with == MountMode.WRITE: mount_with_write = True else: # if self._mount_with is NOT WRITE # check the need of mounting with write permissions and prompt user to enable it if needed mount_with_write = prompt_user_to_enable_mount_with_write_if_needed( self.get_resources_to_build(), self.base_dir, ) builder = ApplicationBuilder( self.get_resources_to_build(), self.build_dir, self.base_dir, self.cache_dir, self.cached, self.is_building_specific_resource, manifest_path_override=self.manifest_path_override, container_manager=self.container_manager, mode=self.mode, parallel=self._parallel, container_env_var=self._container_env_var, container_env_var_file=self._container_env_var_file, build_images=self._build_images, combine_dependencies=not self._create_auto_dependency_layer, build_in_source=self._build_in_source, mount_with_write=mount_with_write, mount_symlinks=self._mount_symlinks, ) self._check_exclude_warning() self._check_rust_cargo_experimental_flag() for f in self.get_resources_to_build().functions: EventTracker.track_event(EventName.BUILD_FUNCTION_RUNTIME.value, f.runtime) self._build_result = builder.build() self._handle_build_post_processing(builder, self._build_result) click.secho("\nBuild Succeeded", fg="green") # try to use relpath so the command is easier to understand, however, # under Windows, when SAM and (build_dir or output_template_path) are # on different drive, relpath() fails. root_stack = SamLocalStackProvider.find_root_stack(self.stacks) out_template_path = root_stack.get_output_template_path(self.build_dir) try: build_dir_in_success_message = os.path.relpath(self.build_dir) output_template_path_in_success_message = os.path.relpath(out_template_path) except ValueError: LOG.debug("Failed to retrieve relpath - using the specified path as-is instead") build_dir_in_success_message = self.build_dir output_template_path_in_success_message = out_template_path if self._print_success_message: msg = self._gen_success_msg( build_dir_in_success_message, output_template_path_in_success_message, os.path.abspath(self.build_dir) == os.path.abspath(DEFAULT_BUILD_DIR), ) click.secho(msg, fg="yellow") except FunctionNotFound as function_not_found_ex: caught_exception = function_not_found_ex raise UserException( str(function_not_found_ex), wrapped_from=function_not_found_ex.__class__.__name__ ) from function_not_found_ex except ( UnsupportedRuntimeException, BuildError, BuildInsideContainerError, UnsupportedBuilderLibraryVersionError, InvalidBuildGraphException, ResourceNotFound, ) as ex: caught_exception = ex click.secho("\nBuild Failed", fg="red") # Some Exceptions have a deeper wrapped exception that needs to be surfaced # from deeper than just one level down. deep_wrap = getattr(ex, "wrapped_from", None) wrapped_from = deep_wrap if deep_wrap else ex.__class__.__name__ raise UserException(str(ex), wrapped_from=wrapped_from) from ex finally: if self.build_in_source: exception_name = type(caught_exception).__name__ if caught_exception else None EventTracker.track_event( EventName.USED_FEATURE.value, UsedFeature.BUILD_IN_SOURCE.value, exception_name ) def _is_sam_template(self) -> bool: """Check if a given template is a SAM template""" template_dict = get_template_data(self._template_file) template_transforms = template_dict.get("Transform", []) if not isinstance(template_transforms, list): template_transforms = [template_transforms] for template_transform in template_transforms: if isinstance(template_transform, str) and template_transform.startswith("AWS::Serverless"): return True return False def _handle_build_pre_processing(self) -> List[Stack]: """ Pre-modify the stacks as required before invoking the build :return: List of modified stacks """ stacks = [] if any(EsbuildBundlerManager(stack).esbuild_configured() for stack in self.stacks): # esbuild is configured in one of the stacks, will check and update stack metadata accordingly for stack in self.stacks: stacks.append(EsbuildBundlerManager(stack).set_sourcemap_metadata_from_env()) self.function_provider.update(stacks, self._use_raw_codeuri, locate_layer_nested=self._locate_layer_nested) return stacks if stacks else self.stacks def _handle_build_post_processing(self, builder: ApplicationBuilder, build_result: ApplicationBuildResult) -> None: """ Add any template modifications necessary before moving the template to build directory :param stack: Stack resources :param template: Current template file :param build_result: Result of the application build :return: Modified template dict """ artifacts = build_result.artifacts stack_output_template_path_by_stack_path = { stack.stack_path: stack.get_output_template_path(self.build_dir) for stack in self.stacks } for stack in self.stacks: modified_template = builder.update_template( stack, artifacts, stack_output_template_path_by_stack_path, ) output_template_path = stack.get_output_template_path(self.build_dir) stack_name = self._stack_name if self._stack_name else "" if self._create_auto_dependency_layer: LOG.debug("Auto creating dependency layer for each function resource into a nested stack") nested_stack_manager = NestedStackManager( stack, stack_name, self.build_dir, modified_template, build_result ) modified_template = nested_stack_manager.generate_auto_dependency_layer_stack() esbuild_manager = EsbuildBundlerManager(stack=stack, template=modified_template, build_dir=self.build_dir) if esbuild_manager.esbuild_configured(): modified_template = esbuild_manager.handle_template_post_processing() move_template(stack.location, output_template_path, modified_template) def _gen_success_msg(self, artifacts_dir: str, output_template_path: str, is_default_build_dir: bool) -> str: """ Generates a success message containing some suggested commands to run Parameters ---------- artifacts_dir: str A string path representing the folder of built artifacts output_template_path: str A string path representing the final template file is_default_build_dir: bool True if the build folder is the folder defined by SAM CLI Returns ------- str A formatted success message string """ validate_suggestion = "Validate SAM template: sam validate" invoke_suggestion = "Invoke Function: sam local invoke" sync_suggestion = "Test Function in the Cloud: sam sync --stack-name {{stack-name}} --watch" deploy_suggestion = "Deploy: sam deploy --guided" start_lambda_suggestion = "Emulate local Lambda functions: sam local start-lambda" if not is_default_build_dir and not self._hook_name: invoke_suggestion += " -t {}".format(output_template_path) deploy_suggestion += " --template-file {}".format(output_template_path) commands = [validate_suggestion, invoke_suggestion, sync_suggestion, deploy_suggestion] # check if we have used a hook package before building if self._hook_name: hook_package_flag = f" --hook-name {self._hook_name}" start_lambda_suggestion += hook_package_flag invoke_suggestion += hook_package_flag commands = [invoke_suggestion, start_lambda_suggestion] msg = f"""\nBuilt Artifacts : {artifacts_dir} Built Template : {output_template_path} Commands you can use next ========================= """ # add bullet point then join all the commands with new line msg += "[*] " + f"{os.linesep}[*] ".join(commands) return msg @staticmethod def _setup_build_dir(build_dir: str, clean: bool) -> str: build_path = pathlib.Path(build_dir) if os.path.abspath(str(build_path)) == os.path.abspath(str(pathlib.Path.cwd())): exception_message = ( "Failing build: Running a build with build-dir as current working directory " "is extremely dangerous since the build-dir contents is first removed. " "This is no longer supported, please remove the '--build-dir' option from the command " "to allow the build artifacts to be placed in the directory your template is in." ) raise InvalidBuildDirException(exception_message) if build_path.exists() and os.listdir(build_dir) and clean: # build folder contains something inside. Clear everything. shutil.rmtree(build_dir) build_path.mkdir(mode=BUILD_DIR_PERMISSIONS, parents=True, exist_ok=True) # ensure path resolving is done after creation: https://bugs.python.org/issue32434 return str(build_path.resolve()) @property def container_manager(self) -> Optional[ContainerManager]: return self._container_manager @property def function_provider(self) -> SamFunctionProvider: # Note(xinhol): despite self._function_provider is Optional # self._function_provider will be assigned with a non-None value in __enter__() and # this function is only used in the context (after __enter__ is called) # so we can assume it is not Optional here return self._function_provider # type: ignore @property def layer_provider(self) -> SamLayerProvider: # same as function_provider() return self._layer_provider # type: ignore @property def build_dir(self) -> str: return self._build_dir @property def base_dir(self) -> str: # Note(xinhol): self._base_dir will be assigned with a str value if it is None in __enter__() return self._base_dir # type: ignore @property def cache_dir(self) -> str: return self._cache_dir @property def cached(self) -> bool: return self._cached @property def use_container(self) -> bool: return self._use_container @property def stacks(self) -> List[Stack]: return self._stacks @property def manifest_path_override(self) -> Optional[str]: if self._manifest_path: return os.path.abspath(self._manifest_path) return None @property def mode(self) -> Optional[str]: return self._mode @property def use_base_dir(self) -> bool: return self._use_raw_codeuri @property def resources_to_build(self) -> ResourcesToBuildCollector: """ Function return resources that should be build by current build command. This function considers Lambda Functions and Layers with build method as buildable resources. Returns ------- ResourcesToBuildCollector """ return ( self.collect_build_resources(self._resource_identifier) if self._resource_identifier else self.collect_all_build_resources() ) @property def create_auto_dependency_layer(self) -> bool: return self._create_auto_dependency_layer @property def build_result(self) -> Optional[ApplicationBuildResult]: return self._build_result def collect_build_resources(self, resource_identifier: str) -> ResourcesToBuildCollector: """Collect a single buildable resource and its dependencies. For a Lambda function, its layers will be included. Parameters ---------- resource_identifier : str Resource identifier for the resource to be built Returns ------- ResourcesToBuildCollector ResourcesToBuildCollector containing the buildable resource and its dependencies Raises ------ ResourceNotFound raises ResourceNotFound is the specified resource cannot be found. """ result = ResourcesToBuildCollector() # Get the functions and its layer. Skips if it's inline. self._collect_single_function_and_dependent_layers(resource_identifier, result) self._collect_single_buildable_layer(resource_identifier, result) if not result.functions and not result.layers: # Collect all functions and layers that are not inline all_resources = [func.name for func in self.function_provider.get_all() if not func.inlinecode] all_resources.extend([layer.name for layer in self.layer_provider.get_all()]) available_resource_message = ( f"{resource_identifier} not found. Possible options in your " f"template: {all_resources}" ) LOG.info(available_resource_message) raise ResourceNotFound(f"Unable to find a function or layer with name '{resource_identifier}'") return result def collect_all_build_resources(self) -> ResourcesToBuildCollector: """Collect all buildable resources. Including Lambda functions and layers. Returns ------- ResourcesToBuildCollector ResourcesToBuildCollector that contains all the buildable resources. """ result = ResourcesToBuildCollector() excludes: Tuple[str, ...] = self._exclude if self._exclude is not None else () result.add_functions( [ func for func in self.function_provider.get_all() if (func.name not in excludes) and func.function_build_info.is_buildable() ] ) result.add_layers( [ layer for layer in self.layer_provider.get_all() if (layer.name not in excludes) and BuildContext.is_layer_buildable(layer) ] ) return result @property def is_building_specific_resource(self) -> bool: """ Whether customer requested to build a specific resource alone in isolation, by specifying function_identifier to the build command. Ex: sam build MyServerlessFunction :return: True if user requested to build specific resource, False otherwise """ return bool(self._resource_identifier) def _collect_single_function_and_dependent_layers( self, resource_identifier: str, resource_collector: ResourcesToBuildCollector ) -> None: """ Populate resource_collector with function with provided identifier and all layers that function need to be build in resource_collector Parameters ---------- resource_collector: Collector that will be populated with resources. """ function = self.function_provider.get(resource_identifier) if not function: # No function found return resource_collector.add_function(function) resource_collector.add_layers([layer for layer in function.layers if BuildContext.is_layer_buildable(layer)]) def _collect_single_buildable_layer( self, resource_identifier: str, resource_collector: ResourcesToBuildCollector ) -> None: """ Populate resource_collector with layer with provided identifier. Parameters ---------- resource_collector Returns ------- """ layer = self.layer_provider.get(resource_identifier) if not layer: # No layer found return if layer and layer.build_method is None: LOG.error("Layer %s is missing BuildMethod Metadata.", self._function_provider) raise MissingBuildMethodException(f"Build method missing in layer {resource_identifier}.") resource_collector.add_layer(layer) @staticmethod def is_layer_buildable(layer: LayerVersion): # if build method is not specified, it is not buildable if not layer.build_method: LOG.debug("Skip building layer without a build method: %s", layer.full_path) return False # no need to build layers that are already packaged as a zip file if isinstance(layer.codeuri, str) and layer.codeuri.endswith(".zip"): LOG.debug("Skip building zip layer: %s", layer.full_path) return False # skip build the functions that marked as skip-build if layer.skip_build: LOG.debug("Skip building pre-built layer: %s", layer.full_path) return False return True _EXCLUDE_WARNING_MESSAGE = "Resource expected to be built, but marked as excluded.\nBuilding anyways..." def _check_exclude_warning(self) -> None: """ Prints warning message if a single resource to build is also being excluded """ excludes: Tuple[str, ...] = self._exclude if self._exclude is not None else () if self._resource_identifier in excludes: LOG.warning(self._EXCLUDE_WARNING_MESSAGE) def _check_rust_cargo_experimental_flag(self) -> None: """ Prints warning message and confirms if user wants to use beta feature """ WARNING_MESSAGE = ( 'Build method "rust-cargolambda" is a beta feature.\n' "Please confirm if you would like to proceed\n" 'You can also enable this beta feature with "sam build --beta-features".' ) resources_to_build = self.get_resources_to_build() is_building_rust = False for function in resources_to_build.functions: if function.metadata and function.metadata.get("BuildMethod", "") == "rust-cargolambda": is_building_rust = True break if is_building_rust: prompt_experimental(ExperimentalFlag.RustCargoLambda, WARNING_MESSAGE) @property def build_in_source(self) -> Optional[bool]: return self._build_in_source