sync_flow.py

"""SyncFlow base class """
import logging
from abc import ABC, abstractmethod
from enum import Enum
from os import environ
from pathlib import Path
from threading import Lock
from typing import TYPE_CHECKING, Any, Dict, List, NamedTuple, Optional, Set, cast

from boto3.session import Session

from samcli.lib.build.app_builder import ApplicationBuildResult
from samcli.lib.providers.provider import ResourceIdentifier, Stack, get_resource_by_id
from samcli.lib.sync.exceptions import MissingLockException, MissingPhysicalResourceError
from samcli.lib.utils.boto_utils import get_boto_client_provider_from_session_with_config
from samcli.lib.utils.lock_distributor import LockChain, LockDistributor
from samcli.lib.utils.resources import RESOURCES_WITH_LOCAL_PATHS

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.sync.sync_context import SyncContext

# Logging with multiple processes is not safe. Use a log queue in the future.
# https://docs.python.org/3/howto/logging-cookbook.html#:~:text=Although%20logging%20is%20thread%2Dsafe,across%20multiple%20processes%20in%20Python.
LOG = logging.getLogger(__name__)


def get_default_retry_config() -> Optional[Dict]:
    """
    Returns a default retry config if nothing is overriden by environment variables
    """
    if environ.get("AWS_MAX_ATTEMPTS") or environ.get("AWS_RETRY_MODE"):
        return None
    return {"max_attempts": 10, "mode": "standard"}


class ApiCallTypes(Enum):
    """API call stages that can be locked on"""

    BUILD = "Build"
    UPDATE_FUNCTION_CONFIGURATION = "UpdateFunctionConfiguration"
    UPDATE_FUNCTION_CODE = "UpdateFunctionCode"


class ResourceAPICall(NamedTuple):
    """Named tuple for a resource and its potential API calls"""

    shared_resource: str
    api_calls: List[ApiCallTypes]


