""" WatchManager for Sync Watch Logic """ import logging import platform import threading import time from pathlib import Path from typing import TYPE_CHECKING, Dict, List, Optional, Set from watchdog.events import EVENT_TYPE_MODIFIED, EVENT_TYPE_OPENED, FileSystemEvent from samcli.lib.providers.exceptions import InvalidTemplateFile, MissingCodeUri, MissingLocalDefinition from samcli.lib.providers.provider import ResourceIdentifier, Stack, get_all_resource_ids from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider from samcli.lib.sync.continuous_sync_flow_executor import ContinuousSyncFlowExecutor from samcli.lib.sync.exceptions import InfraSyncRequiredError, MissingPhysicalResourceError, SyncFlowException from samcli.lib.sync.infra_sync_executor import InfraSyncExecutor, InfraSyncResult from samcli.lib.sync.sync_flow_factory import SyncFlowFactory from samcli.lib.utils.code_trigger_factory import CodeTriggerFactory from samcli.lib.utils.colors import Colored, Colors from samcli.lib.utils.path_observer import HandlerObserver from samcli.lib.utils.resource_trigger import OnChangeCallback, TemplateTrigger from samcli.local.lambdafn.exceptions import ResourceNotFound if TYPE_CHECKING: # pragma: no cover from samcli.commands.build.build_context import BuildContext from samcli.commands.deploy.deploy_context import DeployContext from samcli.commands.package.package_context import PackageContext from samcli.commands.sync.sync_context import SyncContext DEFAULT_WAIT_TIME = 1 LOG = logging.getLogger(__name__) class WatchManager: _stacks: Optional[List[Stack]] _template: str _build_context: "BuildContext" _package_context: "PackageContext" _deploy_context: "DeployContext" _sync_context: "SyncContext" _sync_flow_factory: Optional[SyncFlowFactory] _sync_flow_executor: ContinuousSyncFlowExecutor _executor_thread: Optional[threading.Thread] _observer: HandlerObserver _trigger_factory: Optional[CodeTriggerFactory] _waiting_infra_sync: bool _color: Colored _auto_dependency_layer: bool _disable_infra_syncs: bool def __init__( self, template: str, build_context: "BuildContext", package_context: "PackageContext", deploy_context: "DeployContext", sync_context: "SyncContext", auto_dependency_layer: bool, disable_infra_syncs: bool, watch_exclude: Dict[str, List[str]], ): """Manager for sync watch execution logic. This manager will observe template and its code resources. Automatically execute infra/code syncs when changes are detected. Parameters ---------- template : str Template file path build_context : BuildContext BuildContext package_context : PackageContext PackageContext deploy_context : DeployContext DeployContext """ self._stacks = None self._template = template self._build_context = build_context self._package_context = package_context self._deploy_context = deploy_context self._sync_context = sync_context self._auto_dependency_layer = auto_dependency_layer self._disable_infra_syncs = disable_infra_syncs self._sync_flow_factory = None self._sync_flow_executor = ContinuousSyncFlowExecutor() self._executor_thread = None self._observer = HandlerObserver() self._trigger_factory = None self._waiting_infra_sync = False self._color = Colored() self._watch_exclude = watch_exclude def queue_infra_sync(self) -> None: """Queue up an infra structure sync. A simple bool flag is suffice """ if self._disable_infra_syncs: LOG.info( self._color.color_log( msg="You have enabled the --code flag, which limits sam sync updates to code changes only. To do a " "complete infrastructure and code sync, remove the --code flag.", color=Colors.WARNING, ), extra=dict(markup=True), ) return self._waiting_infra_sync = True def _update_stacks(self) -> None: """ Reloads template and its stacks. Update all other member that also depends on the stacks. This should be called whenever there is a change to the template. """ self._stacks = SamLocalStackProvider.get_stacks(self._template, use_sam_transform=False)[0] self._sync_flow_factory = SyncFlowFactory( self._build_context, self._deploy_context, self._sync_context, self._stacks, self._auto_dependency_layer, ) self._sync_flow_factory.load_physical_id_mapping() self._trigger_factory = CodeTriggerFactory(self._stacks, Path(self._build_context.base_dir)) def _add_code_triggers(self) -> None: """Create CodeResourceTrigger for all resources and add their handlers to observer""" if not self._stacks or not self._trigger_factory: return resource_ids = get_all_resource_ids(self._stacks) for resource_id in resource_ids: try: additional_excludes = self._watch_exclude.get(str(resource_id), []) trigger = self._trigger_factory.create_trigger( resource_id, self._on_code_change_wrapper(resource_id), additional_excludes ) except (MissingCodeUri, MissingLocalDefinition): LOG.warning( self._color.color_log( msg="CodeTrigger not created as CodeUri or DefinitionUri is missing for %s.", color=Colors.WARNING, ), str(resource_id), extra=dict(markup=True), ) continue except ResourceNotFound: LOG.warning( self._color.color_log( msg="CodeTrigger not created as %s is not found or is with a S3 Location.", color=Colors.WARNING, ), str(resource_id), extra=dict(markup=True), ) continue if not trigger: continue self._observer.schedule_handlers(trigger.get_path_handlers()) def _add_template_triggers(self) -> None: """Create TemplateTrigger and add its handlers to observer""" stacks = SamLocalStackProvider.get_stacks(self._template, use_sam_transform=False)[0] for stack in stacks: template = stack.location template_trigger = TemplateTrigger(template, stack.name, lambda _=None: self.queue_infra_sync()) try: template_trigger.validate_template() except InvalidTemplateFile: LOG.warning( self._color.color_log(msg="Template validation failed for %s in %s", color=Colors.WARNING), template, stack.name, extra=dict(markup=True), ) self._observer.schedule_handlers(template_trigger.get_path_handlers()) def _execute_infra_context(self, first_sync: bool = False) -> InfraSyncResult: """Execute infrastructure sync Returns ---------- InfraSyncResult Returns information containing whether infra sync executed plus resources to do code sync on """ self._infra_sync_executor = InfraSyncExecutor( self._build_context, self._package_context, self._deploy_context, self._sync_context ) return self._infra_sync_executor.execute_infra_sync(first_sync) def _start_code_sync(self) -> None: """Start SyncFlowExecutor in a separate thread.""" if not self._executor_thread or not self._executor_thread.is_alive(): self._executor_thread = threading.Thread( target=lambda: self._sync_flow_executor.execute( exception_handler=self._watch_sync_flow_exception_handler ) ) self._executor_thread.start() def _stop_code_sync(self) -> None: """Blocking call that stops SyncFlowExecutor and waits for it to finish.""" if self._executor_thread and self._executor_thread.is_alive(): self._sync_flow_executor.stop() self._executor_thread.join() def start(self) -> None: """Start WatchManager and watch for changes to the template and its code resources.""" # The actual execution is done in _start() # This is a wrapper for gracefully handling Ctrl+C or other termination cases. try: self.queue_infra_sync() if self._disable_infra_syncs: self._start_sync() LOG.info( self._color.color_log(msg="Sync watch started.", color=Colors.SUCCESS), extra=dict(markup=True) ) self._start() except KeyboardInterrupt: LOG.info( self._color.color_log(msg="Shutting down sync watch...", color=Colors.PROGRESS), extra=dict(markup=True) ) self._observer.stop() self._stop_code_sync() LOG.info(self._color.color_log(msg="Sync watch stopped.", color=Colors.SUCCESS), extra=dict(markup=True)) def _start(self) -> None: """Start WatchManager and watch for changes to the template and its code resources.""" first_sync = True self._observer.start() while True: if self._waiting_infra_sync: self._execute_infra_sync(first_sync) first_sync = False time.sleep(1) def _start_sync(self) -> None: """Update stacks and populate all triggers""" self._observer.unschedule_all() self._update_stacks() self._add_template_triggers() self._add_code_triggers() self._start_code_sync() def _execute_infra_sync(self, first_sync: bool = False) -> None: """Logic to execute infra sync.""" LOG.info( self._color.color_log( msg="Queued infra sync. Waiting for in progress code syncs to complete...", color=Colors.PROGRESS ), extra=dict(markup=True), ) self._waiting_infra_sync = False self._stop_code_sync() try: LOG.info(self._color.color_log(msg="Starting infra sync.", color=Colors.PROGRESS), extra=dict(markup=True)) infra_sync_result = self._execute_infra_context(first_sync) except Exception as e: LOG.error( self._color.color_log( msg="Failed to sync infra. Code sync is paused until template/stack is fixed.", color=Colors.FAILURE ), exc_info=e, extra=dict(markup=True), ) # Unschedule all triggers and only add back the template one as infra sync is incorrect. self._observer.unschedule_all() self._add_template_triggers() else: # Update stacks and repopulate triggers # Trigger are not removed until infra sync is finished as there # can be code changes during infra sync. self._start_sync() if not infra_sync_result.infra_sync_executed: # This is for initiating code sync for all resources # To improve: only initiate code syncs for ones with template changes self._queue_up_code_syncs(infra_sync_result.code_sync_resources) LOG.info( self._color.color_log( msg="Skipped infra sync as the local template is in sync with the cloud template.", color=Colors.SUCCESS, ), extra=dict(markup=True), ) if len(infra_sync_result.code_sync_resources) != 0: LOG.info("Required code syncs are queued up.") else: LOG.info( self._color.color_log(msg="Infra sync completed.", color=Colors.SUCCESS), extra=dict(markup=True) ) def _queue_up_code_syncs(self, resource_ids_with_code_sync: Set[ResourceIdentifier]) -> None: """ For ther given resource IDs, create sync flow tasks in the queue Parameters ---------- resource_ids_with_code_sync: Set[ResourceIdentifier] The set of resource IDs to be synced """ if not self._sync_flow_factory: return for resource_id in resource_ids_with_code_sync: sync_flow = self._sync_flow_factory.create_sync_flow(resource_id, self._build_context.build_result) if sync_flow: self._sync_flow_executor.add_delayed_sync_flow(sync_flow) def _on_code_change_wrapper(self, resource_id: ResourceIdentifier) -> OnChangeCallback: """Wrapper method that generates a callback for code changes. Parameters ---------- resource_id : ResourceIdentifier Resource that associates to the callback Returns ------- OnChangeCallback Callback function """ def on_code_change(event: Optional[FileSystemEvent] = None) -> None: """ Custom event handling to create a new sync flow if a file was modified. Parameters ---------- event: Optional[FileSystemEvent] The event that triggered the change """ if event and event.event_type == EVENT_TYPE_OPENED: # Ignore all file opened events since this event is # added in addition to a create or modified event, # causing an infinite loop of sync flow creations LOG.debug("Ignoring file system OPENED event") return if ( platform.system().lower() == "linux" and event and event.event_type == EVENT_TYPE_MODIFIED and event.is_directory ): # Linux machines appear to emit an additional event when # a file gets updated; a folder modfied event # If folder/file.txt gets updated, there will be two events: # 1. file.txt modified event # 2. folder modified event # We want to ignore the second event # # It looks like the other way a folder modified event can happen # is if the permissions of the folder were changed LOG.debug(f"Ignoring file system MODIFIED event for folder {event.src_path!r}") return # sync flow factory should always exist, but guarding just incase if not self._sync_flow_factory: LOG.debug("Sync flow factory not defined, skipping trigger") return sync_flow = self._sync_flow_factory.create_sync_flow(resource_id) if sync_flow and not self._waiting_infra_sync: self._sync_flow_executor.add_delayed_sync_flow(sync_flow, dedup=True, wait_time=DEFAULT_WAIT_TIME) return on_code_change def _watch_sync_flow_exception_handler(self, sync_flow_exception: SyncFlowException) -> None: """Exception handler for watch. Simply logs unhandled exceptions instead of failing the entire process. Parameters ---------- sync_flow_exception : SyncFlowException SyncFlowException """ exception = sync_flow_exception.exception if isinstance(exception, MissingPhysicalResourceError): LOG.warning( self._color.color_log( msg="Missing physical resource. Infra sync will be started.", color=Colors.WARNING ), extra=dict(markup=True), ) self.queue_infra_sync() elif isinstance(exception, InfraSyncRequiredError): LOG.warning( self._color.yellow( f"Infra sync is required for {exception.resource_identifier} due to: " + f"{exception.reason}. Infra sync will be started." ), extra=dict(markup=True), ) self.queue_infra_sync() else: LOG.error( self._color.color_log(msg="Code sync encountered an error.", color=Colors.FAILURE), exc_info=exception, extra=dict(markup=True), )