Avoid always loading nested stack on update

Previously, when calling StackResource._validate_nested_resources() (which
we do whenever we create or update a nested stack), we would load the
nested stack into memory to validate the number of resources in the nested
stack, unless the max_resources_per_stack config option was set to -1. This
meant we would load the nested stack into memory in the same engine as the
parent on every update.

To reduce the memory high-water mark, fetch the information we need over
RPC from another engine instead.

To ensure this is only called once, move the call into the validate code.
(Previously it was called again in the create/update itself.)

Change-Id: I78d12ecc8240c697e26893ae2d7172b60883fb93
Partial-Bug: #1731349

heat/engine/resources/stack_resource.py

@@ -222,6 +222,7 @@ class StackResource(resource.Resource):
 
         parsed_template = self._child_parsed_template(child_template,
                                                       child_env)
+        self._validate_nested_resources(parsed_template)
 
         # Note we disable rollback for nested stacks, since they
         # should be rolled back by the parent stack on failure
@@ -248,7 +249,6 @@ class StackResource(resource.Resource):
 
     def _child_parsed_template(self, child_template, child_env):
         parsed_template = self._parse_child_template(child_template, child_env)
-        self._validate_nested_resources(parsed_template)
 
         # Don't overwrite the attributes_schema for subclasses that
         # define their own attributes_schema.
@@ -260,12 +260,16 @@ class StackResource(resource.Resource):
     def _validate_nested_resources(self, templ):
         if cfg.CONF.max_resources_per_stack == -1:
             return
+
         total_resources = (len(templ[templ.RESOURCES]) +
                            self.stack.total_resources(self.root_stack_id))
 
-        if self.nested():
-            # It's an update and these resources will be deleted
-            total_resources -= len(self.nested().resources)
+        identity = self.nested_identifier()
+        if identity is not None:
+            existing = self.rpc_client().list_stack_resources(self.context,
+                                                              identity)
+            # Don't double-count existing resources during an update
+            total_resources -= len(existing)
 
         if (total_resources > cfg.CONF.max_resources_per_stack):
             message = exception.StackResourceLimitExceeded.msg_fmt

heat/tests/test_provider_template.py

@@ -311,7 +311,6 @@ class ProviderTemplateTest(common.HeatTestCase):
         temp_res = template_resource.TemplateResource('test_t_res',
                                                       definition, stack)
         temp_res.resource_id = 'dummy_id'
-        self.assertIsNone(temp_res.validate())
         temp_res.nested_identifier = mock.Mock()
         temp_res.nested_identifier.return_value = {'foo': 'bar'}
 
@@ -319,6 +318,8 @@ class ProviderTemplateTest(common.HeatTestCase):
         output = {'outputs': [{'output_key': 'Foo', 'output_value': None,
                                'output_error': 'it is all bad'}]}
         temp_res._rpc_client.show_stack.return_value = [output]
+        temp_res._rpc_client.list_stack_resources.return_value = []
+        self.assertIsNone(temp_res.validate())
         self.assertRaises(exception.TemplateOutputError,
                           temp_res.FnGetAtt, 'Foo')