class SyncFlow(ABC):
    """Base class for a SyncFlow"""

    _log_name: str
    _build_context: "BuildContext"
    _deploy_context: "DeployContext"
    _sync_context: "SyncContext"
    _stacks: Optional[List[Stack]]
    _session: Optional[Session]
    _physical_id_mapping: Dict[str, str]
    _locks: Optional[Dict[str, Lock]]
    # Local hash represents the state of a particular sync flow
    # We store the hash value in sync state toml file as value
    _local_sha: Optional[str]
    _application_build_result: Optional[ApplicationBuildResult]

    def __init__(
        self,
        build_context: "BuildContext",
        deploy_context: "DeployContext",
        sync_context: "SyncContext",
        physical_id_mapping: Dict[str, str],
        log_name: str,
        stacks: Optional[List[Stack]] = None,
        application_build_result: Optional[ApplicationBuildResult] = None,
    ):
        """
        Parameters
        ----------
        build_context : BuildContext
            BuildContext used for build related parameters
        deploy_context : BuildContext
            DeployContext used for this deploy related parameters
        sync_context: SyncContext
            SyncContext object that obtains sync information.
        physical_id_mapping : Dict[str, str]
            Mapping between resource logical identifier and physical identifier
        log_name : str
            Name to be used for logging purposes
        stacks : List[Stack], optional
            List of stacks containing a root stack and optional nested stacks
         application_build_result: Optional[ApplicationBuildResult]
            Pre-build ApplicationBuildResult which can be re-used during SyncFlows
        """
        self._build_context = build_context
        self._deploy_context = deploy_context
        self._sync_context = sync_context
        self._log_name = log_name
        self._stacks = stacks
        self._session = None
        self._physical_id_mapping = physical_id_mapping
        self._locks = None
        self._local_sha = None
        self._application_build_result = application_build_result

    def set_up(self) -> None:
        """Clients and other expensives setups should be handled here instead of constructor"""
        pass

    def _get_session(self) -> Session:
        if not self._session:
            self._session = Session(profile_name=self._deploy_context.profile, region_name=self._deploy_context.region)
        return self._session

    def _boto_client(self, client_name: str):
        default_retry_config = get_default_retry_config()
        if not default_retry_config:
            LOG.debug("Creating boto client (%s) with user's retry config", client_name)
            return get_boto_client_provider_from_session_with_config(self._get_session())(client_name)

        LOG.debug("Creating boto client (%s) with default retry config", client_name)
        return get_boto_client_provider_from_session_with_config(self._get_session(), retries=default_retry_config)(
            client_name
        )

    @property
    @abstractmethod
    def sync_state_identifier(self) -> str:
        """
        Sync state is the unique identifier for each sync flow
        We store the identifier in sync state toml file as key
        """
        raise NotImplementedError("sync_state_identifier")

    @abstractmethod
    def gather_resources(self) -> None:
        """Local operations that need to be done before comparison and syncing with remote
        Ex: Building lambda functions
        """
        raise NotImplementedError("gather_resources")

    def _update_local_hash(self) -> None:
        """Updates the latest local hash of the sync flow which then can be used for comparison for next run"""
        if not self._local_sha:
            LOG.debug("%sNo local hash is configured, skipping to update local hash", self.log_prefix)
            return

        self._sync_context.update_resource_sync_state(self.sync_state_identifier, self._local_sha)

    def compare_local(self) -> bool:
        """Comparison between local resource and its local stored state.
        If the resources are identical, sync and gather dependencies will be skipped.
        Simply return False if there is no comparison needed.
        Ex: Comparing local Lambda function artifact with stored SHA256

        Returns
        -------
        bool
            Return True if current resource and cached are in sync. Skipping rest of the execution.
            Return False otherwise.
        """
        stored_sha = self._sync_context.get_resource_latest_sync_hash(self.sync_state_identifier)
        LOG.debug("%sLocal SHA: %s Stored SHA: %s", self.log_prefix, self._local_sha, stored_sha)
        if self._local_sha and stored_sha and self._local_sha == stored_sha:
            return True
        return False

    @abstractmethod
    def compare_remote(self) -> bool:
        """Comparison between local and remote resources.
        This can be used for optimization if comparison is a lot faster than sync.
        If the resources are identical, sync and gather dependencies will be skipped.
        Simply return False if there is no comparison needed.
        Ex: Comparing local Lambda function artifact with remote SHA256

        Returns
        -------
        bool
            Return True if local and remote are in sync. Skipping rest of the execution.
            Return False otherwise.
        """
        raise NotImplementedError("compare_remote")

    @abstractmethod
    def sync(self) -> None:
        """Step that syncs local resources with remote.
        Ex: Call UpdateFunctionCode for Lambda Functions
        """
        raise NotImplementedError("sync")

    @abstractmethod
    def gather_dependencies(self) -> List["SyncFlow"]:
        """Gather a list of SyncFlows that should be executed after the current change.
        This can be sync flows for other resources that depends on the current one.
        Ex: Update Lambda functions if a layer sync flow creates a new version.

        Returns
        ------
        List[SyncFlow]
            List of sync flows that need to be executed after the current one finishes.
        """
        raise NotImplementedError("update_dependencies")

    @abstractmethod
    def _get_resource_api_calls(self) -> List[ResourceAPICall]:
        """Get resources and their associating API calls. This is used for locking purposes.
        Returns
        -------
        Dict[str, List[str]]
            Key as resource logical ID
            Value as list of api calls that the resource can make
        """
        raise NotImplementedError("_get_resource_api_calls")

    def has_locks(self) -> bool:
        """Check if a sync flow has locks and needs to enter a lock context
        Returns
        -------
        bool
            whether or not a sync flow contains locks
        """
        return bool(self._locks)

    def get_lock_keys(self) -> Set[str]:
        """Get a list of function + API calls that can be used as keys for LockDistributor

        Returns
        -------
        Set[str]
            Set of keys for all resources and their API calls
        """
        lock_keys = set()
        for resource_api_calls in self._get_resource_api_calls():
            for api_call in resource_api_calls.api_calls:
                lock_keys.add(SyncFlow._get_lock_key(resource_api_calls.shared_resource, api_call))
        return lock_keys

    def set_locks_with_distributor(self, distributor: LockDistributor):
        """Set locks to be used with a LockDistributor. Keys should be generated using get_lock_keys().

        Parameters
        ----------
        distributor : LockDistributor
            Lock distributor
        """
        self.set_locks_with_dict(distributor.get_locks(self.get_lock_keys()))

    def set_locks_with_dict(self, locks: Dict[str, Lock]):
        """Set locks to be used. Keys should be generated using get_lock_keys().

        Parameters
        ----------
        locks : Dict[str, Lock]
            Dict of locks with keys from get_lock_keys()
        """
        self._locks = locks

    @staticmethod
    def _get_lock_key(logical_id: str, api_call: ApiCallTypes) -> str:
        """Get a single lock key for a pair of resource and API call.

        Parameters
        ----------
        logical_id : str
            Logical ID of a resource.
        api_call : str
            API call the resource will use.

        Returns
        -------
        str
            String key created with logical ID and API call name.
        """
        return f"{logical_id}_{api_call.value}"

    def _get_lock_chain(self) -> LockChain:
        """Return a LockChain object for all the locks

        Returns
        -------
        Optional[LockChain]
            A LockChain object containing all locks. None if there are no locks.
        """
        if self._locks:
            return LockChain(self._locks)
        raise MissingLockException("Missing Locks for LockChain")

    def _get_resource(self, resource_identifier: str) -> Optional[Dict[str, Any]]:
        """Get a resource dict with resource identifier

        Parameters
        ----------
        resource_identifier : str
            Resource identifier

        Returns
        -------
        Optional[Dict[str, Any]]
            Resource dict containing its template fields.
        """
        return get_resource_by_id(self._stacks, ResourceIdentifier(resource_identifier)) if self._stacks else None

    def get_physical_id(self, resource_identifier: str) -> str:
        """Get the physical ID of a resource using physical_id_mapping. This does not directly check with remote.

        Parameters
        ----------
        resource_identifier : str
            Resource identifier

        Returns
        -------
        str
            Resource physical ID

        Raises
        ------
        MissingPhysicalResourceError
            Resource does not exist in the physical ID mapping.
            This could mean remote and local templates are not in sync.
        """
        physical_id = self._physical_id_mapping.get(resource_identifier)
        if not physical_id:
            raise MissingPhysicalResourceError(resource_identifier)

        return physical_id

    @abstractmethod
    def _equality_keys(self) -> Any:
        """This method needs to be overridden to distinguish between multiple instances of SyncFlows
        If the return values of two instances are the same, then those two instances will be assumed to be equal.

        Returns
        -------
        Any
            Anything that can be hashed and compared with "=="
        """
        raise NotImplementedError("_equality_keys is not implemented.")

    def __hash__(self) -> int:
        return hash((type(self), self._equality_keys()))

    def __eq__(self, o: object) -> bool:
        if type(o) is not type(self):
            return False
        return cast(bool, self._equality_keys() == cast(SyncFlow, o)._equality_keys())

    @property
    def log_name(self) -> str:
        """
        Returns
        -------
        str
            Human readable name/identifier for logging purposes
        """
        return self._log_name

    @property
    def log_prefix(self) -> str:
        """
        Returns
        -------
        str
            Log prefix to be used for logging.
        """
        return f"SyncFlow [{self.log_name}]: "

    def execute(self) -> List["SyncFlow"]:
        """Execute the sync flow and returns a list of dependent sync flows.
        Skips sync() and gather_dependencies() if compare() is True

        Returns
        -------
        List[SyncFlow]
            A list of dependent sync flows
        """
        dependencies: List["SyncFlow"] = list()
        LOG.debug("%sSetting Up", self.log_prefix)
        self.set_up()
        LOG.debug("%sGathering Resources", self.log_prefix)
        self.gather_resources()
        LOG.debug("%sComparing with Remote", self.log_prefix)
        if (not self.compare_local()) and (not self.compare_remote()):
            LOG.debug("%sSyncing", self.log_prefix)
            self.sync()
            LOG.debug("%sUpdating local hash of the sync flow", self.log_prefix)
            self._update_local_hash()
            LOG.debug("%sGathering Dependencies", self.log_prefix)
            dependencies = self.gather_dependencies()
        else:
            LOG.info("%sSkipping resource update as the content didn't change", self.log_prefix)
        LOG.debug("%sFinished", self.log_prefix)
        return dependencies


