fix(Accelerate): Added Locking to Layer Building (aws#358)

* Added Locking to Layer Building

* Fixed LockChain Exception

* Added Comment

Author: CoshUS

samcli/lib/providers/provider.py

@@ -51,7 +51,7 @@ class Function(NamedTuple):
     # to get credentials to run the container with. This gives a much higher fidelity simulation of cloud Lambda.
     rolearn: Optional[str]
     # List of Layers
-    layers: List
+    layers: List["LayerVersion"]
     # Event
     events: Optional[List]
     # Metadata

samcli/lib/sync/exceptions.py

@@ -86,6 +86,10 @@ def stack_resource_names(self) -> List[str]:
         return self._stack_resource_names
 
 
+class MissingLockException(Exception):
+    """Exception for not having an associated lock to be used."""
+
+
 class UriNotFoundException(Exception):
     """Exception used for not having a URI field that the resource requires"""
 

samcli/lib/sync/flows/function_sync_flow.py

@@ -1,6 +1,6 @@
 """Base SyncFlow for Lambda Function"""
 import logging
-from typing import Any, Dict, List, Optional, TYPE_CHECKING, cast
+from typing import Any, Dict, List, TYPE_CHECKING, cast
 
 from boto3.session import Session
 
@@ -21,7 +21,7 @@
 class FunctionSyncFlow(SyncFlow):
     _function_identifier: str
     _function_provider: SamFunctionProvider
-    _function: Optional[Function]
+    _function: Function
     _lambda_client: Any
     _lambda_waiter: Any
     _lambda_waiter_config: Dict[str, Any]
@@ -57,7 +57,7 @@ def __init__(
         )
         self._function_identifier = function_identifier
         self._function_provider = self._build_context.function_provider
-        self._function = self._function_provider.functions.get(self._function_identifier)
+        self._function = cast(Function, self._function_provider.functions.get(self._function_identifier))
         self._lambda_client = None
         self._lambda_waiter = None
         self._lambda_waiter_config = {"Delay": 1, "MaxAttempts": 60}

samcli/lib/sync/flows/layer_sync_flow.py

@@ -83,19 +83,20 @@ def set_up(self) -> None:
 
     def gather_resources(self) -> None:
         """Build layer and ZIP it into a temp file in self._zip_file"""
-        builder = ApplicationBuilder(
-            self._build_context.collect_build_resources(self._layer_identifier),
-            self._build_context.build_dir,
-            self._build_context.base_dir,
-            self._build_context.cache_dir,
-            cached=False,
-            is_building_specific_resource=True,
-            manifest_path_override=self._build_context.manifest_path_override,
-            container_manager=self._build_context.container_manager,
-            mode=self._build_context.mode,
-        )
-        LOG.debug("%sBuilding Layer", self.log_prefix)
-        self._artifact_folder = builder.build().get(self._layer_identifier)
+        with self._get_lock_chain():
+            builder = ApplicationBuilder(
+                self._build_context.collect_build_resources(self._layer_identifier),
+                self._build_context.build_dir,
+                self._build_context.base_dir,
+                self._build_context.cache_dir,
+                cached=False,
+                is_building_specific_resource=True,
+                manifest_path_override=self._build_context.manifest_path_override,
+                container_manager=self._build_context.container_manager,
+                mode=self._build_context.mode,
+            )
+            LOG.debug("%sBuilding Layer", self.log_prefix)
+            self._artifact_folder = builder.build().get(self._layer_identifier)
 
         zip_file_path = os.path.join(tempfile.gettempdir(), f"data-{uuid.uuid4().hex}")
         self._zip_file = make_zip(zip_file_path, self._artifact_folder)
@@ -179,7 +180,7 @@ def gather_dependencies(self) -> List[SyncFlow]:
         return dependencies
 
     def _get_resource_api_calls(self) -> List[ResourceAPICall]:
-        return []
+        return [ResourceAPICall(self._layer_identifier, ["Build"])]
 
     def _get_latest_layer_version(self):
         """Fetches all layer versions from remote and returns the latest one"""

samcli/lib/sync/flows/zip_function_sync_flow.py

@@ -71,19 +71,20 @@ def set_up(self) -> None:
 
     def gather_resources(self) -> None:
         """Build function and ZIP it into a temp file in self._zip_file"""
-        builder = ApplicationBuilder(
-            self._build_context.collect_build_resources(self._function_identifier),
-            self._build_context.build_dir,
-            self._build_context.base_dir,
-            self._build_context.cache_dir,
-            cached=False,
-            is_building_specific_resource=True,
-            manifest_path_override=self._build_context.manifest_path_override,
-            container_manager=self._build_context.container_manager,
-            mode=self._build_context.mode,
-        )
-        LOG.debug("%sBuilding Function", self.log_prefix)
-        self._artifact_folder = builder.build().get(self._function_identifier)
+        with self._get_lock_chain():
+            builder = ApplicationBuilder(
+                self._build_context.collect_build_resources(self._function_identifier),
+                self._build_context.build_dir,
+                self._build_context.base_dir,
+                self._build_context.cache_dir,
+                cached=False,
+                is_building_specific_resource=True,
+                manifest_path_override=self._build_context.manifest_path_override,
+                container_manager=self._build_context.container_manager,
+                mode=self._build_context.mode,
+            )
+            LOG.debug("%sBuilding Function", self.log_prefix)
+            self._artifact_folder = builder.build().get(self._function_identifier)
 
         zip_file_path = os.path.join(tempfile.gettempdir(), "data-" + uuid.uuid4().hex)
         self._zip_file = make_zip(zip_file_path, self._artifact_folder)
@@ -134,4 +135,7 @@ def sync(self) -> None:
             os.remove(self._zip_file)
 
     def _get_resource_api_calls(self) -> List[ResourceAPICall]:
-        return []
+        resource_calls = list()
+        for layer in self._function.layers:
+            resource_calls.append(ResourceAPICall(layer.full_path, ["Build"]))
+        return resource_calls

samcli/lib/sync/sync_flow.py

@@ -10,7 +10,7 @@
 
 from samcli.lib.providers.provider import ResourceIdentifier, Stack
 from samcli.lib.utils.lock_distributor import LockDistributor, LockChain
-from samcli.lib.sync.exceptions import MissingPhysicalResourceError
+from samcli.lib.sync.exceptions import MissingLockException, MissingPhysicalResourceError
 
 if TYPE_CHECKING:
     from samcli.commands.deploy.deploy_context import DeployContext
@@ -179,7 +179,7 @@ def _get_lock_key(logical_id: str, api_call: str) -> str:
         """
         return logical_id + "_" + api_call
 
-    def _get_lock_chain(self) -> Optional[LockChain]:
+    def _get_lock_chain(self) -> LockChain:
         """Return a LockChain object for all the locks
 
         Returns
@@ -189,7 +189,7 @@ def _get_lock_chain(self) -> Optional[LockChain]:
         """
         if self._locks:
             return LockChain(self._locks)
-        return None
+        raise MissingLockException("Missing Locks for LockChain")
 
     def _get_resource(self, resource_identifier: str) -> Optional[Dict[str, Any]]:
         """Get a resource dict with resource identifier

tests/unit/lib/sync/flows/test_layer_sync_flow.py

@@ -1,7 +1,7 @@
 import base64
 import hashlib
 from unittest import TestCase
-from unittest.mock import Mock, patch, call, ANY, mock_open, PropertyMock
+from unittest.mock import MagicMock, Mock, patch, call, ANY, mock_open, PropertyMock
 
 from parameterized import parameterized
 
@@ -86,6 +86,8 @@ def test_setup_gather_resources(
         given_file_checksum = Mock()
         patched_file_checksum.return_value = given_file_checksum
 
+        self.layer_sync_flow._get_lock_chain = MagicMock()
+
         self.layer_sync_flow.gather_resources()
 
         self.build_context_mock.collect_build_resources.assert_called_with(self.layer_identifier)
@@ -112,6 +114,10 @@ def test_setup_gather_resources(
         self.assertEqual(self.layer_sync_flow._zip_file, given_zip_location)
         self.assertEqual(self.layer_sync_flow._local_sha, given_file_checksum)
 
+        self.layer_sync_flow._get_lock_chain.assert_called_once()
+        self.layer_sync_flow._get_lock_chain.return_value.__enter__.assert_called_once()
+        self.layer_sync_flow._get_lock_chain.return_value.__exit__.assert_called_once()
+
     def test_compare_remote(self):
         given_lambda_client = Mock()
         self.layer_sync_flow._lambda_client = given_lambda_client
@@ -325,6 +331,12 @@ def test_get_latest_layer_version_error(self):
         with self.assertRaises(NoLayerVersionsFoundError):
             self.layer_sync_flow._get_latest_layer_version()
 
+    @patch("samcli.lib.sync.flows.layer_sync_flow.ResourceAPICall")
+    def test_get_resource_api_calls(self, resource_api_call_mock):
+        result = self.layer_sync_flow._get_resource_api_calls()
+        self.assertEqual(len(result), 1)
+        resource_api_call_mock.assert_called_once_with(self.layer_identifier, ["Build"])
+
 
 class TestFunctionLayerReferenceSync(TestCase):
     def setUp(self):

tests/unit/lib/sync/flows/test_zip_function_sync_flow.py

@@ -46,6 +46,8 @@ def test_gather_resources(
         file_checksum_mock.return_value = "sha256_value"
         sync_flow = self.create_function_sync_flow()
 
+        sync_flow._get_lock_chain = MagicMock()
+
         sync_flow.set_up()
         sync_flow.gather_resources()
 
@@ -54,6 +56,9 @@ def test_gather_resources(
         make_zip_mock.assert_called_once_with("temp_folder" + os.sep + "data-uuid_value", "ArtifactFolder1")
         file_checksum_mock.assert_called_once_with("zip_file", sha256_mock.return_value)
         self.assertEqual("sha256_value", sync_flow._local_sha)
+        sync_flow._get_lock_chain.assert_called_once()
+        sync_flow._get_lock_chain.return_value.__enter__.assert_called_once()
+        sync_flow._get_lock_chain.return_value.__exit__.assert_called_once()
 
     @patch("samcli.lib.sync.flows.zip_function_sync_flow.base64.b64decode")
     @patch("samcli.lib.sync.sync_flow.Session")
@@ -146,3 +151,26 @@ def test_sync_s3(self, session_mock, getsize_mock, uploader_mock, exists_mock, r
             FunctionName="PhysicalFunction1", S3Bucket="bucket_name", S3Key="bucket/key"
         )
         remove_mock.assert_called_once_with("zip_file")
+
+    @patch("samcli.lib.sync.flows.zip_function_sync_flow.ResourceAPICall")
+    def test_get_resource_api_calls(self, resource_api_call_mock):
+        build_context = MagicMock()
+        layer1 = MagicMock()
+        layer2 = MagicMock()
+        layer1.full_path = "Layer1"
+        layer2.full_path = "Layer2"
+        function_mock = MagicMock()
+        function_mock.layers = [layer1, layer2]
+        build_context.function_provider.functions.get.return_value = function_mock
+        sync_flow = ZipFunctionSyncFlow(
+            "Function1",
+            build_context=build_context,
+            deploy_context=MagicMock(),
+            physical_id_mapping={},
+            stacks=[MagicMock()],
+        )
+
+        result = sync_flow._get_resource_api_calls()
+        self.assertEqual(len(result), 2)
+        resource_api_call_mock.assert_any_call("Layer1", ["Build"])
+        resource_api_call_mock.assert_any_call("Layer2", ["Build"])