def get_definition_path(
    resource: Dict, identifier: str, use_base_dir: bool, base_dir: str, stacks: List[Stack]
) -> Optional[Path]:
    """
    A helper method used by non-function sync flows to resolve definition file path
    that are relative to the child stack to absolute path for nested stacks

    Parameters
    -------
    resource: Dict
        The resource's template dict
    identifier: str
        The logical ID identifier of the resource
    use_base_dir: bool
        Whether or not the base_dir option was used
    base_dir: str
        Base directory if provided, otherwise the root template directory
    stacks: List[Stack]
        The list of stacks for the application

    Returns
    -------
    Optional[Path]
        A resolved absolute path for the definition file
    """
    definition_field_names = RESOURCES_WITH_LOCAL_PATHS.get(resource.get("Type", ""))
    if not definition_field_names:
        LOG.error("Couldn't find definition field name for resource %s", identifier)
        return None
    definition_field_name = definition_field_names[0]
    LOG.debug("Found definition field name as %s", definition_field_name)

    properties = resource.get("Properties", {})
    definition_file = properties.get(definition_field_name)
    definition_path = None
    if definition_file:
        definition_path = Path(base_dir).joinpath(definition_file)
        if not use_base_dir:
            child_stack = Stack.get_stack_by_full_path(ResourceIdentifier(identifier).stack_path, stacks)
            if child_stack:
                definition_path = Path(child_stack.location).parent.joinpath(definition_file)
    return definition_path